SpringCloud Stream消息驱动

简介

什么是SpringCloudStream

官方定义 SpringCloud Stream 是一个构建消息驱动微服务对框架。
应用程序通过 inputs 或者 outputs 来与Springcloud Stream 中 binder 对象交互通过我们配置来 binding（绑定），而 SpringCloud Stream 的 binder 对象负责与消息中间件交互，所以我们只需要搞清楚如何与 Spring Coud Stream 交互就可以方便使用消息驱动的方式。
通过使用SpringIntegration 来连接消息代理中间件实现消息事件驱动。
Spring Cloud Stream 为一些供应商的消息中间件产品提供来个性化的自动化配置实现，引用来发布-订阅、消费组、分区的三个核心概念。
==目前值支持 RabbitMQ、Kafka==

SpringCloud Stream消息驱动
官网 https://spring.io/projects/spring-cloud-stream#overview

Spring Cloud Stream 是用于构建与消息传递系统的高度可伸缩的事件驱动微服务架构，该框架提供来一个灵活的编程模型，它简历在已经建立和熟悉的Spring 熟语和最佳实践上，包括支持持久化的发布订阅、消费组以及消息分区这三个核心概念

API：

https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/3.0.1.RELEASE/reference/html/

中文指导手册：

https://m.wang1314.com/doc/webapp/topic/20971999.html

设计思想

标准MQ

为什么用Cloud Stream

比方说我们用到了RabbitMQ 和 Kafka ，由于这两个消息中间件的架构上的不同，像RabbitMQ 有 exchange，Kafka有 Topic 和 Partitions 分区，
SpringCloud Stream消息驱动
这些中间件的差异性导致我们实际项目开发给我们造成一定困扰，我们如果用了两个消息队列的其中一种，后面业务需求，我们想往另一个消息队列进行迁移，这时候无疑就是一个灾难性的，一大堆东西都需要推到重新做，因为它跟我们系统耦合度很高，这时候SpringCloud Stream 给我们提供了一种解耦合的方式

stream凭什么可以统一底层差异

Biinder

INPUT 对应于消费者
OUTPUT 对应于生产者
在没有绑定器这个概念的情况下，我们springBoot 应用要直接与消息中间件进行交互的时候，由于消息中间件构造不同，他们的实现细节上会有较大的差异性，通过定义绑定器作为中间层，完美地实现了应用与消息中间件细节之间的隔离。通过向应用程序暴露统一的Channel 通道，使得应用程序不需要再考虑各种不同的消息中间件实现
通过绑定器Binder作为中间层，实现了应用程序与消息中间件的隔离

通过定义绑定器Binder 作为中间层，实现了应用程序与消息中间件之间细节的隔离。
Stream 中的消息通信方式遵循了发布订阅 Topic主题进行广播

Spring Cloud Stream标准流程套路

编码API和常用注解

案例：

消息驱动生产者

新建模块 cloud-stream-rabbitmq-provider8801
pom

<dependencies>
    <!--stream rabbit -->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    </dependency>
    <!--eureka client-->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!--监控-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>
    <!--热部署-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-devtools</artifactId>
        <scope>runtime</scope>
        <optional>true</optional>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

server:
  port: 8801

spring:
  application:
    name: cloud-stream-provider
  cloud:
    stream:
      binders: #在此处配置要绑定的rabbitmq的服务信息
        defaultRabbit: #表示定义的名称，用于binding整合
          type: rabbit #消息组件类型
          environment: #设置rabbitmq的相关环境配置
            spring:
              rabbitmq:
                host: ip端口号  #RabbitMQ在本机的用localhost，在服务器的用服务器的ip地址
                port: 5672
                username: guest
                password: guest
      bindings: #服务的整合处理
        output: #这个名字是一个通道的名称
          destination: studyExchange #表示要使用的Exchange名称定义
          content-type: application/json #设置消息类型，本次为json
          binder: defaultRabbit #设置要绑定的消息服务的具体设置（爆红不影响使用，位置没错）

eureka:
  client:
    service-url:
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 #设置心跳的时间间隔（默认是30S)
    lease-expiration-duration-in-seconds: 5 #如果超过5S间隔就注销节点 默认是90s
    instance-id: send-8801.com #在信息列表时显示主机名称
    prefer-ip-address: true #访问的路径变为IP地址

