Title

一、RabbitMQ安装

#Centos7 安装
 
#注意/etc/hosts文件 ip和主机名对应
wget https://github.com/rabbitmq/rabbitmq-server/releases/download/rabbitmq_v3_6_10/rabbitmq-server-3.6.10-1.el7.noarch.rpm
yum install epel-release -y
yum install rabbitmq-server-3.6.10-1.el7.noarch.rpm
rabbitmq-plugins enable rabbitmq_management
cp /usr/share/doc/rabbitmq-server-3.6.10/rabbitmq.config.example /etc/rabbitmq/rabbitmq.config
systemctl restart rabbitmq-server
systemctl status rabbitmq-server
 
#创建用户 授权
rabbitmqctl  add_user gmu gmu1592618
rabbitmqctl set_permissions -p / gmu ".*" ".*" ".*"

#启动服务
service rabbitmq-server start　　

进入cmd 安装pika

pip install pika

二、示例

1. 服务端和客户端一对一 

import pika

credentials = pika.PlainCredentials("gmu","gmu1592618")  #授权的账号 密码

connection = pika.BlockingConnection(pika.ConnectionParameters('45.77.203.27',5672,'/',credentials=credentials)) #建立socket

channel = connection.channel() #创建rabbit协议通道

channel.queue_declare('team1') #通过通道生成一个队列

channel.basic_publish(exchange='',
                      routing_key = 'team1',          #队列
                      body = 'hello I am first msg' #发送的消息
                      )
print(" [x] Sent 'hello I am first msg'")
connection.close()  #断开连接

import pika

credentials = pika.PlainCredentials('gmu','gmu1592618') #授权的账号和密码

connection = pika.BlockingConnection(pika.ConnectionParameters('45.77.203.27',5672,'/',credentials=credentials)) #建立socket

channel = connection.channel() #创建协议通道

channel.queue_declare('team1') #通过通道生成一个队列

def callback(ch,method,properties,body):
    print(ch)          #上面channel = connection.channel()对象
    print(method)      #除了服务端本身的数据，还带一些参数
    print(properties)  #属性
    print(body)        #byte数据

channel.basic_consume(callback,queue='team1',no_ack=True)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()  #阻塞

2.消息持久化

#- 开启一个服务端,两个客户端
#- 服务端向队列中存放一个值,一客户端从队列中取到数据,在睡20秒期间中断,表示出错,它不会报告给服务端
#- 这时队列中为零,另一客户端也不会取到值
# no_ack=True 表示客户端处理完了不需要向服务端确认消息

import pika

credentials = pika.PlainCredentials("gmu","gmu1592618")  #授权的账号 密码

connection = pika.BlockingConnection(pika.ConnectionParameters('45.77.203.27',5672,'/',credentials=credentials)) #建立socket

channel = connection.channel() #创建rabbit协议通道

channel.queue_declare('team1') #通过通道生成一个队列

channel.basic_publish(exchange='',
                      routing_key = 'team1',          #队列
                      body = 'hello I am first msg' #发送的消息
                      )
print(" [x] Sent 'hello I am first msg'")
connection.close()  #断开连接

import pika,time

credentials = pika.PlainCredentials('gmu','gmu1592618') #授权的账号和密码

connection = pika.BlockingConnection(pika.ConnectionParameters('45.77.203.27',5672,'/',credentials=credentials)) #建立socket

channel = connection.channel() #创建协议通道

channel.queue_declare('team1') #通过通道生成一个队列

def callback(ch,method,properties,body):
    print("received msg...start process",body)
    time.sleep(10)
    print("end process...")

channel.basic_consume(callback,queue='team1',no_ack=True)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()  #阻塞

3.队列持久化

#队列持久化
 
channel.queue_declare(queue='hello',durable=True) # ***
systemctl restart rabbitmq-server       #重启服务发现hello队列还在,但是消息不在
rabbitmqctl list_queues
    #team1
 
 
