快捷導(dǎo)航

Springcloud Stream消息驅(qū)動工具使用介紹

更新時間：2022年09月14日 11:11:37 作者：扎哇太棗糕

SpringCloud Stream由一個中間件中立的核組成，應(yīng)用通過SpringCloud Stream插入的input(相當(dāng)于消費者consumer，它是從隊列中接收消息的)和output(相當(dāng)于生產(chǎn)者producer，它是發(fā)送消息到隊列中的)通道與外界交流

springcloud Stream

什么是springcloud Stream

現(xiàn)在市面上有很多的消息中間件，每一個公司使用的都有所不同，為了減少學(xué)習(xí)的成本，springcloud Stream可以讓我們不再關(guān)注消息中間件MQ的具體細(xì)節(jié)，我們只需要通過適配綁定的方式即可實現(xiàn)不同MQ之間的切換，但是遺憾的是springcloud Stream目前只支持RabbitMQ和Kafka。

SpringCloud Stream是一個構(gòu)建消息驅(qū)動微服務(wù)的框架，應(yīng)用程序通過inputs或者 outputs來與SpringCloud Stream中的binder進(jìn)行交互，我們可以通過配置來binding ，而 SpringCloud Stream 的binder負(fù)責(zé)與中間件交互，所以我們只需要搞清楚如何與Stream交互就可以很方便的使用消息驅(qū)動了！

什么是Binder

Binder是SpringCloud Stream的一個抽象概念，是應(yīng)用與消息中間件之間的粘合劑，通過定義綁定器Binder作為中間層，實現(xiàn)了應(yīng)用程序與消息中間件細(xì)節(jié)之間的隔離，可以動態(tài)的改變消息的destinations（對應(yīng)于 Kafka的topic，RabbitMQ的exchanges），這些都可以通過外部配置項來做到，甚至可以任意的改變中間件的類型但是不需要修改一行代碼

為什么使用Stream

比方說我們用到了RabbitMQ和Kafka，由于這兩個消息中間件的架構(gòu)上的不同像RabbitMQ有exchange，kafka有Topic和Partitions分區(qū)，這些中間件的差異性導(dǎo)致我們實際項目開發(fā)給我們造成了一定的困擾，我們?nèi)绻昧藘蓚€消息隊列的其中一種，后面的業(yè)務(wù)需求，我想往另外一種消息隊列進(jìn)行遷移;這時候無疑就是一個災(zāi)難性的，一大堆東西都要重新推倒重新做，因為它跟我們的系統(tǒng)耦合了，這襯候springcloud Stream給我們提供了一種解耦合的方式。

Stream使用案例

前置知識

Stream處理消息的架構(gòu)

Source、Sink：簡單的可理解為參照對象是Spring Cloud Stream自身，從Stream發(fā)布消息就是輸出，接受消息就是輸入。Channel：通道，是隊列Queue的一種抽象，在消息通訊系統(tǒng)中就是實現(xiàn)存儲和轉(zhuǎn)發(fā)的媒介。Binder：消息的生產(chǎn)者和消費者中間層，實現(xiàn)了應(yīng)用程序與消息中間件細(xì)節(jié)之間的隔離

通過以上兩張圖片可知，消息的處理流向是：消息生產(chǎn)者處理完業(yè)務(wù)邏輯之后消息到達(dá)source中，接著前往Channel通道進(jìn)行排隊，然后通過binder綁定器將消息數(shù)據(jù)發(fā)送到底層mq，然后又通過binder綁定器接收到底層mq發(fā)送來的消息數(shù)據(jù)，接著前往Channel通道進(jìn)行排隊，由Sink接收到消息數(shù)據(jù)，消息消費者拿到消息數(shù)據(jù)執(zhí)行相應(yīng)的業(yè)務(wù)邏輯

Stream常用注解

消息生產(chǎn)者8801模塊搭建

第一步：創(chuàng)建一個maven模塊，引入相關(guān)依賴，最主要的就是stream整合rabbitmq的依賴

<!--stream的rabbitmq依賴-->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

第二步：配置文件的編寫

server:
port: 8801

