基于 RabbitMQ 的实时消息推送

博主新开公众号“不太灵光的程序员” ，
关注公众号，每日八点有干货推送

1 实现服务器端推送的几种方式

Web 应用都是基于 HTTP 协议的请求/响应模式，无法像 TCP 协议那样保持长连接，因此Web 应用就很难像手机那样实现实时的消息推送。就目前来看，Web 应用的消息推送方式主要有以下几种：

1.1 Ajax短轮询

Ajax 轮询主要通过页面端的 JS定时异步刷新任务来实现数据的加载，但这种方式实时效果较差，而且对服务端的压力也较大。

1.2 Ajax长轮询

长轮询主要也是通过 Ajax 机制，但区别于传统的Ajax 应用，长轮询的服务器端会在没有数据时阻塞请求直到有新的数据产生或者请求超时才返回，之后客户端再重新建立连接获取数据，具体实现方式见图 1 所示。但长轮询服务端会长时间地占用资源，如果消息频繁发送的话会给服务端带来较大的压力。

1.3 WebSocket 双向通信

WebSocket 是 HTML5 中一种新的通信协议，能够实现浏览器与服务器之间全双工通信。如果浏览器和服务端都支持 WebSocket 协议的话，该方式实现的消息推送无疑是最高效、简洁的。并且最新版本的 IE、Firefox、Chrome 等浏览器都已经支持 WebSocket 协议，Apache Tomcat 7.0.27 以后的版本也开始支持 WebSocket。

2 RabbitMQ简介

AMQP，即 AdvancedMessage Queuing Protocol，高级消息队列协议是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP 的主要特征是面向消息、队列和路由，可靠且安全。RabbitMQ 是一个开源的 AMQP 实现，服务器端用 Erlang 语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP 等，支持 Ajax。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。RabbitMQ 中有一些概念需要我们在使用前先搞清楚，主要包括以下几个：

2.1 ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。Connection是RabbitMQ的socket链接，它封装了socket协议相关部分逻辑。ConnectionFactory为Connection的制造工厂。
Channel是我们与RabbitMQ打交道的最重要的一个接口，我们大部分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

2.2 Queue

Queue（队列）是RabbitMQ的内部对象，用于存储消息，用下图表示。

RabbitMQ中的消息都只能存储在Queue中，生产者（下图中的P）生产消息并最终投递到Queue中，消费者（下图中的C）可以从Queue中获取消息并消费。

多个消费者可以订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。

2.3 Message acknowledgment

在实际应用中，可能会发生消费者收到Queue中的消息，但没有处理完成就宕机（或出现其他意外）的情况，这种情况下就可能会导致消息丢失。为了避免这种情况发生，我们可以要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message acknowledgment）后才将该消息从Queue中移除；如果RabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开，则RabbitMQ会将该消息发送给其他消费者（如果存在多个消费者）进行处理。这里不存在timeout概念，一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者，除非它的RabbitMQ连接断开。
这里会产生另外一个问题，如果我们的开发人员在处理完业务逻辑后，忘记发送回执给RabbitMQ，这将会导致严重的bug——Queue中堆积的消息会越来越多；消费者重启后会重复消费这些消息并重复执行业务逻辑…

2.4 Message durability

如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。但依然解决不了小概率丢失事件的发生（比如RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），如果我们需要对这种小概率事件也要管理起来，那么我们要用到事务。由于这里仅为RabbitMQ的简单介绍，所以这里将不讲解RabbitMQ相关的事务。

2.5 Prefetch count

前面我们讲到如果有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时如果每个消息的处理时间不同，就有可能会导致某些消费者一直在忙，而另外一些消费者很快就处理完手头工作并一直空闲的情况。我们可以通过设置prefetchCount来限制Queue每次发送给每个消费者的消息数，比如我们设置prefetchCount=1，则Queue每次给每个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。

2.6 Exchange

