ActiveMQ第六弹:设置多个并行的消费者

简介: 消息队列本来就是一种经典的生产者与消费者模式。生产者向消息队列中发送消息,消费者从消息队列中获取消息来消费。 消息的传送一般由一个代理来实现的,那就是Message broker(即消息代理)。

消息队列本来就是一种经典的生产者与消费者模式。生产者向消息队列中发送消息,消费者从消息队列中获取消息来消费。

img_0fe8379de75917e2a6cf8c3e55d44630.png

消息的传送一般由一个代理来实现的,那就是Message broker(即消息代理)。Message broker有两大职责,一是消息路由,二是数据转换。这就好比A给B寄信,如果不使用邮局的话,就要自己想办法送达,费时费力,而通过邮局的话,只要B的地址在邮局中注册过,那么天涯海角也能送达。这里的邮局扮演的角色就像消息系统中的Message broker。

img_5e4597b91f0d1084061ab5e0b108fe0c.png

众所周知,消息队列是典型的’send and forget’原则的体现,生产者只管发送,不管消息的后续处理。为了最大效率的完成对消息队列中的消息的消费,一般可以同时起多个一模一样的消费者,以并行的方式来拉取消息队列中的消息。这样的好处有多个:

  1. 加快处理消息队列中的消息。

  2. 增强稳定性,如果一个消费者出现问题,不会影响对消息队列中消息的处理。

img_fde96d07ea2242d1157aa80114e1c2cc.png

使用Spring JMS来配置多个Listener实例其实也相当简单,只需要配置下MessageListenerContainer就行。

1
2
3
4
5
6
<bean class="org.springframework.jms.listener.SimpleMessageListenerContainer">
    <property name="connectionFactory" ref="connectionFactory"/>
    <property name="destinationName" value="${jms.queue.name}"/>
    <property name="messageListener" ref="messageReceiver"/>
    <property name="concurrentConsumers" value="4"/>
</bean>

多配置一个属性concurrentConsumers,设置值为4,就是同时启动4个Listener实例来消费消息。

使用MessageSender来发送100条消息,可以检查消息处理的顺序会发生变化。

1
2
3
    for (int i = 0; i < 100; i++) {
        messageSender.send(String.format("message %d",i));
    }
1
2
3
4
5
6
7
8
9
...
Received: message 4
Received: message 7
Received: message 6
Received: message 5
Received: message 8
Received: message 10
Received: message 9

除了设置一个固定的Listener数量,也可以设置一个Listener区间,这样MessageListenerContainer可以根据消息队列中的消息规模自动调整并行数量。

1
2
3
4
5
6
    <bean class="org.springframework.jms.listener.SimpleMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="destinationName" value="${jms.queue.name}"/>
        <property name="messageListener" ref="messageReceiver"/>
        <property name="concurrency" value="4-8"/>
    </bean>

这次使用的是concurrency属性,4-8表示最小并发数是4,最大并发数为8,当然也可以给一个固定值,比如5,这样就相当于concurrentConsumers属性了。

本章中的完整源码可从完整代码可从https://github.com/huangbowen521/SpringJMSSample下载。

相关文章
|
4月前
|
消息中间件 Java Kafka
如何在Kafka分布式环境中保证消息的顺序消费?深入剖析Kafka机制,带你一探究竟!
【8月更文挑战第24天】Apache Kafka是一款专为实时数据管道和流处理设计的分布式平台,以其高效的消息发布与订阅功能著称。在分布式环境中确保消息按序消费颇具挑战。本文首先介绍了Kafka通过Topic分区实现消息排序的基本机制,随后详细阐述了几种保证消息顺序性的策略,包括使用单分区Topic、消费者组搭配单分区消费、幂等性生产者以及事务支持等技术手段。最后,通过一个Java示例演示了如何利用Kafka消费者确保消息按序消费的具体实现过程。
138 3
|
5月前
|
消息中间件 Java Kafka
kafka Linux环境搭建安装及命令创建队列生产消费消息
kafka Linux环境搭建安装及命令创建队列生产消费消息
113 4
|
5月前
|
消息中间件 RocketMQ
MetaQ/RocketMQ 原理问题之当消费集群规模较大时,处理分配不到队列的Consumer的问题如何解决
MetaQ/RocketMQ 原理问题之当消费集群规模较大时,处理分配不到队列的Consumer的问题如何解决
|
6月前
|
消息中间件 监控 应用服务中间件
消息队列 MQ操作报错合集之重启Broker后,积压数出现为负数是什么导致的
在使用消息队列MQ时,可能会遇到各种报错情况。以下是一些常见的错误场景、可能的原因以及解决建议的汇总:1.连接错误、2.消息发送失败、3.消息消费报错、4.消息重试与死信处理、5.资源与权限问题、6.配置错误、7.系统资源限制、8.版本兼容性问题。
168 1
消息队列 MQ操作报错合集之重启Broker后,积压数出现为负数是什么导致的
|
6月前
|
消息中间件 运维 Apache
消息队列 MQ产品使用合集之消费者在消费完成后没有关闭链接,导致连接数达到上限,该如何解决
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
6月前
|
消息中间件 存储 负载均衡
消息队列 MQ产品使用合集之POP消费模式是否可以保证消息顺序性
阿里云消息队列MQ(Message Queue)是一种高可用、高性能的消息中间件服务,它允许您在分布式应用的不同组件之间异步传递消息,从而实现系统解耦、流量削峰填谷以及提高系统的可扩展性和灵活性。以下是使用阿里云消息队列MQ产品的关键点和最佳实践合集。
|
7月前
|
消息中间件
第十五章 RabbitMQ 延迟队列
第十五章 RabbitMQ 延迟队列
42 0
|
7月前
|
消息中间件 网络协议 Kafka
Kafka【付诸实践 02】消费者和消费者群组+创建消费者实例+提交偏移量(自动、手动)+监听分区再平衡+独立的消费者+消费者其他属性说明(实例源码粘贴可用)【一篇学会使用Kafka消费者】
【2月更文挑战第21天】Kafka【付诸实践 02】消费者和消费者群组+创建消费者实例+提交偏移量(自动、手动)+监听分区再平衡+独立的消费者+消费者其他属性说明(实例源码粘贴可用)【一篇学会使用Kafka消费者】
212 3
|
7月前
|
消息中间件 监控 中间件
【工作中问题解决实践 十一】Kafka消费者消费堆积且频繁rebalance
【工作中问题解决实践 十一】Kafka消费者消费堆积且频繁rebalance
440 0