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的数据。

订阅发布模式

??再来看“事件分发”的场景,架构如下图所示,使用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

Bruce,2016/12/11

时间: 2024-12-14 18:44:30

RabbitMQ下的生产消费者模式与订阅发布模式的相关文章

【并发】9、借助redis 实现生产消费,消息订阅发布模式队列

这个就是一个消息可以被多次消费的范例了 其实这个实现的方式可以参考我之前的设计模式,观察者模式 https://www.cnblogs.com/cutter-point/p/5249780.html 不过有一点需要注意一下啊,这个消息发布的时候,好像是不支持字节数据的,里面好像会对字节进行转换,这样的结果就是导致我最后无法吧相应的字节转换成我之前序列化的对象 不知道是不是ObjectInputStream和ObjectOutputStream实现不是很好的原因,还是什么,反正反序列化的时候,有些

4 交换机-fanout(订阅发布模式)

目录 订阅发布模式 1.交换器(Exchange) 1.1.创建交换器 1.2 .推送消息到交换器 2.临时队列 3.绑定(bingdings) 5.代码例子 5.1.生产者代码示例 5.2.消费者代码示例 订阅发布模式 1.交换器(Exchange) 在Work Queue背后,其实是rabbitMQ把每条任务消息只发给一个消费者.本篇中我们将要研究如何把一条消息推送给多个消费者,这种模式被称为publish/subscribe(发布/订阅) RabbitMQ的消息发送模型核心思想是生产者不直

订阅发布模式

场景概述: 有时需要将多个应用程序集成到一个框架中,这些应用程序常见的基础通信方式包含总线模式.代理模式. 或者点对点模式.一些应用程序发送多种类型的消息,其他应用程序可能更关注这些消息类型的组合. 例如,在一个金融系统存在多个应用程序管理同一客户信息的情况,存在一个客户关系管理程序(CRM)掌握客户信息. 一种典型的情况:客户信息存在于其他系统中,且这些系统执行各自客户信息管理函数来处理客户信息. 当某个面向客户的应用程序生成更新客户信息的消息,例如客户地址的修改时,CRM和其他管理客户信息的

Publisher/Subscriber 订阅-发布模式

Publisher/Subscriber 订阅-发布模式 本博后续将陆续整理这些年做的一些预研demo,及一些前沿技术的研究,与大家共研技术,共同进步. 关于发布订阅有很多种实现方式,下面主要介绍WCF中的发布订阅,主要参考书籍<Programming WCF Services>,闲话不多说进入正题.使用传统的双工回调(例子 http://www.cnblogs.com/artech/archive/2007/03/02/661969.html)实现发布订阅模式存在许多缺陷,主要问题是,它会引

Spring基于事件驱动模型的订阅发布模式代码实例详解

代码下载地址:http://www.zuidaima.com/share/1791499571923968.htm 原文:Spring基于事件驱动模型的订阅发布模式代码实例详解 事件驱动模型简介 事件驱动模型也就是我们常说的观察者,或者发布-订阅模型:理解它的几个关键点: 首先是一种对象间的一对多的关系:最简单的如交通信号灯,信号灯是目标(一方),行人注视着信号灯(多方): 当目标发送改变(发布),观察者(订阅者)就可以接收到改变: 观察者如何处理(如行人如何走,是快走/慢走/不走,目标不会管的

AngularJS的简单订阅发布模式例子

控制器之间的交互方式广播 broadcast, 发射 emit 事件 类似于 js中的事件 , 可以自己定义事件 向上传递直到 document 在AngularJs中 向上传递直到 rootScope 观察者模式, 订阅发布模式 类似于js中的事件机制 订阅者.on('xx发布博客', function([内容]){ 通知我, 接收到博客的[内容] }) 发布者.emit('xxx发布博客', {内容}) 优点: 业务和实际触发者分离, 代码维护性相对好 缺点: 代码复杂性更高 Angular

Java里观察者模式(订阅发布模式)

创建主题(Subject)接口 创建订阅者(Observer)接口 实现主题 实现观察者 测试 总结 在公司开发项目,如果碰到一些在特定条件下触发某些逻辑操作的功能的实现基本上都是用的定时器 比如用户注册完后,发送邮件,为了防止邮件发送失败或者发送邮件比较耗时,一般也都是通过定时器去扫库里注册没有发邮件的用户数据 再比如一个订单,在改变状态后,要归档,这也是通过定时器来实现的,扫描订单的数据,通过判断状态来做相对应的处理 但这样处理的话,定时器就会越来越多,总觉得不太好 然后,从一些资讯网站上的

JS实现观察者模式(订阅/发布模式)

实现 /*  * js 观察者模式 又称 订阅/发布模式  * 通过创建"可观察"对象,当发生一个感兴趣的事件时可将该事件通告给  * 所有观察者,从而形成松耦合 */ // 通用的发布者 EventPublisher = Base.extend({ publish: function(data, type) { EventPublisher.publish(data, type); } }, { subscribers : {         any : []    // 事件类型:

Node中EventEmitter以及如何实现JavaScript中的订阅/发布模式

1.EventEmitter Node中很多模块都能够使用EventEmitter,有了EventEmitter才能方便的进行事件的监听.下面看一下Node.js中的EventEmitter如何使用. (1)基本使用 EventEmitter是对事件触发和事件监听功能的封装,在node.js中的event模块中,event模块只有一个对象就是EventEmitter,下面是一个最基本的使用方法: var EventEmitter = require('events').EventEmitter;