美文网首页
(八十一)java版spring cloud+spring bo

(八十一)java版spring cloud+spring bo

作者: IT小跑兵 | 来源:发表于2019-06-18 09:55 被阅读0次

    电子商务平台源码请加企鹅求求:三伍三六贰四柒二伍九。使用消费组实现消息消费的负载均衡

    通常在生产环境,我们的每个服务都不会以单节点的方式运行在生产环境,当同一个服务启动多个实例的时候,这些实例都会绑定到同一个消息通道的目标主题(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
    

    到这里,对于消费分组的示例就已经完成了。分别运行上面实现的生产者与消费者,其中消费者我们启动多个实例。通过控制台,我们可以发现每个生产者发出的消息,会被启动的消费者以轮询的方式进行接收和输出。

    相关文章

      网友评论

          本文标题:(八十一)java版spring cloud+spring bo

          本文链接:https://www.haomeiwen.com/subject/edsxqctx.html