# 从浅入深学习 SpringCloud 微服务架构（十六）

从浅入深学习 SpringCloud 微服务架构（十六）

Stream_2">一、SpringCloudStream：自定义消息通道

1、在子工程 stream_product （子模块）中,创建自定义的消息通道类 MyProcessor.java

/***   spring_cloud_demo\stream_product\src\main\java\djh\it\stream\channel\MyProcessor.java**   2024-5-11 创建 自定义的消息通道类 MyProcessor.java*/package djh.it.stream.channel;import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;public interface MyProcessor {//消息生产者的配置String MYOUTPUT = "myoutput";@Output("myoutput")MessageChannel myoutput();//消息消费者的配置String MYINPUT = "myinput";@Input("myinput")SubscribableChannel myinput();
}

2、在子工程 stream_product （子模块）中，修改消息发送的工具类 MessageSender.java 使用自定义消息通道。

/***  spring_cloud_demo\stream_product\src\main\java\djh\it\stream\producer\MessageSender.java**  2024-5-10  抽取一个消息发送的工具类 MessageSender.java*/package djh.it.stream.producer;import djh.it.stream.channel.MyProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.cloud.stream.annotation.EnableBinding;
//import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;@Component
//@EnableBinding(Source.class)
@EnableBinding(MyProcessor.class)
public class MessageSender {
//    @Autowired
//    private MessageChannel output;
//
//    //发送消息
//    public void send(Object obj){
//        output.send(MessageBuilder.withPayload((obj)).build());
//    }@Autowired@Qualifier(value = "myoutput")private MessageChannel myoutput;//发送消息public void send(Object obj){myoutput.send(MessageBuilder.withPayload((obj)).build());}
}

3、在子工程 stream_product （子模块）中，修改 application.yml 配置文件，添加自定义消息配置。

##  spring_cloud_demo\stream_product\src\main\resources\application.ymlserver:port: 7001  #服务端口
spring:application:nmae: stream_product  #指定服务名rabbitmq:addresses: 127.0.0.1username: guestpassword: guestcloud:stream:bindings:output:  #管道交互destination: djh-default  #指定消息发送的目的地，在 rabbitmq 中，发送到一个 djh-default 的交换机 exchange。myoutput:   # 自定义消息通道destination: djh-custom-outputbinders:  #配置绑定器defaultRabbit:type: rabbit

4、在子工程 stream_consumer （子模块）中,创建自定义的消息通道类 MyProcessor.java

/***   spring_cloud_demo\stream_consumer\src\main\java\djh\it\stream\channel\MyProcessor.java**   2024-5-11 创建 自定义的消息通道类 MyProcessor.java*/package djh.it.stream.channel;import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;public interface MyProcessor {//消息生产者的配置String MYOUTPUT = "myoutput";@Output("myoutput")MessageChannel myoutput();//消息消费者的配置String MYINPUT = "myinput";@Input("myinput")SubscribableChannel myinput();
}

5、在子工程 stream_consumer （子模块）中,修改获取消息工具类 MessageListener.java 使用自定义消息通道。

/***   spring_cloud_demo\stream_consumer\src\main\java\djh\it\stream\consumer\MessageListener.java**   2024-5-10 创建一个获取消息工具类 MessageListener.java*/package djh.it.stream.consumer;import djh.it.stream.channel.MyProcessor;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
//import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.stereotype.Component;@Component
//@EnableBinding(Sink.class)
@EnableBinding(MyProcessor.class)
public class MessageListener {//    //监听 binding 中的消息
//    @StreamListener(Sink.INPUT)
//    public void input(String message) {
//        System.out.println("获取到的消息： " + message);
//    }//监听 binding 中的消息@StreamListener(MyProcessor.MYINPUT)public void input(String message) {System.out.println("获取到的消息： " + message);}
}

6、在子工程 stream_consumer （子模块）中,修改 application.yml 配置文件，添加自定义消息配置。

##  spring_cloud_demo\stream_consumer\src\main\resources\application.ymlserver:port: 7002  #服务端口
spring:application:nmae: stream_consumer  #指定服务名rabbitmq:addresses: 127.0.0.1username: guestpassword: guestcloud:stream:bindings:input:  #管道交互,内置的获取消息的通道，从 djh-default 中获取消息。destination: djh-default  #指定消息发送的目的地，在 rabbitmq 中，发送到一个 djh-default 的交换机 exchange。myinput:   #自定义消息通道destination: djh-custom-outputbinders:  #配置绑定器defaultRabbit:type: rabbit

