python3之rabbitMQ

1、RabbitMQ介绍

  RabbitMQ是一个由erlang开发的AMQP（Advanced Message Queue ）的开源实现。AMQP 的出现其实也是应了广大人民群众的需求，虽然在同步消息通讯的世界里有很多公开标准（如 COBAR的 IIOP ，或者是 SOAP 等），但是在异步消息处理中却不是这样，只有大企业有一些商业实现（如微软的 MSMQ ，IBM 的 Websphere MQ 等），因此，在 2006 年的 6 月，Cisco 、Redhat、iMatix 等联合制定了 AMQP 的公开标准。

MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消 息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

消息队列技术是分布式应用间交换信息的一种技术；消息队列可驻留在内存或磁盘上,队列存储消息直到它们被应用程序读走；通过消息队列，应用程序可独立地执行--它们不需要知道彼此的位置、或在继续执行前不需要等待接收程序接收此消息。

MQ主要作用是接受和转发消息。你可以想想在生活中的一种场景：当你把信件的投进邮筒，邮递员肯定最终会将信件送给收件人。我们可以把MQ比作 邮局和邮递员。

MQ和邮局的主要区别是,它不处理消息,但是,它会接受数据、存储消息数据、转发消息。

2、安装RabbitMQ

linux上安装：

安装配置epel源

   $ rpm -ivh http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm 

安装erlang

   $ yum -y install erlang

安装RabbitMQ

   $ yum -y install rabbitmq-server

windows上安装：

(1)首先，您需要安装支持的 Windows 版Erlang。运行Erlang Windows安装程序。Erlang将出现在开始菜单中，设置erlang的环境变量（C:erl9.2in;），测试erlang是否安装正确：cmd-输入：erl，能看到eshell版本号，说明安装成功！

(2)下载rabbitMQ:(安装下一步完成即可)

https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.7.3/rabbitmq-server-windows-3.7.3.zip

(3)安装RabbitMQ-Plugins，这个相当于是一个管理界面，方便我们在浏览器界面查看RabbitMQ各个消息队列以及交换的工作情况，安装方法是：打开命令行cd进入rabbitmq的sbin目录（我的目录是：C： software rabbitmq rabbitmq_server-3.6.5 sbin），输入：rabbitmq-plugins enable rabbitmq_management命令，稍等会会发现出现plugins安装成功的提示，默认是安装6个插件。

插件安装完之后，在浏览器输入的http：//本地主机：15672进行验证

如果不能打开页面解决方法：首先在命令行输入：rabbitmq-service stop，接着输入rabbitmq-service remove，再接着输入rabbitmq-service install，接着输入rabbitmq-service start，最后重新输入rabbitmq-plugins enable rabbitmq_management试试，我是这样解决的。

创建用户名，密码，绑定角色：

RabbitMQ报错解决方法：

C:RabbitMQ Server
abbitmq_server-3.7.3sbin>rabbitmqctl.bat status
Status of node rabbit@DESKTOP-6JT7D2H ...
Error: unable to perform an operation on node 'rabbit@DESKTOP-6JT7D2H'. Please see diagnostics information and suggestions below.

