Spring Cloud构建微服务架构：消息驱动的微服务（消费组）【Dalston版】

亦凉 2021-12-05 15:23 260阅读 0赞

通过之前的[《消息驱动的微服务（入门）》][Link 1]一文，相信很多朋友已经对Spring Cloud Stream有了一个初步的认识。但是，对于[《消息驱动的微服务（核心概念）》][Link 2]一文中提到的一些核心概念可能还有些迷糊，下面我们将详细的来学习一下这些概念。本文我们就来学习和使用一下“消费组”这一概念。

**重点：消费组可以保证一条消息只能被该组内的一个消费者实例来消费**

### 使用消费组实现消息消费的负载均衡 ###

通常在生产环境，我们的每个服务都不会以单节点的方式运行在生产环境，当同一个服务启动多个实例的时候，这些实例都会绑定到同一个消息通道的目标主题（Topic）上（**解释：Topic在RabbitMQ中的意思就是Exchange,在rabbitmq里面会有多个Exchange对象，而每一个Exchange对象里面可以包含很多的消息生产者的输出通道channel和消息消费者的输入通道channel**）。

默认情况下，当生产者发出一条消息到绑定通道上，这条消息会产生多个副本被每个消费者实例接收和处理，但是有些业务场景之下，我们希望生产者产生的消息只被其中一个实例消费，这个时候我们需要为这些消费者设置消费组来实现这样的功能，实现的方式非常简单，我们只需要在服务消费者端设置`spring.cloud.stream.bindings.input.group`属性即可，比如我们可以这样实现：

*  先创建一个消费者应用SinkReceiver，实现了greetings主题上的输入通道绑定，它的实现如下：  
    **注意：** 这里的模式是“**订阅模式**”，不是“**点对点，一对一**”，所以不需要保证消费者的接收通道和生产者的接收通道相同，只需要保证他们是在一个Topic（exchange）里面即可，因为一个Topic（exchange交换机）里面可以有多个消费者通道（一个通道只能对应一个消费者），多个生产者通道（一个通道只能对应一个生产者）

@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);
        }
    }

User.java

package com.liu.streamhello;
    
    import java.io.Serializable;
    
    public class User implements Serializable {
        private static final long serialVersionUID = 695183437612916152L;
        private String username;
        private int age;
    
        public String getUsername() {
            return username;
        }
    
        public User setUsername(String username) {
            this.username = username;
            return this;
        }
    
        public int getAge() {
            return age;
        }
    
        public User setAge(int age) {
            this.age = age;
            return this;
        }
    }

*  为了将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这个注解下面有讲解
        @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "2000"))
        public MessageSource<String> timerMessageSource() {
            return () -> new GenericMessage<>("{\"name\":\"didi\", \"age\":30}");
        }
    
    }

**讲解：** @InboundChannelAdapter注解应该是生产者的推送的注解，value代表要推送到哪个消息通道（channel）上去，@Poller是生产者推送的注解，fixedDelay是推送的间隔时间周期，每隔一个周期推送一次。  
虽然可以看到生产者推送的通道是Source.OUTPUT,而消费者接收消息的通道是Sink.INPUT，两个通道虽然不一样，但是别忘了，我们这个可不是点对点，一对一的模式，而是订阅topic模式，只要他们生产者和消费者订阅的是同一个topic即可保持通信（**解释：一个topic里面可以有多条消息生产者通道和消费者通道**）

*  为消息生产者SinkSender做一些设置，让它的输出通道绑定目标也指向greetings主题，具体如下：

spring.cloud.stream.bindings.output.destination=greetings

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

本文的原文来自于：[http://blog.didispace.com/spring-cloud-starter-dalston-7-3/][http_blog.didispace.com_spring-cloud-starter-dalston-7-3]

[Link 1]: https://blog.csdn.net/qq_40241957/article/details/95333599
[Link 2]: https://blog.csdn.net/qq_40241957/article/details/95350203
[http_blog.didispace.com_spring-cloud-starter-dalston-7-3]: http://blog.didispace.com/spring-cloud-starter-dalston-7-3/