7、在子工程 stream_product （子模块）中，运行启动类 ProducerApplication.java 进行测试

/***   spring_cloud_demo\stream_product\src\main\java\djh\it\stream\ProducerApplication.java**   2024-5-9 SpringCloudStream 入门案例：启动类 ProducerApplication.java*      1）引入依赖。*      2）配置 application.yml 配置文件。*      3）发送消息的话，定义一个通道接口，通过接口中内置的 messagechannel,(sprngcloudtream 中内置接口 Source)*      4）@EnableBinding 注解 ：绑定对应通道。*      5）发送消息的话，通过 MessageChannel 发送消息，如果需要 MessageChannel --> 通过绑定内置接口获取。*/package djh.it.stream;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class ProducerApplication {public static void main(String[] args) {SpringApplication.run(ProducerApplication.class);}
}

8、在子工程 stream_consumer （子模块）中,运行启动类 ConsumerApplication.java 进行测试。

/***    spring_cloud_demo\stream_consumer\src\main\java\djh\it\stream\ConsumerApplication.java**   2024-5-9 SpringCloudStream 入门案例：启动类 ConsumerApplication.java*      1）引入依赖。*      2）配置 application.yml 配置文件。*      3）定义一个通道接口，通过内置获取消息的接口：Sink*      4）绑定对应通道。*      5）配置一个监听方法 ：当程序从中间件获取数据之后，执行的业务逻辑方法，需要在监听方法上配置 @StreamListener 注解。*/package djh.it.stream;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class ConsumerApplication {public static void main(String[] args) {SpringApplication.run(ConsumerApplication.class);}
}

9、在子工程 stream_product （子模块）中，运行一个测试类 ProducterTest.java 进行测试。

/***  spring_cloud_demo\stream_product\src\test\java\djh\it\stream\ProducterTest.java**  2024-5-10 创建一个测试类 ProducterTest.java*/package djh.it.stream;import djh.it.stream.producer.MessageSender;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;@SpringBootTest
@RunWith(SpringJUnit4ClassRunner.class)
public class ProducterTest {@Autowiredprivate MessageSender messageSender;@Testpublic void testSend(){messageSender.send("hello 测试 工具类");}
}

10、启动 rabbitmqctl-server.bat 服务，并运行测试类 ProducterTest 和 ConsumerApplication 启动类，在 idea Run Dashboard 控制面板，

同样会输出 “获取到的消息： hello 测试工具类”

在这里插入图片描述

Stream_301">二、SpringCloudStream：消息分组

Stream_303">1、SpringCloudStream：消息分组

通常在生产环境，我们的每个服务都不会以单节点的方式运行在生产环境，当同一个服务启动多个实例的时候，这些实例都会绑定到同一个消息通道的目标主题(Topic)上。默认情况下，当生产者发出一条消息到绑定通道上，这条消息会产生多个副本被每个消费者实例接收和处理，但是有些业务场景之下，我们希望生产者产生的消息只被其中一个实例消费，这个时候我们需要为这些消费者设置消费组来实现这样的功能。
实现的方式非常简单，我们只需要在服务消费者端设置 spring.c1oud.stream.bindings.input.group 属性即可。

2、在子工程 stream_consumer （子模块），复制一个更名为：在子工程 stream_consumer_2 （子模块），并把 application.yml 配置文件中的端口号改为：7003

1）子工程 stream_consumer_2 （子模块）中的 pom.xml 文件。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><parent><artifactId>spring_cloud_demo</artifactId><groupId>djh.it</groupId><version>1.0-SNAPSHOT</version></parent><modelVersion>4.0.0</modelVersion><artifactId>stream_consumer_2</artifactId><dependencies><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-stream</artifactId></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-stream-binder-rabbit</artifactId></dependency></dependencies>
</project>
<!-- spring_cloud_demo\stream_consumer_2\pom.xml -->

2）子工程 stream_consumer_2 （子模块）中的 application.yml 文件。

##  spring_cloud_demo\stream_consumer_2\src\main\resources\application.ymlserver:port: 7003  #服务端口
spring:application:nmae: stream_consumer_2  #指定服务名rabbitmq:addresses: 127.0.0.1username: guestpassword: guestcloud:stream:bindings:input:  #管道交互,内置的获取消息的通道，从 djh-default 中获取消息。destination: djh-default  #指定消息发送的目的地，在 rabbitmq 中，发送到一个 djh-default 的交换机 exchange。myinput:   #自定义消息通道destination: djh-custom-outputbinders:  #配置绑定器defaultRabbit:type: rabbit

