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

通常在生产环境，我们的每个服务都不会以单节点的方式运行在生产环境，当同一个服务启动多个实例的时候，这些实例都会绑定到同一个消息通道的目标主题（Topic）上。

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

  • 先创建一个消费者应用SinkReceiver，实现了greetings主题上的输入通道绑定，它的实现如下： ```java @EnableBinding({Sink.class, MySink.class}) @Slf4j public class SinkReceiver {

    @StreamListener(Sink.INPUT) public void receive(User payload) {

    1.   log.info("Received: " + payload);

    } }

@Data class User { private String name; private Integer age; }


- 为了将SinkReceiver的输入通道目标设置为MyGroup主题，以及将该服务的实例设置为同一个消费组，做如下设置：
```java
spring.cloud.stream.bindings.input.group=Service-A
spring.cloud.stream.bindings.input.destination=MyGroup

通过spring.cloud.stream.bindings.input.group属性指定了该应用实例都属于Service-A消费组，而spring.cloud.stream.bindings.input.destination属性则指定了输入通道对应的主题名。

  • 完成了消息消费者之后，我们再来实现一个消息生产者应用SinkSender，具体如下：

    @EnableBinding(Source.class)
@Slf4j
public class SinkSend {

  @Bean
  @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "2000"))
  public MessageSource<String> sendMessage(){
      //@Poller(fixedDelay = "2000") 每隔两秒发送一次消息

      //MessageSource是一个函数式接口 相当于省略了匿名内部类的书写
      MessageSource<String> messageSource = new MessageSource<String>() {
          @Override
          public Message<String> receive() {
              //GenericMessage是Message的实现类
              return new GenericMessage<>("{}");
          }
      };
      //java8写法
      return  () -> new GenericMessage<>("{\"name\":\"snow\", \"age\":25}");
  }
}

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

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

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