码迷,mamicode.com
首页 > 其他好文 > 详细

RabbitMQ

时间:2017-11-09 22:45:27      阅读:257      评论:0      收藏:0      [点我收藏+]

标签:localhost   channel   snmp   alt   使用   ffffff   报告   system   http   

一、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()  #断开连接
producer.py
技术分享
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()  #阻塞
consumer.py

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()  #断开连接
producer.py
技术分享
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()  #阻塞
consumer.py

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()  #断开连接
producer.py

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()  #断开连接
producer.py 
技术分享
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()  #阻塞
consumer.py

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

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()  #断开连接
producer.py
技术分享
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() #阻塞
consumer.py

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()  #断开连接
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() #创建协议通道
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]\n" % 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()
consumer.py

 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()
producer.py
技术分享
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]...\n" % 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()
consumer.py

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)
View Code

执行命令服务端:

技术分享
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()  # 阻塞监听
View Code

 

  

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

  

 

RabbitMQ

标签:localhost   channel   snmp   alt   使用   ffffff   报告   system   http   

原文地址:http://www.cnblogs.com/guotianbao/p/7788694.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!