java B2B2C源碼電子商務平臺 -Spring Cloud Stream同一通道根據消息內容分發不一樣的消費邏輯

時間 2019-12-04

標籤 java b2b2c 源碼電子商務平臺 spring cloud stream 同一通道根據消息內容分發不一樣消費邏輯欄目 Java 简体版

原文原文鏈接

有的時候，咱們對於同一通道中的消息處理，會經過判斷頭信息或者消息內容來作一些差別化處理，好比：可能在消息頭信息中帶入消息版本號，而後經過if判斷來執行不一樣的處理邏輯，其代碼結構多是這樣的：須要JAVA Spring Cloud大型企業分佈式微服務雲構建的B2B2C電子商務平臺源碼一零三八七七四六二六java

@StreamListener(value = TestTopic.INPUT)
public void receiveV1(String payload, @Header("version") String version) {
    if("1.0".equals(version)) {
        // Version 1.0
    }
    if("2.0".equals(version)) {
        // Version 2.0
    }
}
複製代碼

那麼當消息處理邏輯複雜的時候，這段邏輯就會變得特別複雜。針對這個問題，在@StreamListener註解中提供了一個不錯的屬性condition，能夠用來優化這樣的處理結構。spring

動手試試下面經過編寫一個簡單的例子來具體體會一下這個屬性的用法：bash

@EnableBinding(TestApplication.TestTopic.class)
@SpringBootApplication
public class TestApplication {

    public static void main(String[] args) {
        SpringApplication.run(TestApplication.class, args);
    }

    @RestController
    static class TestController {

        @Autowired
        private TestTopic testTopic;

        /**
         * 消息生產接口
         *
         * @param message
         * @return
         */
        @GetMapping("/sendMessage")
        public String messageWithMQ(@RequestParam String message) {
            testTopic.output().send(MessageBuilder.withPayload(message).setHeader("version", "1.0").build());
            testTopic.output().send(MessageBuilder.withPayload(message).setHeader("version", "2.0").build());
            return "ok";
        }

    }

    /**
     * 消息消費邏輯
     */
    @Slf4j
    @Component
    static class TestListener {

        @StreamListener(value = TestTopic.INPUT, condition = "headers['version']=='1.0'")
        public void receiveV1(String payload, @Header("version") String version) {
            log.info("Received v1 : " + payload + ", " + version);
        }

        @StreamListener(value = TestTopic.INPUT, condition = "headers['version']=='2.0'")
        public void receiveV2(String payload, @Header("version") String version) {
            log.info("Received v2 : " + payload + ", " + version);
        }

    }

    interface TestTopic {

        String OUTPUT = "example-topic-output";
        String INPUT = "example-topic-input";

        @Output(OUTPUT)
        MessageChannel output();

        @Input(INPUT)
        SubscribableChannel input();

    }

}
複製代碼

內容很簡單，既包含了消息的生產，也包含了消息消費。在/sendMessage接口的定義中，發送了兩條消息，一條消息的頭信息中包含version=1.0，另一條消息的頭信息中包含version=2.0。在消息監聽類TestListener中，對TestTopic.INPUT通道定義了兩個@StreamListener，這兩個監聽邏輯有不一樣的condition，這裏的表達式表示會根據消息頭信息中的version值來作不一樣的處理邏輯分發。app

在啓動應用以前，還要記得配置一下輸入輸出通道對應的物理目標（exchange或topic名），好比：分佈式

spring.cloud.stream.bindings.example-topic-input.destination=test-topic
spring.cloud.stream.bindings.example-topic-input.group=stream-content-route
spring.cloud.stream.bindings.example-topic-output.destination=test-topic
複製代碼

完成了上面配置以後，就能夠啓動應用，並嘗試訪問localhost:8080/sendMessage?message=hello接口來發送一個消息到MQ中了。此時能夠看到相似下面的日誌：微服務

2018-12-24 15:50:33.361  INFO 17683 --- [content-route-1] c.d.stream.TestApplication$TestListener  : Received v1 : hello, 1.0
2018-12-24 15:50:33.363  INFO 17683 --- [content-route-1] c.d.stream.TestApplication$TestListener  : Received v2 : hello, 2.0
複製代碼

從日誌中能夠看到，兩條帶有不一樣頭信息的消息，分別經過不一樣的監聽處理邏輯輸出了對應的日誌打印。java B2B2C源碼電子商務平臺優化