启动类

@SpringBootApplication
public class StreamMQMain8801 {

    public static void main(String[] args) {
        SpringApplication.run(StreamMQMain8801.class, args);
    }

}

新建业务类

新建service包 --- IMessageProvider接口与实现类

public interface IMessageProvider {
    public String send();
}

@EnableBinding(Source.class)    //定义消息的推送管道（Source是spring的）
public class IMessageProviderImpl implements IMessageProvider {

    @Resource
    private MessageChannel output;  //消息发送管道

    @Override
    public String send() {
        String serial = UUID.randomUUID().toString();
        output.send(MessageBuilder.withPayload(serial).build());     //MessageBuilder是spring的integration.support.MessageBuilder
        System.out.println("*******serial: " + serial);
        return null;
    }
}

@RestController
public class SendMessageController {

    @Resource
    private IMessageProvider iMessageProvider;

    @GetMapping("/sendMessage")
    public String sendMessage(){
        return iMessageProvider.send();
    }

}

消费者

新建模块cloud-stream-rabbitmq-consumer8802

<dependencies>
    <!--stream rabbit -->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    </dependency>
    <!--eureka client-->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!--监控-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>
    <!--热部署-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-devtools</artifactId>
        <scope>runtime</scope>
        <optional>true</optional>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

server:
  port: 8802

spring:
  application:
    name: cloud-stream-provider
  cloud:
    stream:
      binders: #在此处配置要绑定的rabbitmq的服务信息
        defaultRabbit: #表示定义的名称，用于binding整合
          type: rabbit #消息组件类型
          environment: #设置rabbitmq的相关环境配置
            spring:
              rabbitmq:
                host: ip端口  #RabbitMQ在本机的用localhost，在服务器的用服务器的ip地址
                port: 5672
                username: 用户名
                password: 密码
      bindings: #服务的整合处理
        input: #这个名字是一个通道的名称
          destination: studyExchange #表示要使用的Exchange名称定义
          content-type: application/json #设置消息类型，本次为json，本文要设置为“text/plain”
          binder: defaultRabbit #设置要绑定的消息服务的具体设置（爆红不影响使用，位置没错）

eureka:
  client:
    service-url:
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 #设置心跳的时间间隔（默认是30S)
    lease-expiration-duration-in-seconds: 5 #如果超过5S间隔就注销节点 默认是90s
    instance-id: receive-8802.com #在信息列表时显示主机名称
    prefer-ip-address: true #访问的路径变为IP地址

主启动类

@SpringBootApplication
public class StreamMQMain8802 {

    public static void main(String[] args) {
        SpringApplication.run(StreamMQMain8802.class, args);
    }

}

新建 Controller

@EnableBinding(Sink.class)
@Controller
public class ReceiveMessageListenerController {

    @Value("${server.port}")
    private String serverPort;

    @StreamListener(Sink.INPUT) //监听
    public void input(Message<String> message){
        System.out.println("消费者1号------>收到的消息：" + message.getPayload() + "\t port：" + serverPort);
    }

}

启动Eureka 消息驱动生产者，消费者

http://localhost:8801/sendMessage （8801发送消息）

SpringCloud Stream消息驱动
8802接收到消息：

SpringCloud Stream消息驱动

个人博客： http://blog.yanxiaolong.cn/ .

SpringCloud Stream消息驱动

SpringCloud Stream消息驱动

简介

设计思想

编码API和常用注解

案例：

继续阅读

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的简单使用

neo4j之cypher使用文档

GitHub连夜封杀！这份阿里 10W 字内部 Java 字面试手册到底有多强？

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结

mybatis_入门程序Mybatis入门

vue-cli简介（中文翻译）

AOP编程_Android优雅权限框架(1)概念基础，2021金三银四前言正文大纲正文

Ajax发送和获取json数据到Spring mvc 1.spring mvc后端2.web前段

Effective Java 8:通用程序设计

OOM三种类型

工厂模式-三种类型

【递归】高效率求2的n次幂

win10本地scala和spark安装安装scala安装spark

scala (3) Function 和 Method

JSONObject包导入异常 java.lang.NoClassDefFoundErrorweb项目的导入包的问题