Most common reasons for this are:

 * Target node is unreachable (e.g. due to hostname resolution, TCP connection or firewall issues)
 * CLI tool fails to authenticate with the server (e.g. due to CLI tool's Erlang cookie not matching that of the server)
 * Target node is not running

In addition to the diagnostics info below:

 * See the CLI, clustering and networking guides on http://rabbitmq.com/documentation.html to learn more
 * Consult server logs on node rabbit@DESKTOP-6JT7D2H

DIAGNOSTICS
===========

attempted to contact: ['rabbit@DESKTOP-6JT7D2H']

rabbit@DESKTOP-6JT7D2H:
  * connected to epmd (port 4369) on DESKTOP-6JT7D2H
  * epmd reports node 'rabbit' uses port 25672 for inter-node and CLI tool traffic
  * TCP connection succeeded but Erlang distribution failed

  * Authentication failed (rejected by the remote node), please check the Erlang cookie


Current node details:
 * node name: 'rabbitmqcli57@DESKTOP-6JT7D2H'
 * effective user's home directory: C:UsersAdministrator.DESKTOP-6JT7D2H
 * Erlang cookie hash: RmzKErjVZUcsMU8wSgBGbA==

解决方法：
将C:Users	racyclock.erlang.cookie 文件拷贝到C:WindowsSystem32configsystemprofile替换掉.erlang.cookie文件
重启rabbitMQ服务：net stop RabbitMQ && net start RabbitMQ

查看用户及用户角色：rabbitmqctl.bat list_users

C:RabbitMQ Server
abbitmq_server-3.7.3sbin>rabbitmqctl.bat list_users
Listing users ...
guest   [administrator]

新增用户：rabbitmqctl.bat add_user username password

C:RabbitMQ Server
abbitmq_server-3.7.3sbin>rabbitmqctl.bat add_user admin 123456
Adding user "admin" ...

C:RabbitMQ Server
abbitmq_server-3.7.3sbin>rabbitmqctl.bat list_users
Listing users ...
admin   []
guest   [administrator]

用户角色：

rabbitmq用户角色可分为五类：超级管理员, 监控者, 策略制定者, 普通管理者以及其他。

(1) 超级管理员(administrator)

可登陆管理控制台(启用management plugin的情况下)，可查看所有的信息，并且可以对用户，策略(policy)进行操作。

(2) 监控者(monitoring)

可登陆管理控制台(启用management plugin的情况下)，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等) 

(3) 策略制定者(policymaker)

可登陆管理控制台(启用management plugin的情况下), 同时可以对policy进行管理。

(4) 普通管理者(management)

仅可登陆管理控制台(启用management plugin的情况下)，无法看到节点信息，也无法对策略进行管理。

(5) 其他的

无法登陆管理控制台，通常就是普通的生产者和消费者。

授权用户为超级管理员：rabbitmqctl.bat set_user_tags admin administrator

C:RabbitMQ Server
abbitmq_server-3.7.3sbin>rabbitmqctl.bat set_user_tags admin administrator
Setting tags for user "admin" to [administrator] ...

C:RabbitMQ Server
abbitmq_server-3.7.3sbin>rabbitmqctl.bat list_users
Listing users ...
admin   [administrator]
guest   [administrator]

用户可以设置多个角色：rabbitmqctl.bat set_user_tags username tag1 tag2 ...

修改用户密码：rabbitmqctl change_password userName newPassword

C:RabbitMQ Server
abbitmq_server-3.7.3sbin>rabbitmqctl.bat change_password admin 888888
Changing password for user "admin" ...

删掉用户：rabbitmqctl.bat delete_user username

C:RabbitMQ Server
abbitmq_server-3.7.3sbin>rabbitmqctl.bat delete_user guest
Deleting user "guest" ...

C:RabbitMQ Server
abbitmq_server-3.7.3sbin>rabbitmqctl.bat list_users
Listing users ...
admin   [administrator]

权限相关命令为：

(1) 设置用户权限

rabbitmqctl  set_permissions  -p  VHostPath  User  ConfP  WriteP  ReadP

(2) 查看(指定hostpath)所有用户的权限信息

rabbitmqctl  list_permissions  [-p  VHostPath]

(3) 查看指定用户的权限信息

rabbitmqctl  list_user_permissions  User

(4)  清除用户的权限信息

rabbitmqctl  clear_permissions  [-p VHostPath]  User

3、python3使用pika python客户端

发出消息（生产者）:

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/2/22 15:51
# @Author  : Py.qi
# @File    : rabbitMQ_send_1.py
# @Software: PyCharm

import pika,sys
print('send....start....')
while True:
    inputso=input('soinsideto:')
    if inputso == 'quit':
        break
#与RabbitMQ服务器建立链接
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
#建立隧道
    channel = connection.channel()
#创建队列名称zhang
    channel.queue_declare(queue='zhang')
#发送信息：exchange指定交换，routing_key指定队列名，body指定消息内容    
　　channel.basic_publish(exchange='',routing_key='zhang',body=inputso)
#关闭链接
    connection.close()

接收者（消费者）：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/2/22 15:57
# @Author  : Py.qi
# @File    : rabbitMQ_rescv_1.py
# @Software: PyCharm

import pika
#创建链接
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
#建立隧道
channel = connection.channel()
#创建队列名，此处也可省略，在找不到队列时创建
channel.queue_declare(queue='zhang')
#使用回调函数callback来接收消息并打印消息
def callback(ch,method,properties,body):
    print('recived:',body)
