标签：localhost channel snmp alt 使用 ffffff 报告 system http
#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]\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()
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]...\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()
从上边所有的例子中你有没有发现，上面的队列都是单向执行的，需要有发送端和接收端。如果远程的一台机器执行完毕再返回结果，那就实现不了了。如果让他执行完返回，这种模式叫什么呢？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() # 阻塞监听
标签：localhost channel snmp alt 使用 ffffff 报告 system http
原文地址：http://www.cnblogs.com/guotianbao/p/7788694.html
