python学习之rabbitmq

import pika
import randomconnection &＃61; pika.BlockingConnection(pika.ConnectionParameters(&＃39;localhost&＃39;)) #先打电话约下&＃xff0c;看能否找到人
channel &＃61; connection.channel() #确定路线
channel.queue_declare(queue&＃61;&＃39;task_queue&＃39;, durable&＃61;True) #确定交通工具&＃xff0c;而且交通工具的名称叫‘task_queue&＃39;,durable &＃61; True表示就是你到了北京以后交通工具依然存在number &＃61; random.randint(1, 1000)
message &＃61; &＃39;hello world:{num}&＃39;.format(num &＃61; number)channel.basic_publish(exchange&＃61;&＃39;&＃39;, #交换机&＃xff0c;此时没有交换机参与&＃xff0c;所以参数为空&＃xff0c;routing_key&＃61;&＃39;task_queue&＃39;, #交通工具的名称body&＃61;message, #要发送给的内容properties&＃61;pika.BasicProperties(delivery_mode&＃61;2,) #表示不管路通不通&＃xff0c;你携带的消息都不会因为外界情况而消失)
print(" [x] Sent %r" % (message,))
connection.close()

import pika
import timehostname &＃61; &＃39;localhost&＃39;
parameters &＃61; pika.ConnectionParameters(hostname)
connection &＃61; pika.BlockingConnection(parameters)channel &＃61; connection.channel()
channel.queue_declare(queue&＃61;&＃39;task_queue&＃39;, durable&＃61;True)def callback(ch, method, properties, body):print(" [x] Received %r" % (body,))# time.sleep(5)print(" [x] Done")ch.basic_ack(delivery_tag&＃61;method.delivery_tag) #回调函数中需要给发布者发送的消息channel.basic_qos(prefetch_count&＃61;1)channel.basic_consume(callback, queue&＃61;&＃39;task_queue&＃39;, no_ack&＃61;False) #no_ack&＃61;False当消费者接到消息后&＃xff0c;需要调用回掉函数告诉发布者&＃xff0c;消息的接受情况
print(&＃39; [*] Waiting for messages. To exit press CTRL&＃43;C&＃39;)
channel.start_consuming()

import pikaconnection &＃61; pika.BlockingConnection(pika.ConnectionParameters(host&＃61;&＃39;localhost&＃39;))
channel &＃61; connection.channel()channel.exchange_declare(exchange&＃61;&＃39;logs&＃39;, #交换机的名称为logsexchange_type&＃61;&＃39;fanout&＃39;)message &＃61; "info: Hello World!"
channel.basic_publish(exchange&＃61;&＃39;logs&＃39;,routing_key&＃61;&＃39;&＃39;,body&＃61;message)
print(" [x] Sent %r" % message)
connection.close()

import pikaconnection &＃61; pika.BlockingConnection(pika.ConnectionParameters(host&＃61;&＃39;localhost&＃39;))
channel &＃61; connection.channel()
channel.exchange_declare(exchange&＃61;&＃39;logs&＃39;, #声明交换机的名称以及交换机发送消息的模式&＃xff0c;这的名称要和publishor的交换机的名称相同&＃xff0c;相当于publishor和consumor同时向exchange寻找&＃xff0c;
#否则publishor和consumor相互找不到&＃xff0c;会迷路&＃xff01;&＃xff01;exchange_type&＃61;&＃39;fanout&＃39;) #这不同的版本有可能会出错&＃xff0c;有的说的是可以写为type &＃61; &＃39;fanout&＃39;,在我的电脑上运行会出错&＃xff0c;
#改为exchange_type &＃61; &＃39;fanout&＃39;仍然会出错&＃xff0c;后来运行cmd->service.msc->找到rabbitmq关闭后再启动就OK了&＃xff0c;
#至于为啥&＃xff0c;如果你找到了&＃xff0c;跟我说一声&＃xff0c;先谢谢啦
result &＃61; channel.queue_declare(exclusive&＃61;True) # 不指定queue名字,rabbit会随机分配一个名字,exclusive&＃61;True会在使用此queue的消费者断开后,自动将queue删除
queue_name &＃61; result.method.queue #得到队列的名字
channel.queue_bind(exchange&＃61;&＃39;logs&＃39;,queue&＃61;queue_name) #将交换机和队列的名字绑定&＃xff0c;也就是说exchange &＃61; ‘logs&＃39;的交换机只能公国queue &＃61; queue_name来发送消息
print(&＃39; [*] Waiting for logs. To exit press CTRL&＃43;C&＃39;)
def callback(ch, method, properties, body):print(" [x] %r" % body)channel.basic_consume(callback,queue&＃61;queue_name,no_ack&＃61;True)
channel.start_consuming()

import pika
import uuidclass FibonacciRpcClient(object):def __init__(self):self.connection &＃61; pika.BlockingConnection(pika.ConnectionParameters(host&＃61;&＃39;localhost&＃39;))self.channel &＃61; self.connection.channel()result &＃61; self.channel.queue_declare(exclusive&＃61;True)self.callback_queue &＃61; result.method.queueself.channel.basic_consume(self.on_response, #类中定义的方法__init__()是为了建立链接&＃xff0c;同时定义一个callback_queue,再basic_publish中传递给server端&＃xff0c;用于存放运行的结果no_ack&＃61;True,queue&＃61;self.callback_queue) #这里的basic_consume只是声明&＃xff0c;如果要取消息应该取callback__queue中取def on_response(self, ch, method, props, body):if self.corr_id &＃61;&＃61; props.correlation_id: #确定发的指令和收到的结果是相互对应的self.response &＃61; bodydef call(self, n):self.response &＃61; Noneself.corr_id &＃61; str(uuid.uuid4())self.channel.basic_publish(exchange&＃61;&＃39;&＃39;,routing_key&＃61;&＃39;rpc_queue&＃39;,properties&＃61;pika.BasicProperties(reply_to&＃61;self.callback_queue,correlation_id&＃61;self.corr_id,),body&＃61;str(n))while self.response is None:self.connection.process_data_events() #去队列中取数据&＃xff0c;不停的循环&＃xff0c;非阻塞版的start_consuming(),去调用basic_consume()return int(self.response)fibonacci_rpc &＃61; FibonacciRpcClient()
print(" [x] Requesting fib(30)")
response &＃61; fibonacci_rpc.call(30)
print(" [.] Got %r" % response)

import pika
import timeconnection &＃61; pika.BlockingConnection(pika.ConnectionParameters(host&＃61;&＃39;localhost&＃39;))
channel &＃61; connection.channel()
channel.queue_declare(queue&＃61;&＃39;rpc_queue&＃39;)def fib(n):if n &＃61;&＃61; 0:return 0elif n &＃61;&＃61; 1:return 1else:return fib(n - 1) &＃43; fib(n - 2)
def on_request(ch, method, props, body): #method存放的是bsic_consume中读取数据routing_key,props中存放的是从basic_consume中读取的protperties的数据&＃xff0c;body存放的是从发送过来的消息。n &＃61; int(body)print(" [.] fib(%s)" % n)response &＃61; fib(n)ch.basic_publish(exchange&＃61;&＃39;&＃39;,routing_key&＃61;props.reply_to,properties&＃61;pika.BasicProperties(correlation_id&＃61; props.correlation_id),body&＃61;str(response))ch.basic_ack(delivery_tag&＃61;method.delivery_tag)channel.basic_qos(prefetch_count&＃61;1)
channel.basic_consume(on_request, queue&＃61;&＃39;rpc_queue&＃39;)print(" [x] Awaiting RPC requests")
channel.start_consuming()