#队列和消息持久化
channel.queue_declare(queue='hello',durable=True)
 
properties=pika.BasicProperties(
    delivery_mode=2,  # make message persistent ***
),
systemctl restart rabbitmq-server       #重启服务发现队列和消息都还在
rabbitmqctl list_queues
    #team1 5

import pika

credentials = pika.PlainCredentials("gmu","gmu1592618")  #授权的账号 密码

connection = pika.BlockingConnection(pika.ConnectionParameters('45.77.203.27',5672,'/',credentials=credentials)) #建立socket

channel = connection.channel() #创建rabbit协议通道

channel.queue_declare('team1',durable=True) #通过通道生成一个队列,且队列持久化

channel.basic_publish(exchange='',
                      routing_key = 'team1',   #队列
                      properties=pika.BasicProperties(
                          delivery_mode=2,  # make message persistent 加参数2 消息持久化
                      ),
                      body = 'hello I am first msg' #发送的消息
                      )
print(" [x] Sent 'hello I am first msg'")
connection.close()  #断开连接

4.模拟客户端中断 观察服务端队列的数据会不会返回(会)　

#1. 生产者端发消息时，加参数 消息持久化
    properties=pika.BasicProperties(
        delivery_mode=2,  # make message persistent
    ),
#2. 消费者端，消息处理完毕时，发送确认包  
    ch.basic_ack(delivery_tag=method.delivery_tag)
 
    channel.basic_consume(callback, #取到消息后，调用callback 函数
      queue='task1',)
      #no_ack=True) #消息处理后，不向rabbit-server确认消息已消费完毕
#- 开启一个服务端,两个客户端
#- 服务端向队列中存放一个值,一客户端从队列中取到数据,在睡20秒期间中断,表示出错,它会报给服务端,服务端队列还有值
#- 这时启动另一客户端还可以取到值　

import pika

credentials = pika.PlainCredentials("gmu","gmu1592618")  #授权的账号 密码

connection = pika.BlockingConnection(pika.ConnectionParameters('45.77.203.27',5672,'/',credentials=credentials)) #建立socket

channel = connection.channel() #创建rabbit协议通道

channel.queue_declare('team1') #通过通道生成一个队列

channel.basic_publish(exchange='',
                      routing_key = 'team1',   #队列
                      properties=pika.BasicProperties(
                          delivery_mode=2,  # make message persistent 加参数2 消息持久化
                      ),
                      body = 'hello I am first msg' #发送的消息
                      )
print(" [x] Sent 'hello I am first msg'")
connection.close()  #断开连接

　

import pika,time

credentials = pika.PlainCredentials('gmu','gmu1592618') #授权的账号和密码

connection = pika.BlockingConnection(pika.ConnectionParameters('45.77.203.27',5672,'/',credentials=credentials)) #建立socket

channel = connection.channel() #创建协议通道

channel.queue_declare('team1') #通过通道生成一个队列

def callback(ch,method,properties,body):
    print("received msg...start process",body)
    time.sleep(10)
    print("end process...")
    ch.basic_ack(delivery_tag=method.delivery_tag) #消费者端，消息处理完毕时，发送确认包  

channel.basic_consume(callback,queue='team1')

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()  #阻塞

三、广播、组播、规则传播

1.广播 fanout

#服务端：
  - 不需要申明队列
#客户端：
  - 每个客户端都需要申明一个队列,自动设置队列名称,收听广播,当收听完后queue删除
  - 把队列绑定到exchange上
#注意：客户端先打开,服务端再打开,客户端会收到消息
  
#应用：
  - 微博粉丝在线,博主发消息,粉丝可以收到
 
#如果服务端先启动向exchange发消息,这时客户端没有启动,没有队列保存数据(exchange不负责保存数据)
#这时数据会丢,队列中没有数据
#exchange只负责转发

import pika,sys

