快捷導(dǎo)航

SpringCloud?Stream?整合RabbitMQ的基本步驟

更新時間：2022年03月30日 10:39:13 作者：毛毛的貓毛

這篇文章主要介紹了SpringCloud?Stream?整合RabbitMQ的基本步驟，從項目介紹到生產(chǎn)者結(jié)合示例代碼給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下

引入SpringCloud
引入SpringCloud Stream相關(guān)依賴
定義綁定接口: 消息生產(chǎn)者(Output…Binding) 、消息消費者(Input…Binding)
@EnableBinding 在對應(yīng)類上進(jìn)行定義
@StreamListener 在對應(yīng)方法上創(chuàng)建監(jiān)聽用來消費消息
調(diào)用output的send()方法生產(chǎn)消息

一、項目介紹

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

mq-service-base ：基礎(chǔ)模塊(包含了共用依賴、共用變量)
mq-service-producer ：生產(chǎn)者
mq-service-consumer ：消費者

注：完全可以在一個工程里實現(xiàn)，這里為了區(qū)分，并為了后續(xù)單獨啟動或停止生產(chǎn)者或消費者做實驗，也為了適應(yīng)實際應(yīng)用項目，所以創(chuàng)建了不同Module

（1）版本

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

（2）項目整體結(jié)構(gòu)

在這里插入圖片描述

（3）基礎(chǔ)模塊

1）pom.xml

這里作為公共模塊引入SpringCloud、Spring Cloud Stream等，其中也再此引入fastjson、lombok等工具依賴
（完整代碼見文章最下面）
其中Spring Cloud Stream如下：

<!-- Spring Cloud Stream, 用于MQ消息發(fā)送-->
<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

二、生產(chǎn)者

（1）結(jié)構(gòu)

在這里插入圖片描述

（2）pom.xml

導(dǎo)入base的依賴即可，因為相關(guān)共用依賴在base中已經(jīng)引入

<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) 調(diào)用方法

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()); 生產(chǎn)消息

三、消費者

（1）結(jié)構(gòu)

在這里插入圖片描述

（2）pom.xml

導(dǎo)入base的依賴即可，因為相關(guān)共用依賴在base中已經(jīng)引入

<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();
}

注：消費者這里與生產(chǎn)者不同，用的是SubscribableChannel ，而生產(chǎn)者用的是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) 調(diào)用方法

CollectionReceiver.java

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

注：主要是兩點