#指定队列接收消息,callback接收消息，queue指定队列，no_ack不给发送者发送确认消息
channel.basic_consume(callback,queue='zhang',no_ack=True)
print('waiting for message,to exit press ctrl+c')
#持续接收消息,阻塞
channel.start_consuming()

 默认情况下，RabbitMQ会按顺序将每条消息发送到下一个使用者，每个消费者按顺序获得同样数量的消息，这种分配方式称为循环法。

（1）消息持久化：

当RabbitMQ退出或者崩溃时，它会忘记队列和消息，需要做两件事来确保消息不会丢失：我们需要将队列和消息标记为持久化。

在创建队列时指定durable为True来标记队列持久化：

channel.queue_declare(queue='hehe',durable=True)

将消息标记为持久化，通过提供值为2的delivery_mode属性

channel.basic_publish(exchange='',
                          routing_key='hehe',
                          body=inputso,
                          properties=pika.BasicProperties(
                              delivery_mode=2,  # make message persistent
                          ))

（2）消息公平分发

如果RabbitMQ只管按顺序把消息发送到每个消费者上，不考虑消费者的负载，很可能出现一个机器配置不高的消费者那里堆积很多消息处理不完，同时配置高的机器却很轻松。为了解决这个问题，可以在每个消费者端，配置perfetch=1，意识就是告诉RabbitMQ在我这个消费者当前消息还没处理完的时候就不要再发新消息了。

为了解决这个问题，我们可以使用basic.qos方法和 prefetch_count = 1设置。这告诉RabbitMQ一次不要向工作人员发送多于一条消息。或者换句话说，不要向工作人员发送新消息，直到它处理并确认了前一个消息。相反，它会将其分派给不是仍然忙碌的下一个工作人员。

channel.basic_qos（prefetch_count = 1）

完整代码：

生产者：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/2/22 15:51
# @Author  : Py.qi
# @File    : rabbitMQ_send_1.py
# @Software: PyCharm

import pika
print('send....start....')
while True:
    inputso=input('soinsideto:')
    if inputso == 'quit':
        break
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.146.129'))
    channel = connection.channel()
    channel.queue_declare(queue='hehe',durable=True)
    channel.basic_publish(exchange='',
                          routing_key='hehe',
                          body=inputso,
                          properties=pika.BasicProperties(
                              delivery_mode=2,  # make message persistent
                          ))
    connection.close()

消费者：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/2/22 15:57
# @Author  : Py.qi
# @File    : rabbitMQ_rescv_1.py
# @Software: PyCharm

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.146.129'))
channel = connection.channel()
channel.queue_declare(queue='hehe',durable=True)
def callback(ch,method,properties,body):
    print('recived:',body)
    ch.basic_ack(delivery_tag=method.delivery_tag)
channel.basic_qos(prefetch_count=1) #确认发送消息个数
channel.basic_consume(callback,queue='hehe',no_ack=True)
print('waiting for message,to exit press ctrl+c')
channel.start_consuming()

（3）消息发布与订阅

发布订阅和简单的消息队列区别在于，发布订阅会将消息发送给所有的订阅者，而消息队列中的数据被消费一次便消失。所以，RabbitMQ实现发布和订阅时，会为每一个订阅者创建一个队列，而发布者发布消息时，会将消息放置在所有相关队列中。

发布消息类似于广播效果，需要用到exchange，在定义exchange时指定类型来决定哪些queue符合条件，可以接收消息：

有几种可用的交换类型：direct, topic, headers 和fanout。我们将关注最后一个fanout。我们创建该类型的交换，并将其称为logs：

channel.exchange_declare(exchange='logs',
                         exchange_type='fanout')

fanout:所有bind到此exchange的queue都可以接收到消息

direct：通过routingKey和exchange决定的哪个唯一的queue可以接收消息

topic：所有符合routingKey（此时可以是一个表达式）的routingKey所bind的queue可以接收消息。

　　表达式符号：#代表一个或多个字符，*代表任何字符

fanout交换，它只是将收到的所有消息广播到它所知道的所有队列中。

要列出服务器上的交换，可以使用命令rabbitmqctl：

sudo rabbitmqctl list_exchanges

广播消息，fanout实例：