credentials = pika.PlainCredentials("gmu","gmu1592618")  #授权的账号 密码

connection = pika.BlockingConnection(pika.ConnectionParameters('45.77.203.27',5672,'/',credentials=credentials)) #建立socket

channel = connection.channel() #创建rabbit协议通道

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

message = ' '.join(sys.argv[1:]) or "info: hello I am first msg"

channel.basic_publish(exchange='logs',
                      routing_key = '',   #队列
                      properties=pika.BasicProperties(
                          delivery_mode=2,  # make message persistent 加参数2 消息持久化
                      ),
                      body = message #发送的消息
                      )
print(" [x] Sent '%s'"%message)
connection.close()  #断开连接

import pika,time

credentials = pika.PlainCredentials('gmu','gmu1592618') #授权的账号和密码

connection = pika.BlockingConnection(pika.ConnectionParameters('45.77.203.27',5672,'/',credentials=credentials)) #建立socket

channel = connection.channel() #创建协议通道
channel.exchange_declare(exchange='logs', type='fanout')

queue_obj = channel.queue_declare(exclusive=True) #不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
queue_name = queue_obj.method.queue

channel.queue_bind(exchange='logs',queue=queue_name) #绑定队列到Exchange

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() #阻塞

2.组播 direct

#客户端一：
    - python3 consumer.py info
#客户端二：
    - python3 consumer.py  error
#客户端三：
    - python3 consumer.py  warning
#客户端四：
    - python3 consumer.py  warning error info
#服务端：
    - python3 producer.py  warning

import pika,sys

credentials = pika.PlainCredentials("gmu","gmu1592618")  #授权的账号 密码

connection = pika.BlockingConnection(pika.ConnectionParameters('45.77.203.27',5672,'/',credentials=credentials)) #建立socket

channel = connection.channel() #创建rabbit协议通道

channel.exchange_declare(exchange='direct_logs',type='direct')

severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or "info: hello I am first msg"

channel.basic_publish(exchange='direct_logs',
                      routing_key = severity,   #队列
                      body = message #发送的消息
                      )
print(" Send %r:%r" % (severity, message))
connection.close()  #断开连接

import pika,sys

credentials = pika.PlainCredentials('gmu','gmu1592618') #授权的账号和密码

connection = pika.BlockingConnection(pika.ConnectionParameters('45.77.203.27',5672,'/',credentials=credentials)) #建立socket

channel = connection.channel() #创建协议通道
channel.exchange_declare(exchange='direct_logs',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, properties, body):
    print(" [x] %r:%r" % (method.routing_key, body))

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

 3.规则传播 topic

#客户端一：#以django 结尾
    - python3 consumer.py *.django
	
#客户端二：#包含mysql.error
    - python3 consumer.py mysql.error
	
#客户端三：#以mysql.开头
    - python3 consumer.py mysql.*
 
#服务端：
    - python3 producer.py  #匹配相应的客户端

import pika,sys

credentials = pika.PlainCredentials("gmu","gmu1592618")  #授权的账号 密码

connection = pika.BlockingConnection(pika.ConnectionParameters('45.77.203.27',5672,'/',credentials=credentials)) #建立socket

channel = connection.channel() #创建rabbit协议通道

channel.exchange_declare(exchange='topic_logs',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()

import pika,sys,time

credentials = pika.PlainCredentials('gmu','gmu1592618') #授权的账号和密码

connection = pika.BlockingConnection(pika.ConnectionParameters('45.77.203.27',5672,'/',credentials=credentials)) #建立socket

channel = connection.channel() #创建协议通道
channel.exchange_declare(exchange='topic_logs',type='topic')

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

binding_keys = sys.argv[1:]
if not binding_keys:
    print(sys.argv[1:])
    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) #绑定队列到Exchange

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()

Remote procedure call (RPC) 远程过程调用