3）子工程 stream_consumer_2 （子模块）中的自定义的消息通道类 MyProcessor.java

/***   spring_cloud_demo\stream_consumer_2\src\main\java\djh\it\stream\channel\MyProcessor.java**   2024-5-11 创建 自定义的消息通道类 MyProcessor.java*/package djh.it.stream.channel;import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;public interface MyProcessor {//消息生产者的配置String MYOUTPUT = "myoutput";@Output("myoutput")MessageChannel myoutput();//消息消费者的配置String MYINPUT = "myinput";@Input("myinput")SubscribableChannel myinput();
}

4）子工程 stream_consumer_2 （子模块）中的获取消息工具类 MessageListener.java

/***   spring_cloud_demo\stream_consumer_2\src\main\java\djh\it\stream\consumer\MessageListener.java**   2024-5-11 创建一个获取消息工具类 MessageListener.java*/package djh.it.stream.consumer;import djh.it.stream.channel.MyProcessor;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
//import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.stereotype.Component;@Component
//@EnableBinding(Sink.class)
@EnableBinding(MyProcessor.class)
public class MessageListener {//    //监听 binding 中的消息
//    @StreamListener(Sink.INPUT)
//    public void input(String message) {
//        System.out.println("获取到的消息： " + message);
//    }//监听 binding 中的消息@StreamListener(MyProcessor.MYINPUT)public void input(String message) {System.out.println("获取到的消息： " + message);}
}

5）子工程 stream_consumer_2 （子模块）中的启动类 ConsumerApplication_2.java

/***   spring_cloud_demo\stream_consumer_2\src\main\java\djh\it\stream\ConsumerApplication_2.java**   2024-5-11 SpringCloudStream 入门案例：启动类 ConsumerApplication_2.java*      1）引入依赖。*      2）配置 application.yml 配置文件。*      3）定义一个通道接口，通过内置获取消息的接口：Sink*      4）绑定对应通道。*      5）配置一个监听方法 ：当程序从中间件获取数据之后，执行的业务逻辑方法，需要在监听方法上配置 @StreamListener 注解。*/package djh.it.stream;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class ConsumerApplication_2 {public static void main(String[] args) {SpringApplication.run(ConsumerApplication_2.class);}
}

3、启动 rabbitmqctl-server.bat 服务，并运行测试类 ProducterTest 和 ConsumerApplication 启动类和 ConsumerApplication_2 启动类，

在 idea Run Dashboard 控制面板，两个消费都启动类都会输出 “获取到的消息： hello 测试工具类”

在这里插入图片描述

4、在子工程 stream_consumer （子模块）的 application.yml 配置文件中，添加消息分组配置。

##  spring_cloud_demo\stream_consumer\src\main\resources\application.ymlserver:port: 7002  #服务端口
spring:application:nmae: stream_consumer  #指定服务名rabbitmq:addresses: 127.0.0.1username: guestpassword: guestcloud:stream:bindings:input:  #管道交互,内置的获取消息的通道，从 djh-default 中获取消息。destination: djh-default  #指定消息发送的目的地，在 rabbitmq 中，发送到一个 djh-default 的交换机 exchange。myinput:   #自定义消息通道destination: djh-custom-outputgroup: group1  #消息分组（同一组只能有一个消息者获取消息）binders:  #配置绑定器defaultRabbit:type: rabbit

5、在子工程 stream_consumer_2 （子模块）的 application.yml 配置文件中，也添加消息分组配置。

##  spring_cloud_demo\stream_consumer_2\src\main\resources\application.ymlserver:port: 7003  #服务端口
spring:application:nmae: stream_consumer_2  #指定服务名rabbitmq:addresses: 127.0.0.1username: guestpassword: guestcloud:stream:bindings:input:  #管道交互,内置的获取消息的通道，从 djh-default 中获取消息。destination: djh-default  #指定消息发送的目的地，在 rabbitmq 中，发送到一个 djh-default 的交换机 exchange。myinput:   #自定义消息通道destination: djh-custom-outputgroup: group1  #消息分组（同一组只能有一个消息者获取消息）binders:  #配置绑定器defaultRabbit:type: rabbit