在上一节我们看到生产者将消息投递到Queue中，实际上这在RabbitMQ中这种事情永远都不会发生。实际的情况是，生产者将消息发送到Exchange（交换器，下图中的X），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。

Exchange是按照什么逻辑将消息路由到Queue的？这个将在Binding一节介绍。
RabbitMQ中的Exchange有四种类型，不同的类型有着不同的路由策略，这将在Exchange Types一节介绍。

2.7 Routing Key

生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，而这个routing key需要与Exchange Type及binding key联合使用才能最终生效。
在Exchange Type与binding key固定的情况下（在正常使用时一般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。
RabbitMQ为routing key设定的长度限制为255 bytes。

2.8 Binding

RabbitMQ中通过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。

2.9 Binding key

在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key；消费者将消息发送给Exchange时，一般会指定一个routing key；当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。这个将在Exchange Types章节会列举实际的例子加以说明。
在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的binding key。
binding key 并不是在所有情况下都生效，它依赖于Exchange Type，比如fanout类型的Exchange就会无视binding key，而是将消息路由到所有绑定到该Exchange的Queue。

2.10 Exchange Types

RabbitMQ常用的Exchange Type有fanout、direct、topic、headers这四种（AMQP规范里还提到两种Exchange Type，分别为system与自定义，这里不予以描述），下面分别进行介绍。

2.11 Fanout

fanout类型的Exchange路由规则非常简单，它会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中。

上图中，生产者（P）发送到Exchange（X）的所有消息都会路由到图中的两个Queue，并最终被两个消费者（C1与C2）消费。

2.12 Direct

direct类型的Exchange路由规则也很简单，它会把消息路由到那些binding key与routing key完全匹配的Queue中。

以上图的配置为例，我们以routingKey=”error”发送消息到Exchange，则消息会路由到Queue1（amqp.gen-S9b…，这是由RabbitMQ自动生成的Queue名称）和Queue2（amqp.gen-Agl…）；如果我们以routingKey=”info”或routingKey=”warning”来发送消息，则消息只会路由到Queue2。如果我们以其他routingKey发送消息，则消息不会路由到这两个Queue中。

2.13 Topic

前面讲到direct类型的Exchange路由规则是完全匹配binding key与routing key，但这种严格的匹配方式在很多情况下不能满足实际业务需求。topic类型的Exchange在匹配规则上进行了扩展，它与direct类型的Exchage相似，也是将消息路由到binding key与routing key相匹配的Queue中，但这里的匹配规则有些不同，它约定：

routing key为一个句点号“. ”分隔的字符串（我们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”

binding key与routing key一样也是句点号“. ”分隔的字符串

binding key中可以存在两种特殊字符“”与“#”，用于做模糊匹配，其中“”用于匹配一个单词，“#”用于匹配多个单词（可以是零个）

以上图中的配置为例，routingKey=”quick.orange.rabbit”的消息会同时路由到Q1与Q2，routingKey=”lazy.orange.fox”的消息会路由到Q1，routingKey=”lazy.brown.fox”的消息会路由到Q2，routingKey=”lazy.pink.rabbit”的消息会路由到Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配）；routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的消息将会被丢弃，因为它们没有匹配任何bindingKey。

2.14 headers

headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。
在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。
该类型的Exchange没有用到过（不过也应该很有用武之地），所以不做介绍。

2.15 RPC

MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）。
但实际的应用场景中，我们很可能需要一些同步处理，需要同步等待服务端将我的消息处理完成后再进行下一步处理。这相当于RPC（Remote Procedure Call，远程过程调用）。在RabbitMQ中也支持RPC。

RabbitMQ中实现RPC的机制是：

客户端发送请求（消息）时，在消息的属性（MessageProperties，在AMQP协议中定义了14中properties，这些属性会随着消息一起发送）中设置两个值replyTo（一个Queue名称，用于告诉服务器处理完成后将通知我的消息发送到这个Queue中）和correlationId（此次请求的标识号，服务器处理完成后需要将此属性返还，客户端将根据这个id了解哪条请求被成功执行了或执行失败）

