springcloud系列—Stream—第8章-3: Spring Cloud Stream 訊息驅動(消費組)
阿新 • • 發佈:2019-01-04
使用消費組實現訊息消費的負載均衡
通常在生產環境,我們的每個服務都不會以單節點的方式執行在生產環境,當同一個服務啟動多個例項的時候,這些例項都會繫結到同一個訊息通道的目標主題(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 |
到這裡,對於消費分組的示例就已經完成了。分別執行上面實現的生產者與消費者,其中消費者我們啟動多個例項。通過控制檯,我們可以發現每個生產者發出的訊息,會被啟動的消費者以輪詢的方式進行接收和輸出。