6、重新启动 rabbitmqctl-server.bat 服务，并运行测试类 ProducterTest 和 ConsumerApplication 启动类和 ConsumerApplication_2 启动类，

在 idea Run Dashboard 控制面板，发现只有一个消费都启动类都会输出 “获取到的消息： hello 测试工具类”

在这里插入图片描述

Stream_538">三、SpringCloudStream：消息分区

1、消息分区

有一些场景需要满足,同一个特征的数据被同一个实例消费,比如同一个id的传感器监测数据必须被同-个实例统计计算分析,否则可能无法获取全部的数据。又比如部分异步任务，首次请求启动task，二次请求取消task，此场景就必须保证两次请求至同一实例.

2、在子工程 stream_producer （子模块）的 application.yml 配置文件中，添加消息分区配置。

##  spring_cloud_demo\stream_product\src\main\resources\application.ymlserver:port: 7001  #服务端口
spring:application:nmae: stream_product  #指定服务名rabbitmq:addresses: 127.0.0.1username: guestpassword: guestcloud:stream:bindings:output:  #管道交互destination: djh-default  #指定消息发送的目的地，在 rabbitmq 中，发送到一个 djh-default 的交换机 exchange。myoutput:   # 自定义消息通道destination: djh-custom-outputproducer:  # 配置分区partition-key-expression: payload  # 分区关键字，对象中的 id 或 对象。partition-count: 2  # 分区大小binders:  #配置绑定器defaultRabbit:type: rabbit

3、在子工程 stream_consumer （子模块）的 application.yml 配置文件中，也添加消息分区配置。

##  spring_cloud_demo\stream_consumer\src\main\resources\application.ymlserver:port: 7002  #服务端口
spring:application:nmae: stream_consumer  #指定服务名rabbitmq:addresses: 127.0.0.1username: guestpassword: guestcloud:stream:instanceCount: 2  # 消费者总数。instanceIndex: 0  # 当前消费者的索引，从 0 开始。bindings:input:  #管道交互,内置的获取消息的通道，从 djh-default 中获取消息。destination: djh-default  #指定消息发送的目的地，在 rabbitmq 中，发送到一个 djh-default 的交换机 exchange。myinput:   #自定义消息通道destination: djh-custom-outputgroup: group1  #消息分组（同一组只能有一个消息者获取消息）consumer:partitioned: true  # 开启分区支持binders:  #配置绑定器defaultRabbit:type: rabbit

3、在子工程 stream_consumer_2 （子模块）的 application.yml 配置文件中，也添加消息分区配置。

##  spring_cloud_demo\stream_consumer_2\src\main\resources\application.ymlserver:port: 7003  #服务端口
spring:application:nmae: stream_consumer_2  #指定服务名rabbitmq:addresses: 127.0.0.1username: guestpassword: guestcloud:stream:instanceCount: 2  # 消费者总数。instanceIndex: 1  # 当前消费者的索引，从 0 开始。bindings:input:  #管道交互,内置的获取消息的通道，从 djh-default 中获取消息。destination: djh-default  #指定消息发送的目的地，在 rabbitmq 中，发送到一个 djh-default 的交换机 exchange。myinput:   #自定义消息通道destination: djh-custom-outputgroup: group2  #消息分组（同一组只能有一个消息者获取消息）consumer:partitioned: true  # 开启分区支持binders:  #配置绑定器defaultRabbit:type: rabbit

4、修改子工程 stream_producer （子模块）的测试类 ProducterTest 进行测试。

/***  spring_cloud_demo\stream_product\src\test\java\djh\it\stream\ProducterTest.java**  2024-5-10 创建一个测试类 ProducterTest.java*/package djh.it.stream;import djh.it.stream.producer.MessageSender;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;@SpringBootTest
@RunWith(SpringJUnit4ClassRunner.class)
public class ProducterTest {@Autowiredprivate MessageSender messageSender;@Testpublic void testSend(){
//        messageSender.send("hello 测试 工具类");for(int i=0;i<5;i++){messageSender.send("0");}}
}

5、重新启动 rabbitmqctl-server.bat 服务，并运行测试类 ProducterTest 和 ConsumerApplication 启动类和 ConsumerApplication_2 启动类，

在 idea Run Dashboard 控制面板，发现只有 ConsumerApplication 一个消费者启动类都会输出 “获取到的消息： 0”

在这里插入图片描述

上一节关联链接请点击：
# 从浅入深学习 SpringCloud 微服务架构（十五）