java B2B2C springmvc mybatis電子商務平臺原始碼-訊息驅動的微服務(消費組)
使用消費組實現訊息消費的負載均衡
需要JAVA Spring Cloud大型企業分散式微服務雲構建的B2B2C電子商務平臺原始碼:壹零叄八柒柒肆六二六
通常在生產環境,我們的每個服務都不會以單節點的方式執行在生產環境,當同一個服務啟動多個例項的時候,這些例項都會繫結到同一個訊息通道的目標主題(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
到這裡,對於消費分組的示例就已經完成了。分別執行上面實現的生產者與消費者,其中消費者我們啟動多個例項。通過控制檯,我們可以發現每個生產者發出的訊息,會被啟動的消費者以輪詢的方式進行接收和輸出。