　　从上边所有的例子中你有没有发现，上面的队列都是单向执行的，需要有发送端和接收端。如果远程的一台机器执行完毕再返回结果，那就实现不了了。如果让他执行完返回，这种模式叫什么呢？RPC（远程过程调用），snmp就是典型的RPC。

　　那RabbitMQ能不能返回呢，怎么返回呢？可以让机器既是发送端又是接收端。但是接收端返回消息怎么返回？可以发送到发过来的queue里么？答案当然是不可以，如果还是存在原先的队列就会直接陷入死循环！所以返回时，需要让消息内部指定再建立一个队列queue，把结果发送新的queue里。

　　同时，为了 "执行命令端 "返回的queue不写死，在 "发送命令端" 给 "执行命令端 "发指令的的时候，同时带一条消息说，你结果返回给哪个queue

　　在执行多个消息任务的时候，怎么区分判断哪个消息是先执行呢？答案就是，在发任务时，再额外提交一个唯一标识符。
task1，task2异步执行，但是返回的顺序是不固定的，为了区分是谁执行完的，在发送的任务添加唯一标识符，这样取回的时候就能区分。
设置一个异步RPC
　　声明一个队列reply_to，作为返回消息结果的队列
　　发送消息队列，消息中带唯一标识uid
　　监听reply_to队列，直到有结果
在类中声明监听 

 发送命令客户端：

import pika,sys,uuid # 发送命令端
import threading,random
#单独起一个线程，只负责发指令
# 1.声明一个队列，作为reply_to返回消息结果的队列
# 2.  发消息到队列，消息里带一个唯一标识符uid，reply_to
# 3.  监听reply_to 的队列，直到有结果
class CMDRpcClient(object):
    def __init__(self):
        credentials = pika.PlainCredentials('gmu','gmu1592618') #授权的账号和密码
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('45.77.203.27',5672,'/',credentials=credentials)) #建立socket
        # self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost',port=5672))

        self.channel = self.connection.channel() #创建协议通道

        result = self.channel.queue_declare(exclusive=True) # 声明一个队列，作为reply_to返回消息结果的队列
        self.callback_queue = result.method.queue  # 命令的执行结果的queue

        # 声明要监听callback_queue
        self.channel.basic_consume(self.on_response, no_ack=True,
                                   queue=self.callback_queue) #声明监听callback_queue队列，收到消息后调用函数on_response

    def on_response(self, ch, method, props, body):
        """
        收到服务器端命令结果后执行这个函数
        :param ch:
        :param method:
        :param props:执行命令端返回的
        :param body:
        :return:
        """
        if self.corr_id == props.correlation_id: #如果发送命令端的uuid对于执行命令端返回的uuid
            self.response = body.decode("gbk")  # 把执行命令端的执行结果赋值给Response

    def call(self, cmd):
        self.response = None
        self.corr_id = str(uuid.uuid4())  # 唯一标识符号
        self.channel.basic_publish(exchange='',
                                   routing_key='rpc_queue',
                                   properties=pika.BasicProperties(
                                       reply_to=self.callback_queue,
                                       correlation_id=self.corr_id,
                                   ),
                                   body=str(cmd))

        while self.response is None:              #检测队列的时候可以同时检测命令队列q中有没有值，有的话执行run函数
            self.connection.process_data_events()  # 检测监听的队列里有没有新消息，如果有，收，如果没有，返回None
            # 检测有没有要发送的新指令
        return self.response