服务器端收到消息并处理

服务器端处理完消息后，将生成一条应答消息到replyTo指定的Queue，同时带上correlationId属性

客户端之前已订阅replyTo指定的Queue，从中收到服务器的应答消息后，根据其中的correlationId属性分析哪条请求被执行了，根据执行结果进行后续业务处理

3 安装RabbitMQ服务

3.1 Windows

1. 下载Erlang：

http://www.erlang.org/downloads

win64: 下载otp_win64_19.2.exe

win32: 下载otp_win32_19.2.exe

2. 下载rabbitMQ-server：

http://www.rabbitmq.com/install-windows.html

网盘地址：

http://pan.baidu.com/s/1i5yv8Zr

http://pan.baidu.com/s/1gfwxI8b

3. rabbitmq 安装后默认会启动

查看 rabbitmq 状态

在安装目录下的 sbin>

rabbitmqctl.bat status

默认启动的是不启动管理模块的，不能登录http://localhost:15672

4. 启动管理模块

修改环境变量

添加系统环境变量ERLANG_HOME，值为安装目录.
修改系统环境变量Path,在PATH变量中添加“%ERL_HOME%\bin”

ERLANG_HOME会默认创建环境变量不配置 rabbitmq 也可以正常启动，但我当时管理模块一直起不来

rabbitmq-service.batstop

3.rabbitmqctl start_app

4.rabbitmq-plugins enable rabbitmq_management

5.rabbitmq-service.bat start

登录http://localhost:15672/guest guest

3.2 Linux

本文提出的方案是基于 RabbitMQ 消息服务器，因此最开始需要安装 RabbitMQ 服务及相关插件。RabbitMQ 是基于 Erlang 语言开发的，所以首先必须安装 Erlang 运行时环境。下面以 CentOS6.5 64 位服务器为例，讲述整个服务的安装过程：

1. 下载 erlang-R15B-02.1.el5.x86_64.rpm 并安装

rpm -ivh erlang-R15B-02.1.el5.x86_64.rpm

2. 下载 rabbitmq-server-3.2.1-1.noarch.rpm 并安装

rpm -ivh rabbitmq-server-3.2.1-1.noarch.rpm

3. 启用相关插件

rabbitmq-plugins enable rabbitmq_management rabbitmq_web_stomprabbitmq_stomp

4. 重启 RabbitMQ 服务

service rabbitmq-server restart

5. 验证是否安装成功

此时我们可能通过 Web 浏览器来查看RabbitMQ 的运行状态，浏览器中输入 http://{server_ip}:15672，用 guest/guest 默认的用户和密码登录后即可查看 RabbitMQ 的运行状态。

4 基于 RabbitMQ 的实时消息推送

RabbitMQ 有很多第三方插件，可以在AMQP 协议基础上做出许多扩展的应用。Web STOMP 插件就是基于 AMQP 之上的 STOMP 文本协议插件，利用 WebSocket 能够轻松实现浏览器和服务器之间的实时消息传递。

4.1 轮询消费模式

此模式下，发送队列的一方把消息存入mq的指定队列后，若有消费者端联入相应队列，即会获取到消息，并且队列中的消息会被消费掉。

若有多个消费端同时连接着队列，则会已轮询的方式将队列中的消息消费掉。

接下来是代码实例：

producer生产者

# !/usr/bin/env python
import pika
credentials = pika.PlainCredentials('admin','123456')
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.56.19',5672,'/',credentials))
channel = connection.channel()#声明queue
channel.queue_declare(queue='balance')# n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
channel.basic_publish(exchange='',routing_key='balance',body='Hello World!')
print(" [x] Sent 'Hello World!'")
connection.close()

发送过队列后，可在MQ服务器中查看队列状态

[root@localhost ~]# rabbitmqctl list_queues
Listing queues ...
hello    1

consumer消费者

