SpringCloud Stream 整合RabbitMQ

红太狼 2023-06-27 06:25 107阅读 0赞

**本篇简单介绍SpringCloud Stream 整合RabbitMQ基本步骤：**

*  引入SpringCloud
 *  引入SpringCloud Stream相关依赖
 *  定义绑定接口: 消息生产者(Output…Binding) 、消息消费者(Input…Binding)
 *  @EnableBinding 在对应类上进行定义
 *  @StreamListener 在对应方法上创建监听用来消费消息
 *  调用output的send()方法生产消息

# 一、项目介绍 #

演示SpringCloud Stream 整合RabbitMQ,项目可以在一个工程里完成，本次建立了一个工程mq-service,其中包含三个Module:

*  mq-service-base ：基础模块(包含了共用依赖、共用变量)
 *  mq-service-producer ：生产者
 *  mq-service-consumer ：消费者

***注：*** *完全可以在一个工程里实现，这里为了区分，并为了后续单独启动或停止生产者或消费者做实验，也为了适应实际应用项目，所以创建了不同Module*

##### （1）版本 #####

*  SpringBoot : 2.0.6.RELEASE
 *  SpringCloud : Finchley.SR2
 *  RabbitMQ : 3.8.1

#### （2）项目整体结构 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3_size_16_color_FFFFFF_t_70]

#### （3）基础模块 ####

##### 1）pom.xml #####

这里作为公共模块引入SpringCloud、Spring Cloud Stream等，其中也再此引入fastjson、lombok等工具依赖  
***（完整代码见文章最下面）***  
其中Spring Cloud Stream如下：

<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>

##### 2) model #####

定义共用的变量，如CollectionRequest.java

# 二、生产者 #

#### （1）结构 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3_size_16_color_FFFFFF_t_70 1]

#### （2）pom.xml ####

导入base的依赖即可，因为相关共用依赖在base中已经引入

<dependency>
    	<groupId>com.zrk</groupId>
    	<artifactId>mq-service-base</artifactId>
    	<version>0.0.1-SNAPSHOT</version>
    </dependency>

#### （3）定义绑定(接口) ####

OutputMessageBinding.java

public interface OutputMessageBinding { 
    
        /** Topic 名称*/
        String OUTPUT = "message-center-out";
    
        @Output(OUTPUT)
        MessageChannel output();
    
    }

#### （4）添加配置 ####

# rabbitmq连接信息
    spring.rabbitmq.addresses=192.168.1.125
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=admin
    spring.rabbitmq.password=123456
    
    spring.cloud.stream.bindings.message-center-out.destination=message-center
    spring.cloud.stream.rabbit.bindings.message-center-out.consumer.exchangeType=fanout

#### (5) 调用方法 ####

CollectionServiceImpl.java

@Service
    @EnableBinding(OutputMessageBinding.class)
    public class CollectionServiceImpl implements CollectionService{ 
    
        @Resource
        private OutputMessageBinding outputMessageBinding;
    
        /** * @param schoolName * @param content */
        @Override
        public void getCollection(String schoolName, String content) { 
    
            CollectionRequest request = new CollectionRequest();
            request.setSchoolName(schoolName);
            request.setContent(content);
    
            outputMessageBinding.output().send(MessageBuilder.withPayload(request).build());
        }
    }

***注：*** 主要是两点

*  @EnableBinding 定义
 *  outputMessageBinding.output().send(MessageBuilder.withPayload(request).build()); 生产消息

# 三、消费者 #

#### （1）结构 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3_size_16_color_FFFFFF_t_70 2]

#### （2）pom.xml ####

导入base的依赖即可，因为相关共用依赖在base中已经引入

<dependency>
    	<groupId>com.zrk</groupId>
    	<artifactId>mq-service-base</artifactId>
    	<version>0.0.1-SNAPSHOT</version>
    </dependency>

#### （3）定义绑定(接口) ####

InputMessageBinding.java

public interface InputMessageBinding { 
    
        String INPUT = "message-center-input";
    
        @Input(INPUT)
        SubscribableChannel input();
    }

***注：*** 消费者这里与生产者不同，用的是**SubscribableChannel** ，而生产者用的是**MessageChannel**

#### （4）添加配置 ####

# rabbitmq连接信息
    spring.rabbitmq.addresses=192.168.1.125
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=admin
    spring.rabbitmq.password=123456
    
    spring.cloud.stream.bindings.message-center-input.destination=message-center
    spring.cloud.stream.bindings.message-center-input.group=${spring.application.name}

#### (5) 调用方法 ####

CollectionReceiver.java

@Slf4j
    @EnableBinding(InputMessageBinding.class)
    public class CollectionReceiver { 
    
        @StreamListener(InputMessageBinding.INPUT)
        public void handle(String value){ 
            log.info("[消息] 接收到发送消息MQ: {}", value);
            CollectionRequest request = JSON.parseObject(value, CollectionRequest.class);
            log.info("处理收集信息：" + request.toString());
        }
    }

***注：*** 主要是两点

*  @EnableBinding 定义
 *  @StreamListener 注册监听

至此，生产者与消费者都创建完成，分别启动两个项目，并调用生产者接口进行验证：

# 四、验证 #

#### 在postman 访问生产者接口： ####

localhost:30110/collection/getCollection?schoolName=‘zrk’&content=‘send message to rabbitmq’  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3_size_16_color_FFFFFF_t_70 3]

#### 观察消费者日志： ####

![在这里插入图片描述][20200105014340582.png]

#### 查看rabbitmq首页 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3_size_16_color_FFFFFF_t_70 4]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3_size_16_color_FFFFFF_t_70 5]  
则证明已经整合成功，接下来将研究一下更多的配置与用法。

如果有需要，可以参考项目完整代码：[https://github.com/zrk333/mq-service][https_github.com_zrk333_mq-service]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200105010258579.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200105011242294.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200105013324194.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200105014237338.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3,size_16,color_FFFFFF,t_70
[20200105014340582.png]: https://img-blog.csdnimg.cn/20200105014340582.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200105015711142.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200105015754545.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMDY3MzA3,size_16,color_FFFFFF,t_70
[https_github.com_zrk333_mq-service]: https://github.com/zrk333/mq-service