class MY_THREAD():
    def __init__(self):
        self.info = {}
        self.help_info = '''        
                run "df -h" 
                --- ------- 
                运行  指令             
                check_task_all      #查看任务列表
                check_task  25413   #查看具体id任务信息，过后删除
                helps               #查看指令帮助
                '''
    def start(self):
        self.helps()
        while True:
            inp = input(">>:").strip()  # run "df -h"  | check_task_all | check_task  25413 | helps |
            if not inp: continue  # 如果为空重新输入
            t1 = threading.Thread(target=self.hasattr_func, args=(inp,))  # 创建新线程 调用反射函数
            t1.start()  # 开始线程

    def hasattr_func(self,inp):
        cmd = inp.split()[0]  # 取命令开头 按空格分割
        if cmd == 'helps':
            self.helps()
        if len(inp.split()) == 'b' and cmd != 'check_task_all':  # 输入 b 结束函数执行
            return
        if hasattr(self, cmd):  # 是否存在
            func = getattr(self, cmd)  # 调用
            rer = func(inp)  # 执行  传入命令行 run "df -h"  | check_task_all | check_task  25413 | helps |
            if rer is not None:
                for i in rer:
                    print("任务id:%s" % i)

    #执行查询命令
    def run(self,inp):
        cmd = inp.split('"')[1]  #获取命令cmd:  df -h
        tak_id = random.randint(10000, 99999)  # 任务ID生成
        obj = CMDRpcClient()  # 生成连接类
        response = obj.call(cmd)  # 传入查询命令，得到命令的执行结果
        self.info[tak_id] = [cmd,response]  # 写入字典 tak_id{  命令 命令执行结果}
        return self.info

    #查看所有任务
    def check_task_all(self,inp):
        for task_id,value in self.info:
            print('任务ID %s, 命令 %s,'%(task_id,value[0]))

    #查看指定任务
    def check_task(self,inp):
        task_id = inp.split()[1] #拿到要查询的任务ID
        del_id = ''
        for k,value in self.info:
            if k == task_id:
                del_id = k
                print('命令 %s, 执行结果 %s'%(value[0],value[1]))
        del self.info[del_id] #查询过后删掉

    # 打印帮助信息
    def helps(self):
        print(self.help_info)

执行命令服务端：

import pika,sys,subprocess #执行命令端
import threading
# #1. 定义run_cmd函数
# #2. 声明接收指令的队列名rpc_queue
# #3. 开始监听队列，收到消息后 调用run_cmd函数
# #4. 把run_cmd执行结果，发送回客户端指定的reply_to 队列

class RPC_SERVER():
    def __init__(self):
        self.credentials = pika.PlainCredentials("gmu", "gmu1592618")  # 授权的账号 密码
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('45.77.203.27', 5672, '/', credentials=self.credentials))  # 建立socket
        # self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost',port=5672))
        self.channel = self.connection.channel()  # 创建rabbit协议通道
        self.channel.queue_declare(queue='rpc_queue')  # 声明一个队列

    # 执行命令并返回结果
    def run_cmd(self,cmd):
        cmd_obj = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        result = cmd_obj.stdout.read() + cmd_obj.stderr.read()
        return result

    #回调函数
    def on_request(self,ch, method, props, body):
        cmd = body.decode("utf-8")

        print(" [.] run (%s)" % cmd)
        response = self.run_cmd(cmd)  # 命令的执行结果

        ch.basic_publish(exchange='',
                         routing_key=props.reply_to,  # reply_to 是发送命令端传过来的队列
                         properties=pika.BasicProperties(correlation_id=props.correlation_id),
                         # 把发送命令端的uuid赋给自己correlation_id
                         body=response)  # 把命令的执行结果放回队列中
        ch.basic_ack(delivery_tag=method.delivery_tag)  # 明确的告诉客户端消息被处理了

    # 每当监听到队列中有命令消息时 生成一个线程执行on_request函数，拿到结果并放回队列当中
    def My_thread(self,ch, method, props, body):
        t1 = threading.Thread(target=self.on_request, args=(ch, method, props, body))
        t1.start()

    def start(self):
        self.channel.basic_consume(self.My_thread, queue='rpc_queue')  # 开始监听队列，队列中有值时 启动一个线程
        print(" [x] Awaiting RPC requests")
        self.channel.start_consuming()  # 阻塞监听