Spring Cloud構建微服務架構:訊息驅動的微服務(消費組)【Dalston版】
阿新 • • 發佈:2022-05-05
通過之前的《訊息驅動的微服務(入門)》一文,相信很多朋友已經對Spring Cloud Stream有了一個初步的認識。但是,對於《訊息驅動的微服務(核心概念)》一文中提到的一些核心概念可能還有些迷糊,下面我們將詳細的來學習一下這些概念。本文我們就來學習和使用一下“消費組”這一概念。
使用消費組實現訊息消費的負載均衡
通常在生產環境,我們的每個服務都不會以單節點的方式執行在生產環境,當同一個服務啟動多個例項的時候,這些例項都會繫結到同一個訊息通道的目標主題(Topic)上。
預設情況下,當生產者發出一條訊息到繫結通道上,這條訊息會產生多個副本被每個消費者例項接收和處理,但是有些業務場景之下,我們希望生產者產生的訊息只被其中一個例項消費,這個時候我們需要為這些消費者設定消費組來實現這樣的功能,實現的方式非常簡單,我們只需要在服務消費者端設定 spring.cloud.stream.bindings.input.group
- 先建立一個消費者應用
SinkReceiver
,實現了greetings
主題上的輸入通道繫結,它的實現如下:
@EnableBinding(value = {Sink.class}) public class SinkReceiver { private static Logger logger = LoggerFactory.getLogger(SinkReceiver.class); @StreamListener(Sink.INPUT) public void receive(User user) { logger.info("Received: " + user); } }
- 為了將
SinkReceiver
的輸入通道目標設定為greetings
主題,以及將該服務的例項設定為同一個消費組,做如下設定:
spring.cloud.stream.bindings.input.group=Service-A
spring.cloud.stream.bindings.input.destination=greetings
通過 spring.cloud.stream.bindings.input.group
屬性指定了該應用例項都屬於 Service-A
消費組,而 spring.cloud.stream.bindings.input.destination
屬性則指定了輸入通道對應的主題名。
- 完成了訊息消費者之後,我們再來實現一個訊息生產者應用
SinkSender
,具體如下:
@EnableBinding(value = {Source.class})
public class SinkSender {
private static Logger logger = LoggerFactory.getLogger(SinkSender.class);
@Bean
@InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "2000"))
public MessageSource<String> timerMessageSource() {
return () -> new GenericMessage<>("{"name":"didi", "age":30}");
}
}
- 為訊息生產者
SinkSender
做一些設定,讓它的輸出通道繫結目標也指向greetings
主題,具體如下:
spring.cloud.stream.bindings.output.destination=greetings
到這裡,對於消費分組的示例就已經完成了。分別執行上面實現的生產者與消費者,其中消費者我們啟動多個例項。通過控制檯,我們可以發現每個生產者發出的訊息,會被啟動的消費者以輪詢的方式進行接收和輸出。
部落格原文:http://blog.didispace.com/spring-cloud-starter-dalston-7-3/
但對依賴的Spring Boot和Spring Cloud版本做了升級。