RabbitMQ下的生产消费者模式与订阅发布模式

简介:

所谓模式,就是在某种场景下,一类问题及其解决方案的总结归纳。生产消费者模式与订阅发布模式是使用消息中间件时常用的两种模式,用于功能解耦和分布式系统间的消息通信,以下面两种场景为例:

  • 数据接入 
    假设有一个用户行为采集系统,负责从App端采集用户点击行为数据。通常会将数据上报和数据处理分离开,即App端通过REST API上报数据,后端拿到数据后放入队列中就立刻返回,而数据处理则另外使用Worker从队列中取出数据来做,如下图所示。


    这样做的好处有:第一,功能分离,上报的API接口不关心数据处理功能,只负责接入数据;第二,数据缓冲,数据上报的速率是不可控的,取决于用户使用频率,采用该模式可以一定程度地缓冲数据;第三,易于扩展,在数据量大时,通过增加数据处理Worker来扩展,提高处理速率。这便是典型的生产消费者模式,数据上报为生产者,数据处理为消费者。

  • 事件分发 
    假设有一个电商系统,那么,用户“收藏”、“下单”、“付款”等行为都是非常重要的事件,通常后端服务在完成相应的功能处理外,还需要在这些事件点上做很多其他处理动作,比如发送短信通知、记录用户积分等等。我们可以将这些额外的处理动作放到每个模块中,但这并不是优雅的实现,不利于功能解耦和代码维护。 
    我们需要的是一个事件分发系统,在各个功能模块中将对应的事件发布出来,由对其感兴趣的处理者进行处理。这里涉及两个角色:A对B感兴趣,A是处理者,B是事件,由事件处理器完成二者的绑定,并向消息中心订阅事件。服务模块是后端的业务逻辑服务,在不同的事件点发布事件,事件经过消息中心分发给事件处理器对应的处理者。整个流程如下图所示。这边是典型的订阅发布模式

     

  可以看到,生产消费者模式与订阅发布模式都离不开消息中间件来作为消息中转站,开源的消息中间件有很多,各有优劣。本文将重点探讨RabbitMQ的特性,以及如何实现上述的两种场景。

RabbitMQ核心概念

  如果你只是想使用一下RabbitMQ,那么参考官方教程修改一下就可以跑起来了,很简单。如果你有一些概念上的疑惑,不妨与笔者一起来总结一下RabbitMQ的核心概念。

  • 通信方式 
    RabbitMQ是基于AMQP协议来实现的消息中间件。AMQP,类似于HTTP协议,也是一个应用层的协议,网络层使用TCP来通信。因此,RabbitMQ也是典型的C-S模型,准确地说是C-S-C模型,因为伴随着RabbitMQ的使用,总是会有Producer与Consumer两个Client和一个Broker Server。


    Client要与Server进行通信,就必须先建立连接,RabbitMQ中有Connection与Channel两个概念,前者就是一个TCP连接,后者是在这个连接上的虚拟概念,负责逻辑上的数据传递,因此,为了节省资源,一般在一个客户端中建立一个Connection,每次使用时再分配一个Channel即可。

  • 消息体 
    Message是RabbitMQ中的消息体概念。类似HTTP传输中,有header和body两部分数据,Message中也有Attributes和Payload两部分数据,前者是一些元信息,后者是传递的消息数据实体。

  • 消息投递 
    Exchange、Queue与Routing Key三个概念是理解RabbitMQ消息投递的关键。RabbitMQ中一个核心的原则是,消息不能直接投递到Queue中。Producer只能将自己的消息投递到Exchange中,由Exchange按照routing_key投递到对应的Queue中,具体的架构参见下图。细细品味就会体会到这样设计的精妙之处。


    那么,具体实现时,如何完成这三者关系的绑定?总结起来是两点:第一,在Consumer Worker中,声明自己对哪个Exchange感兴趣,并将自己的Queue绑定到自己感兴趣的一组routing_key上,建立相应的映射关系;第二,在Producer中,将消息投递一个Exchange中,并指明它的routing_key。由此可见,Queue这个概念只是对Consumer可见,Producer并不关心消息被投递到哪个Queue中。 
    看过RabbitMQ的”Hello World”教程的童鞋可能会发现在那里面的图中并没有看到Exchange和routing_key的踪迹,但这并不意味着RabbitMQ可以支持直接将消息投递到Queue中,而是在内部使用了默认的Exchange和routing_key了。默认情况下,RabbitMQ使用名称为“amq.direct”的Direct Exchange,routing_key默认名字与Queue保持一致。 
    搞清楚上述概念,就不难理解Exchange的四种类型了。Direct、Fanout、Topic、Headers,区别在于如何将消息从Exchange投递到Queue中。Direct使用具体的routing_key来投递;Fanout则忽略routing_key,直接广播给所有的Queue;Topic是使用模糊匹配来对一组routing_key进行投递;Headers也是忽略routing_key,使用消息中的Headers信息来投递。

  • 消息可靠性 
    不同于HTTP的同步访问,RabbitMQ中,Producer并不知道消息是否被可靠地投递到了Consumer中处理。那么,RabbitMQ是如何保证消息的可靠投递?主要是两点:第一,消息确认机制。Consumer处理完消息后,需要发送确认消息给Broker Server,可以选择“确认接收”、“丢弃”、“重新投递”三种方式。如果Consumer在Broker Server收到确认消息之前挂了,Broker Server便会重新投递该消息。第二,可以选择数据持久化,这样即使RabbitMQ重启,也不会丢失消息。