spring:
application:
name: cloud-stream-provider
cloud:
stream:
binders: # 在此處配置要綁定的rabbitmq的服務(wù)信息；
defaultRabbit: # 表示定義的名稱，用于于binding整合
type: rabbit # 消息組件類型
environment: # 設(shè)置rabbitmq的相關(guān)的環(huán)境配置
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
bindings: # 服務(wù)的整合處理
output: # 這個名字是一個通道的名稱
destination: studyExchange # 表示要使用的Exchange名稱定義
content-type: application/json # 設(shè)置消息類型，本次為json，文本則設(shè)置“text/plain”
binder: defaultRabbit # 設(shè)置要綁定的消息服務(wù)的具體設(shè)置

eureka:
client: # 客戶端進(jìn)行Eureka注冊的配置
service-url:
defaultZone: http://localhost:7001/eureka

第三步：主程序類

@SpringBootApplication
public class CloudStreamRabbitmqProvider8801Application {
    public static void main(String[] args) {
        SpringApplication.run(CloudStreamRabbitmqProvider8801Application.class, args);
        System.out.println("啟動成功");
    }
}

第四步：業(yè)務(wù)層service代碼編寫，注意：這里實現(xiàn)類注入的對象由之前的dao層對象換成了channel通道對象，詳細(xì)的發(fā)送由實現(xiàn)類的第12完成

public interface IMessageProviderService {
    /**
     * 定義消息的推送管道
     *
     * @return
     */
    String send();
}

@EnableBinding(Source.class)
public class MessageProviderServiceImpl implements IMessageProviderService {
    /**
     * 消息發(fā)送管道/信道
     */
    @Resource
    private MessageChannel output;
    @Override
    public String send() {
        String serial = UUID.randomUUID().toString();
        output.send(MessageBuilder.withPayload(serial).build());

        System.out.println("*****serial: " + serial);
        return serial;
    }
}

第五步： controller接口

@RestController
public class SendMessageController {
    @Resource
    private IMessageProviderService messageProviderService;
    @GetMapping(value = "/sendMessage")
    public String sendMessage() {
        return messageProviderService.send();
    }
}

消息消費者8802模塊搭建

第一步：創(chuàng)建一個maven模塊，引入相關(guān)依賴，最主要的就是stream整合rabbitmq的依賴

<!--stream的rabbitmq依賴-->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

第二步：配置文件的編寫，與生產(chǎn)者的區(qū)別就在于bindings下的是input而不是output

server:
port: 8802

spring:
application:
name: cloud-stream-consumer
cloud:
stream:
binders: # 在此處配置要綁定的rabbitmq的服務(wù)信息；
defaultRabbit: # 表示定義的名稱，用于于binding整合
type: rabbit # 消息組件類型
environment: # 設(shè)置rabbitmq的相關(guān)的環(huán)境配置
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
bindings: # 服務(wù)的整合處理
input: # 這個名字是一個通道的名稱
destination: studyExchange # 表示要使用的Exchange名稱定義
content-type: application/json # 設(shè)置消息類型，本次為json，文本則設(shè)置“text/plain”
binder: defaultRabbit # 設(shè)置要綁定的消息服務(wù)的具體設(shè)置
eureka:
client: # 客戶端進(jìn)行Eureka注冊的配置
service-url:
defaultZone: http://localhost:7001/eureka

第三步：主程序類

@SpringBootApplication
public class CloudStreamRabbitmqConsumer8802Application {
    public static void main(String[] args) {
        SpringApplication.run(CloudStreamRabbitmqConsumer8802Application.class, args);
        System.out.println("啟動成功");
    }
}

第四步： controller接口，使用url請求生產(chǎn)者8801，即可在消費者8802端接收到8801發(fā)送的消息

@Component
@EnableBinding(Sink.class)
public class ReceiveMessageListener {
    @Value("${server.port}")
    private String serverPort;
    @StreamListener(Sink.INPUT)
    public void input(Message<String> message) {
        System.out.println("消費者1號 ----> port:" + serverPort + "\t從8801接受到的消息是：" + message.getPayload());
    }
}

兩個模塊搭建完成進(jìn)行測試，首先啟動注冊中心7001，然后分別啟動消息生產(chǎn)者8801和消息消費者8802，通過url請求訪問8001的發(fā)送消息請求，會向指定管道中發(fā)送一條消息，如果此時這個管道中有消費者即可接收到這條消息。而如何指定消息的管道歸屬呢，就是通過配置文件中的indings.input.destination來指定，命名相同的服務(wù)就會處在同一條管道中