SpringCloud Stream 整合RabbitMQ的基本步驟

2022-03-30 13:01:33

一、專案介紹

演示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）專案整體結構

（3）基礎模組

1）pom.xml

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

<!-- Spring Cloud Stream, 用於MQ訊息傳送-->
<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）結構

（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()); 生產訊息

三、消費者