生产消费者模式

  搞清楚了RabbitMQ的核心概念,要针对特定的场景来设计使用方案就很简单了,基本上就是上述RabbitMQ架构图的变迁。让我们先来看看文章开头提到的“数据接入”场景,如何实现生产消费者模式。 



这里增加了一下场景复杂度:对于上报的数据,如果是special的行为,需要优先处理。从上图可以看到,数据上报端负责将数据投递到RabbitMQ对应的Exchange,并指明routing_key是common还是special。数据处理端,可以根据情况启多个Woker来消费数据,但至少需要两个,一个用来处理common数据,一个用来处理special的数据。注意:当需要增加多个Worker来消费同一类数据时,需要保持Queue名字一致,比如上面的Common数据。

 

订阅发布模式

  再来看“事件分发”的场景,架构如下图所示,使用event name/id来作为RabbitMQ的routing key的名字。Event Processor 01对event 01 和event 02感兴趣,则在启动Consumer Worker时,将自己的Queue绑定到这两个routing key上即可,其他Event Processor也是如此,这样便完成了事件的订阅。当有事件发布时,消息便会按照event name/id被投递到对应的Queue中。 

  由此可见,在不同的应用中,变化的只是routing_key与Consumer Queue的绑定关系,在充分理解RabbitMQ的核心概念后处理这些应该也是得心应手了。

 

出处:http://blog.csdn.net/zwgdft/article/details/53561277

分类: 设计模式
 
本文转自左正博客园博客,原文链接: http://www.cnblogs.com/soundcode/p/7156641.html,如需转载请自行联系原作者
 
