Spring Cloud Stream如何消費自己生產的訊息
摘要：
在上一篇《Spring Cloud Stream如何處理訊息重複消費》
中，我們通過消費組的配置解決了多例項部署情況下訊息重複消費這一入門時的常見問題。本文將繼續說說在另外一個被經常問到的問題：如果微服務生產的訊息自己也想要消費一份，應該如何實現呢？
常見錯誤
在放出...
在上一篇ofollow,noindex" target="_blank">《Spring Cloud Stream如何處理訊息重複消費》 中，我們通過消費組的配置解決了多例項部署情況下訊息重複消費這一入門時的常見問題。本文將繼續說說在另外一個被經常問到的問題：如果微服務生產的訊息自己也想要消費一份，應該如何實現呢？
常見錯誤
在放出標準答案前，先放出一個常見的錯誤姿勢和告警資訊（以便您可以通過搜尋引擎找到這裡^_^）。以下錯誤基於Spring Boot 2.0.5、Spring Cloud Finchley SR1。
首先，根據入門示例，為了生產和消費訊息，需要定義兩個通道：一個輸入、一個輸出。比如下面這樣：
public interface TestTopic {
String OUTPUT = "example-topic";
String INPUT = "example-topic";
@Output(OUTPUT)
MessageChannel output();
@Input(INPUT)
SubscribableChannel input();
}
通過INPUT
和OUTPUT
使用相同的名稱，讓生產訊息和消費訊息指向相同的Topic，從而實現消費自己發出的訊息。
接下來，建立一個HTTP介面，並通過上面定義的輸出通道觸來生產訊息，比如：
@Slf4j
@RestController
public class TestController {
@Autowired
private TestTopic testTopic;
@GetMapping("/sendMessage")
public String messageWithMQ(@RequestParam String message) {
testTopic.output().send(MessageBuilder.withPayload(message).build());
return "ok";
}
}
已經有生產訊息的實現，下面來建立對輸入通道的監聽，以實現訊息的消費邏輯。
@Slf4j
@Component
public class TestListener {
@StreamListener(TestTopic.INPUT)
public void receive(String payload) {
log.info("Received: " + payload);
throw new RuntimeException("BOOM!");
}
}
最後，在應用主類中，使用@EnableBinding註解來開啟它，比如：
@EnableBinding(TestTopic.class)
@SpringBootApplication
public class TestApplication {
public static void main(String[] args) {
SpringApplication.run(TestApplication.class, args);
}
}
看似天衣無縫的操作，然而在啟動的瞬間，你可能收到了下面這樣的錯誤：
org.springframework.beans.factory.BeanDefinitionStoreException: Invalid bean definition with name 'example-topic' defined in com.didispace.stream.TestTopic: bean definition with this name already exists - Root bean: class [null]; scope=; abstract=false; lazyInit=false; autowireMode=0; dependencyCheck=0; autowireCandidate=true; primary=false; factoryBeanName=com.didispace.stream.TestTopic; factoryMethodName=input; initMethodName=null; destroyMethodName=null at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinition(BindingBeanDefinitionRegistryUtils.java:64) ~[spring-cloud-stream-2.0.1.RELEASE.jar:2.0.1.RELEASE] at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerOutputBindingTargetBeanDefinition(BindingBeanDefinitionRegistryUtils.java:54) ~[spring-cloud-stream-2.0.1.RELEASE.jar:2.0.1.RELEASE] at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.lambda$registerBindingTargetBeanDefinitions$0(BindingBeanDefinitionRegistryUtils.java:86) ~[spring-cloud-stream-2.0.1.RELEASE.jar:2.0.1.RELEASE] at org.springframework.util.ReflectionUtils.doWithMethods(ReflectionUtils.java:562) ~[spring-core-5.0.9.RELEASE.jar:5.0.9.RELEASE] at org.springframework.util.ReflectionUtils.doWithMethods(ReflectionUtils.java:541) ~[spring-core-5.0.9.RELEASE.jar:5.0.9.RELEASE] at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinitions(BindingBeanDefinitionRegistryUtils.java:76) ~[spring-cloud-stream-2.0.1.RELEASE.jar:2.0.1.RELEASE] at org.springframework.cloud.stream.config.BindingBeansRegistrar.registerBeanDefinitions(BindingBeansRegistrar.java:45) ~[spring-cloud-stream-2.0.1.RELEASE.jar:2.0.1.RELEASE] at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.lambda$loadBeanDefinitionsFromRegistrars$1(ConfigurationClassBeanDefinitionReader.java:358) ~[spring-context-5.0.9.RELEASE.jar:5.0.9.RELEASE] at java.util.LinkedHashMap.forEach(LinkedHashMap.java:684) ~[na:1.8.0_151] at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitionsFromRegistrars(ConfigurationClassBeanDefinitionReader.java:357) ~[spring-context-5.0.9.RELEASE.jar:5.0.9.RELEASE] at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitionsForConfigurationClass(ConfigurationClassBeanDefinitionReader.java:145) ~[spring-context-5.0.9.RELEASE.jar:5.0.9.RELEASE] at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitions(ConfigurationClassBeanDefinitionReader.java:117) ~[spring-context-5.0.9.RELEASE.jar:5.0.9.RELEASE] at org.springframework.context.annotation.ConfigurationClassPostProcessor.processConfigBeanDefinitions(ConfigurationClassPostProcessor.java:328) ~[spring-context-5.0.9.RELEASE.jar:5.0.9.RELEASE] at org.springframework.context.annotation.ConfigurationClassPostProcessor.postProcessBeanDefinitionRegistry(ConfigurationClassPostProcessor.java:233) ~[spring-context-5.0.9.RELEASE.jar:5.0.9.RELEASE] at org.springframework.context.support.PostProcessorRegistrationDelegate.invokeBeanDefinitionRegistryPostProcessors(PostProcessorRegistrationDelegate.java:271) ~[spring-context-5.0.9.RELEASE.jar:5.0.9.RELEASE] at org.springframework.context.support.PostProcessorRegistrationDelegate.invokeBeanFactoryPostProcessors(PostProcessorRegistrationDelegate.java:91) ~[spring-context-5.0.9.RELEASE.jar:5.0.9.RELEASE] at org.springframework.context.support.AbstractApplicationContext.invokeBeanFactoryPostProcessors(AbstractApplicationContext.java:694) ~[spring-context-5.0.9.RELEASE.jar:5.0.9.RELEASE] at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:532) ~[spring-context-5.0.9.RELEASE.jar:5.0.9.RELEASE] at org.springframework.boot.web.reactive.context.ReactiveWebServerApplicationContext.refresh(ReactiveWebServerApplicationContext.java:61) ~[spring-boot-2.0.5.RELEASE.jar:2.0.5.RELEASE] at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:780) [spring-boot-2.0.5.RELEASE.jar:2.0.5.RELEASE] at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:412) [spring-boot-2.0.5.RELEASE.jar:2.0.5.RELEASE] at org.springframework.boot.SpringApplication.run(SpringApplication.java:333) [spring-boot-2.0.5.RELEASE.jar:2.0.5.RELEASE] at org.springframework.boot.SpringApplication.run(SpringApplication.java:1277) [spring-boot-2.0.5.RELEASE.jar:2.0.5.RELEASE] at org.springframework.boot.SpringApplication.run(SpringApplication.java:1265) [spring-boot-2.0.5.RELEASE.jar:2.0.5.RELEASE] at com.didispace.stream.TestApplication.main(TestApplication.java:13) [classes/:na]
正確姿勢
根據錯誤提示：Invalid bean definition with name 'example-topic' defined in com.didispace.stream.TestTopic: bean definition with this name already exists
，沒有啟動成功的原因是已經存在了一個名為example-topic
的Bean，那麼為什麼會重複建立這個Bean呢？
實際上，在F版的Spring Cloud Stream中，當我們使用@Output
和@Input
註解來定義訊息通道時，都會根據傳入的通道名稱來建立一個Bean。而在上面的例子中，我們定義的@Output
和@Input
名稱是相同的，因為我們系統輸入和輸出是同一個Topic，這樣才能實現對自己生產訊息的消費。
既然這樣，我們定義相同的通道名是行不通了，那麼我們只能通過定義不同的通道名，併為這兩個通道配置相同的目標Topic來將這一對輸入輸出指向同一個實際的Topic。對於上面的錯誤程式，只需要做如下兩處改動：
第一步：修改通道名，使用不同的名字
public interface TestTopic {
String OUTPUT = "example-topic-output";
String INPUT = "example-topic-input";
@Output(OUTPUT)
MessageChannel output();
@Input(INPUT)
SubscribableChannel input();
}
第二步：在配置檔案中，為這兩個通道設定相同的Topic名稱，比如：
spring.cloud.stream.bindings.example-topic-input.destination=aaa-topic spring.cloud.stream.bindings.example-topic-output.destination=aaa-topic
這樣，這兩個輸入輸出通道就會都指向名為aaa-topic
的Topic了。
最後，再啟動該程式，沒有報錯。然後訪問介面：localhost:8080/sendMessage?message=hello-didi
，可以在控制檯中看到如下資訊：
2018-11-17 23:24:10.425INFO 32039 --- [ctor-http-nio-2] o.s.a.r.c.CachingConnectionFactory: Attempting to connect to: [localhost:5672] 2018-11-17 23:24:10.453INFO 32039 --- [ctor-http-nio-2] o.s.a.r.c.CachingConnectionFactory: Created new connection: rabbitConnectionFactory.publisher#266753da:0/SimpleConnection@627fba83 [delegate=amqp://[email protected]:5672/, localPort= 60752] 2018-11-17 23:24:10.458INFO 32039 --- [ctor-http-nio-2] o.s.amqp.rabbit.core.RabbitAdmin: Auto-declaring a non-durable, auto-delete, or exclusive Queue (aaa-topic.anonymous.fNUxZ8C0QIafxrhkFBFI1A) durable:false, auto-delete:true, exclusive:true. It will be redeclared if the broker stops and is restarted while the connection factory is alive, but all messages will be lost. 2018-11-17 23:24:10.483INFO 32039 --- [IafxrhkFBFI1A-1] com.didispace.stream.TestListener: Received: hello-didi
消費自己生產的訊息成功了！讀者也還可以訪問一下應用的/actuator/beans
端點，看看當前Spring上下文中有哪些Bean，應該可以看到有下面Bean，也就是上面分析的兩個通道的Bean物件
"example-topic-output": {
"aliases": [],
"scope": "singleton",
"type": "org.springframework.integration.channel.DirectChannel",
"resource": null,
"dependencies": []
},
"example-topic-input": {
"aliases": [],
"scope": "singleton",
"type": "org.springframework.integration.channel.DirectChannel",
"resource": null,
"dependencies": []
},
後記
其實大部分開發者在使用Spring Cloud Stream時候碰到的問題都源於對Spring Cloud Stream的核心概念還是不夠理解。所以，還是推薦讀一下下面的文章和示例：
程式碼示例
本文示例讀者可以通過檢視下面倉庫的中的stream-consumer-self
專案：