文档章节

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

程序猿DD
 程序猿DD
发布于 11/26 20:58
字数 1057
阅读 1545
收藏 42

最近收到好几个类似的问题:使用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、收藏、转发给予支持!

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

© 著作权归作者所有

共有 人打赏支持
程序猿DD
粉丝 388
博文 65
码字总数 85516
作品 4
闵行
私信 提问
Spring Cloud (十五)Stream 入门、主要概念与自定义消息发送与接收

前言 不写随笔的日子仿佛就是什么都没有产出一般……上节说到要学Spring Cloud Bus,这里发现按照官方文档的顺序反而会更好些,因为不必去后边的章节去为当前章节去打基础,所以我们先学习S...

东北小狐狸
07/31
0
0
Spring Cloud Stream如何消费自己生产的消息?

在上一篇《Spring Cloud Stream如何处理消息重复消费》中,我们通过消费组的配置解决了多实例部署情况下消息重复消费这一入门时的常见问题。本文将继续说说在另外一个被经常问到的问题:如果...

程序猿DD
11/28
0
0
Spring Cloud Stream应用与自定义RocketMQ Binder:编程模型

前言: 本文作者张天,节选自笔者与其合著的《Spring Cloud微服务架构进阶》,即将在八月出版问世。本文将其中Spring Cloud Stream应用与自定义Rocketmq Binder的内容抽取出来,主要介绍Spr...

aoho
06/23
0
0
SpringCloud实战9-Stream消息驱动

官方定义 Spring Cloud Stream 是一个构建消息驱动微服务的框架。   应用程序通过 inputs 或者 outputs 来与 Spring Cloud Stream 中binder 交互,通过我们配置来 binding ,而 Spring Clo...

狂小白
05/29
0
0
(十七) 整合spring cloud云架构 -消息驱动 Spring Cloud Stream

在使用spring cloud云架构的时候,我们不得不使用Spring cloud Stream,因为消息中间件的使用在项目中无处不在,我们公司后面做了娱乐方面的APP,在使用spring cloud做架构的时候,其中消息的...

itcloud
08/28
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Spak—— sparkCore源码解析之RangePartitioner源码

   分区过程概览 RangePartitioner分区执行原理: 计算总体的数据抽样大小sampleSize,计算规则是:至少每个分区抽取20个数据或者最多1M的数据量。 根据sampleSize和分区数量计算每个分区的...

freeli
9分钟前
0
0
从内部自用到对外服务,配置管理的演进和设计优化实践

本文整理自阿里巴巴中间件技术专家彦林在中国开源年会上的分享,通过此文,您将了解到: 微服务给配置管理所带来的变化 配置管理演进过程中的设计思考 配置管理开源后的新探索 配置中心控制台...

阿里云官方博客
10分钟前
0
0
MySQL用户管理,常用MySQL语句、MySQL数据库备份恢复

12月6日任务 13.4 mysql用户管理 13.5 常用sql语句 13.6 mysql数据库备份恢复 13.4 mysql用户管理 grant all on *.* to 'user1' identified by 'passwd'; grant SELECT,UPDATE,INSERT on db......

zgxlinux
11分钟前
1
0
Spring异常之Druid – unregister mbean error

Spring异常之Druid – unregister mbean error 2017年04月19日 12:13:42 Dr.Zhu 阅读数:6688 版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/zt_fucker/arti...

linjin200
16分钟前
1
0
微信小程序webview问题

今天在改小程序的时候在使用webview的时候切换webview的地址行为,出现了诡异的情况。 默认querystring里会有多个?符号,使用的时候被微信给截取了,导致程序找不到改页面。 而且querystri...

钟元OSS
19分钟前
1
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部