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

简介: Spring Cloud Stream如何处理消息重复消费

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

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

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

详细也可查看原文:消息驱动的微服务(消费组)

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

问题重现

构建消息消费端

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

interface ExampleBinder {
    String NAME = "example-topic";
    @Input(NAME)
    SubscribableChannel input();
}

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

@EnableBinding(ExampleBinder.class)
public class ExampleReceiver {
    private static Logger logger = LoggerFactory.getLogger(ExampleReceiver.class);
    @StreamListener(ExampleBinder.NAME)
    public void receive(String payload) {
        logger.info("Received: " + payload);
    }
}

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

@SpringBootApplication
public class ExampleApplication {
    public static void main(String[] args) {
        SpringApplication.run(ExampleApplication.class, args);
    }
}
spring.application.name=stream-consumer-group
server.port=0

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

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

构建消息生产端

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

@RunWith(SpringRunner.class)
@EnableBinding(value = {ExampleApplicationTests.ExampleBinder.class})
public class ExampleApplicationTests {
  @Autowired
  private ExampleBinder exampleBinder;
  @Test
  public void exampleBinderTester() {
        exampleBinder.output().send(MessageBuilder.withPayload("Produce a message from : http://blog.didispace.com").build());
  }
  public interface ExampleBinder {
    String NAME = "example-topic";
    @Output(NAME)
    MessageChannel output();
  }
}

启动上述测试用例之后,可以发现之前启动的两个实例都收到的消息,并在日志中打印了:Received: Produce a message from : http://blog.didispace.com。消息重复消费的问题成功重现!

使用消费组解决问题

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

spring.cloud.stream.bindings.example-topic.group=aaa

当我们指定了某个绑定所指向的消费组之后,往当前主题发送的消息在每个订阅消费组中,只会有一个订阅者接收和消费,从而实现了对消息的负载均衡。只所以之前会出现重复消费的问题,是由于默认情况下,任何订阅都会产生一个匿名消费组,所以每个订阅实例都会有自己的消费组,从而当有消息发送的时候,就形成了广播的模式。

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

代码示例

本文示例读者可以通过查看下面仓库的中的stream-consumer-group项目:

如果您对这些感兴趣,欢迎star、follow、收藏、转发给予支持!

以下专题教程也许您会有兴趣

目录
相关文章
|
4月前
|
监控 负载均衡 Java
深入理解Spring Cloud中的服务网关
深入理解Spring Cloud中的服务网关
|
4月前
|
Java 开发工具 git
实现基于Spring Cloud的配置中心
实现基于Spring Cloud的配置中心
|
4月前
|
设计模式 监控 Java
解析Spring Cloud中的断路器模式原理
解析Spring Cloud中的断路器模式原理
|
4月前
|
资源调度 Java 调度
Spring Cloud Alibaba 集成分布式定时任务调度功能
Spring Cloud Alibaba 发布了 Scheduling 任务调度模块 [#3732]提供了一套开源、轻量级、高可用的定时任务解决方案,帮助您快速开发微服务体系下的分布式定时任务。
14979 29
|
4月前
|
负载均衡 Java Spring
Spring cloud gateway 如何在路由时进行负载均衡
Spring cloud gateway 如何在路由时进行负载均衡
511 15
|
3月前
|
消息中间件 Java 开发工具
【Azure 事件中心】Spring Cloud Stream Event Hubs Binder 发送Event Hub消息遇见 Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially 异常
【Azure 事件中心】Spring Cloud Stream Event Hubs Binder 发送Event Hub消息遇见 Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially 异常
|
4月前
|
Java Spring
spring cloud gateway在使用 zookeeper 注册中心时,配置https 进行服务转发
spring cloud gateway在使用 zookeeper 注册中心时,配置https 进行服务转发
115 3
|
4月前
|
缓存 Java 开发者
Spring循环依赖问题之Spring循环依赖如何解决
Spring循环依赖问题之Spring循环依赖如何解决
|
4月前
|
消息中间件 Java 开发者
Spring Cloud微服务框架:构建高可用、分布式系统的现代架构
Spring Cloud是一个开源的微服务框架,旨在帮助开发者快速构建在分布式系统环境中运行的服务。它提供了一系列工具,用于在分布式系统中配置、服务发现、断路器、智能路由、微代理、控制总线、一次性令牌、全局锁、领导选举、分布式会话、集群状态等领域的支持。
191 5
下一篇
无影云桌面