Spring Cloud Stream如何处理消息重复消费？

最近收到好几个类似的问题：使用Spring Cloud Stream操作RabbitMQ或Kafka的时候，出现消息重复消费的问题。通过沟通与排查下来主要还是用户对消费组的认识不够。其实，在之前的博文以及《Spring Cloud微服务实战》一书中都有提到关于消费组的概念以及作用。

那么什么是消费组呢？为什么要用消费组？它解决什么问题呢？摘录一段之前博文的内容，来解答这些疑问：

通常在生产环境，我们的每个服务都不会以单节点的方式运行在生产环境，当同一个服务启动多个实例的时候，这些实例都会绑定到同一个消息通道的目标主题（Topic）上。默认情况下，当生产者发出一条消息到绑定通道上，这条消息会产生多个副本被每个消费者实例接收和处理（出现上述重复消费问题）。但是有些业务场景之下，我们希望生产者产生的消息只被其中一个实例消费，这个时候我们需要为这些消费者设置消费组来实现这样的功能。

下面，通过一个例子来看看如何使用消费组。

问题重现

构建消息消费端

第一步：创建绑定接口，绑定example-topic输入通道（默认情况下，会绑定到RabbitMQ的同名Exchange或Kafaka的同名Topic）。

第二步：对上述输入通道创建监听与处理逻辑。

第三步；创建应用主类和配置文件

这里设置server.port=0，以方便在本地启动多实例来重现问题。

完成上述操作之后，启动两个该应用的实例，以备后续调用。

构建消息生产端

比较简单，需要注意的是，使用@Output创建一个同名的输出绑定，这样发出的消息才能被上述启动的实例接收到。具体实现如下：

使用消费组解决问题

如何解决上述消息重复消费的问题呢？我们只需要在配置文件中增加如下配置即可：

另外，需要注意上述配置中example-topic是在代码中@Output和@Input中传入的名字。

-END-