生产者：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/2/24 17:05
# @Author  : Py.qi
# @File    : rabbit_send_fanout.py
# @Software: PyCharm
import pika
import sys
connection=pika.BlockingConnection(pika.ConnectionParameters(host='192.168.146.129'))
channel=connection.channel()
#指定交换类型
channel.exchange_declare(exchange='logs',exchange_type='fanout')
#message = ' '.join(sys.argv[1:]) or 'info:hello world!'
for i in range(10):
    channel.basic_publish(exchange='logs',routing_key='',body=str(i))
    print('[x]sent %r'%i)
connection.close()

消费者：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/2/24 17:12
# @Author  : Py.qi
# @File    : rabbit_recv_fanout.py
# @Software: PyCharm

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='192.168.146.129'))
channel = connection.channel()
#指定交换类型
channel.exchange_declare(exchange='logs',
                         exchange_type='fanout')
#让服务器随机生成队列名，一旦消费者链接关闭，队列将被删除
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
#将队列绑定到交换上
channel.queue_bind(exchange='logs',
                   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()

接收消息时指定队列， exchange type = direct

RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据关键字判断应该将数据发送至指定队列。

生产者：发布消息

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/2/26 9:25
# @Author  : Py.qi
# @File    : rabbit_send_direct.py
# @Software: PyCharm

import pika
import sys

conn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.146.129'))
channel = conn.channel()
#绑定exchange类型为指定接收
channel.exchange_declare(exchange='direct_logs',exchange_type='direct')
#判断参数个数
severity=sys.argv[1] if len(sys.argv) > 1 else 'info'
#发送消息
message = ' '.join(sys.argv[2:]) or 'hello world'
#消息发送到exchange交换队列
channel.basic_publish(exchange='direct_logs',routing_key=severity,body=message)
print('[x]sent %r:%r'%(severity,message))
conn.close()

消费者：接收消息

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/2/26 9:54
# @Author  : Py.qi
# @File    : rabbit_recver_direct.py
# @Software: PyCharm

import pika,sys
conn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.146.129'))
channel = conn.channel()
channel.exchange_declare(exchange='direct_logs',exchange_type='direct')
#随机队列名，结束即删除
result = channel.queue_declare(exclusive=True)
queue_name=result.method.queue
#接收队列消息关键参数
severities=sys.argv[1:]
if not severities:
    sys.stderr.write('Usage:%s[info][warning][error]
'%sys.argv[0])
    sys.exit(1)
#循环将消息发送到有关键字队列中
for severity in severities:
    channel.queue_bind(exchange='direct_logs',
                       queue=queue_name,
                       routing_key=severity)
print('[*]waiting for logs,To exit press CTRL+C')
def callback(ch,method,properites,body):
    print('[x]%r:%r'%(method.routing_key,body))
channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=True)
channel.start_consuming()

模糊匹配：

exchange type = topic

在topic类型下，可以让队列绑定几个模糊的关键字，之后发送者将数据发送到exchange，exchange将传入”路由值“和 ”关键字“进行匹配，匹配成功，则将数据发送到指定队列。

#：表示可以匹配0个或多个单词

*：表示只能匹配一个单词

生产者：发送消息

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/2/26 10:42
# @Author  : Py.qi
# @File    : rabbit_send_topic.py
# @Software: PyCharm

import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='192.168.146.129'))
channel = connection.channel()
#指定exchange为模糊匹配topic
channel.exchange_declare(exchange='topic_logs',
                         exchange_type='topic')

routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='topic_logs',
                      routing_key=routing_key,
                      body=message)
print(" [x] Sent %r:%r" % (routing_key, message))
connection.close()

消费者：接收消息

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/2/26 10:43
# @Author  : Py.qi
# @File    : rabbit_recve_topic.py
# @Software: PyCharm
import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='192.168.146.129'))
channel = connection.channel()
#exchange模式topic
channel.exchange_declare(exchange='topic_logs',
                         exchange_type='topic')

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

binding_keys = sys.argv[1:]
if not binding_keys:
    sys.stderr.write("Usage: %s [binding_key]...
" % sys.argv[0])
    sys.exit(1)

for binding_key in binding_keys:
    channel.queue_bind(exchange='topic_logs',
                       queue=queue_name,
                       routing_key=binding_key)

print(' [*] Waiting for logs. To exit press CTRL+C')

def callback(ch, method, properties, body):
    print(" [x] %r:%r" % (method.routing_key, body))

channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=True)

channel.start_consuming()