相关实践学习
快速体验阿里云云消息队列RocketMQ版
本实验将带您快速体验使用云消息队列RocketMQ版Serverless系列实例进行获取接入点、创建Topic、创建订阅组、收发消息、查看消息轨迹和仪表盘。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
消息中间件 存储 NoSQL
RocketMQ实战—6.生产优化及运维方案
本文围绕RocketMQ集群的使用与优化,详细探讨了六个关键问题。首先,介绍了如何通过ACL配置实现RocketMQ集群的权限控制,防止不同团队间误用Topic。其次,讲解了消息轨迹功能的开启与追踪流程,帮助定位和排查问题。接着,分析了百万消息积压的处理方法,包括直接丢弃、扩容消费者或通过新Topic间接扩容等策略。此外,提出了针对RocketMQ集群崩溃的金融级高可用方案,确保消息不丢失。同时,讨论了为RocketMQ增加限流功能的重要性及实现方式,以提升系统稳定性。最后,分享了从Kafka迁移到RocketMQ的双写双读方案,确保数据一致性与平稳过渡。
|
消息中间件 监控 数据挖掘
【有奖实践】轻量消息队列(原 MNS)订阅 OSS 事件实时处理文件变动
当你需要对对象存储 OSS(Object Storage Service)中的文件变动进行实时处理、同步、监听、业务触发、日志记录等操作时,你可以通过设置 OSS 的事件通知规则,自定义关注的文件,并将 OSS 事件推送到轻量消息队列(原 MNS)的队列或主题中,开发者的服务即可及时收到相关通知,并通过消费消息进行后续的业务处理。
406 102
|
消息中间件
RabbitMQ的 RPC 消息模式你会了吗?
【9月更文挑战第11天】RabbitMQ 的 RPC(远程过程调用)消息模式允许客户端向服务器发送请求并接收响应。其基本原理包括:1) 客户端发送请求,创建回调队列并设置关联标识符;2) 服务器接收请求并发送响应至回调队列;3) 客户端根据关联标识符接收并匹配响应。实现步骤涵盖客户端和服务器的连接、信道创建及请求处理。注意事项包括关联标识符唯一性、回调队列管理、错误处理及性能考虑。RPC 模式适用于构建可靠的分布式应用程序,但需根据需求调整优化。
584 3
|
消息中间件 Java 测试技术
RocketMQ实战—7.生产集群部署和生产参数
本文详细介绍了RocketMQ生产集群的部署与调优过程,包括集群规划、环境搭建、参数配置和优化策略。
RocketMQ实战—7.生产集群部署和生产参数
|
消息中间件 对象存储
轻量消息队列(原 MNS)订阅 OSS 事件实践
使用轻量消息队列订阅OSS事件,实时处理文件变动,赢取ins风U型枕(限量500个)。访问活动页面,完成实操并上传截图即可参与领奖。活动时间:即日起至2025年2月28日16:00。奖品数量有限,先到先得,快来报名吧!
429 2
|
消息中间件 网络协议 RocketMQ
RocketMQ Controller 模式 始终更新成本机ip
ontrollerAddr=192.168.24.241:8878 但是日志输出Update controller leader address to 127.0.0.1:8878。导致访问失败
454 4
|
消息中间件 测试技术
通过轻量消息队列(原MNS)主题HTTP订阅+ARMS实现自定义数据多渠道告警
轻量消息队列(原MNS)以其简单队列模型、轻量化协议及按量后付费模式,成为阿里云产品间消息传输首选。本文通过创建主题、订阅、配置告警集成等步骤,展示了该产品在实际应用中的部分功能,确保消息的可靠传输。
477 2
|
消息中间件 存储 负载均衡
我服了,RocketMQ消费者负载均衡内核是这样设计的
文章为理解RocketMQ的负载均衡机制提供了深入的技术洞察,并对如何在实际应用中扩展和定制负载均衡策略提供了有价值的见解。
我服了,RocketMQ消费者负载均衡内核是这样设计的
|
消息中间件 应用服务中间件 网络安全
rabbitMQ镜像模式搭建
rabbitMQ镜像模式搭建
|
消息中间件 开发者
【RabbitMQ深度解析】Topic交换器与模式匹配:掌握消息路由的艺术!
【8月更文挑战第24天】在消息队列(MQ)体系中,交换器作为核心组件之一负责消息路由。特别是`topic`类型的交换器,它通过模式匹配实现消息的精准分发,适用于发布-订阅模式。不同于直接交换器和扇形交换器,`topic`交换器支持更复杂的路由策略,通过带有通配符(如 * 和 #)的模式字符串来定义队列与交换器间的绑定关系。
622 2