# _*_coding:utf-8_*_
__author__ = 'Alex Li'
import pikacredentials = pika.PlainCredentials('admin','123456')
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.56.19',5672,'/',credentials))
channel = connection.channel()# You may ask why we declare the queue again ‒ we have already declared it in our previous code.
# We could avoid that if we were sure that the queue already exists. For example if send.py program
# was run before. But we're not yet sure which program to run first. In such cases it's a good
# practice to repeat declaring the queue in both programs.
channel.queue_declare(queue='balance')def callback(ch, method, properties, body):print(" [x] Received %r" % body)channel.basic_consume(callback,queue='balance',no_ack=True)print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

接收队列后，查看一下队列状态

[root@localhost ~]#  rabbitmqctl list_queues
Listing queues ...
hello    0

4.2 队列持久化

当rabbitMQ意外宕机时，可能会有持久化保存队列的需求（队列中的消息不消失）。

producer

# Cheng
# !/usr/bin/env python
import pikacredentials = pika.PlainCredentials('admin','123456')
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.56.19',5672,'/',credentials))
channel = connection.channel()# 声明queue
channel.queue_declare(queue='durable',durable=True)# n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
channel.basic_publish(exchange='',routing_key='durable',body='Hello cheng!',properties=pika.BasicProperties(delivery_mode=2,  # make message persistent))
print(" [x] Sent 'Hello cheng!'")
connection.close()

执行后查看队列，记下队列名字与队列中所含消息的数量

[root@localhost ~]# rabbitmqctl list_queues
Listing queues ...
durable    1
#重启rabbitmq
[root@localhost ~]# systemctl restart rabbitmq-server
#重启完毕后再次查看
[root@localhost ~]# rabbitmqctl list_queues
Listing queues ...
durable   #队列以及消息并未消失

执行消费者代码

cunsumer

# Cheng
# _*_coding:utf-8_*_
__author__ = 'Alex Li'
import pikacredentials = pika.PlainCredentials('admin','123456')
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.56.19',5672,'/',credentials))
channel = connection.channel()# You may ask why we declare the queue again ‒ we have already declared it in our previous code.
# We could avoid that if we were sure that the queue already exists. For example if send.py program
# was run before. But we're not yet sure which program to run first. In such cases it's a good
# practice to repeat declaring the queue in both programs.
channel.queue_declare(queue='durable',durable=True)def callback(ch, method, properties, body):print(" [x] Received %r" % body)ch.basic_ack(delivery_tag=method.delivery_tag)channel.basic_consume(callback,queue='durable',#no_ack=True)print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

可正确接收到信息。

再次查看队列的情况。

[root@localhost ~]# rabbitmqctl list_queues
Listing queues ...
durable    0

4.3 广播模式

当producer发送消息到队列后，所有的consumer都会收到消息，需要注意的是，此模式下producer与concerned之间的关系类似与广播电台与收音机，如果广播后收音机没有接受到，那么消息就会丢失。

建议先执行concerned

concerned

# _*_coding:utf-8_*_
__author__ = 'Alex Li'
import pikacredentials = pika.PlainCredentials('admin','123456')
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.56.19',5672,'/',credentials))
channel = connection.channel()channel.exchange_declare(exchange='Clogs',type='fanout')result = channel.queue_declare(exclusive=True)  # 不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
queue_name = result.method.queuechannel.queue_bind(exchange='Clogs',queue=queue_name)print(' [*] Waiting for logs. To exit press CTRL+C')def callback(ch, method, properties, body):print(" [x] %r" % body)channel.basic_consume(callback,queue=queue_name,no_ack=True)channel.start_consuming()

producer

import pika
import syscredentials = pika.PlainCredentials('admin','123456')
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.56.19',5672,'/',credentials))
channel = connection.channel()channel.exchange_declare(exchange='Clogs',type='fanout')message = ' '.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange='Clogs',routing_key='',body=message)
print(" [x] Sent %r" % message)
connection.close()