生产者_procudure_send_消息持久化
importpika
connection = pika.BlockingConnection(pika.ConnectionParameters(
'localhost')) # rabbit默认端口5672建立一个基本的 socket连接
channel = connection.channel() #声明一个管道 在管道里面发消息
#声明queue
channel.queue_declare(queue='hello',durable=True)#durable=True队列持久化
# n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
channel.basic_publish(exchange='',
routing_key='hello', # queue名字
body='Hello World!', # body发送的消息
properties=pika.BasicProperties(delivery_mode=2),#消息持久化
)
print(" [x] Sent 'Hello World!'")
connection.close()
成都地区优秀IDC服务器托管提供商(成都创新互联).为客户提供专业的德阳机房服务器托管,四川各地服务器托管,德阳机房服务器托管、多线服务器托管.托管咨询专线：18982081108
消费者_consumer_recive_消息公平分发
# _*_coding:utf-8_*_
__author__ ='Alex Li'
importpika
connection = pika.BlockingConnection(pika.ConnectionParameters(
'localhost'))#rabbit默认端口5672建立一个基本的 socket连接
channel = connection.channel()#声明一个管道 在管道里面收消息
# You may ask why we declare the queue again ‒ we have already declared it in our previous code.
# We could avoid that if we were sure that the queue already exists. For example if send.py program
# was run before. But we're not yet sure which program to run first. In such cases it's a good
# practice to repeat declaring the queue in both programs.
channel.queue_declare(queue='hello1')#声明queue消费持久化的消息
defcallback(ch, method, properties, body):#处理消息
print("---->",ch,method,properties)#ch管道内存对象地址 method:发给queue的信息
print(" [x] Received %r"% body)
ch.basic_ack(delivery_tag=method.delivery_tag)#手动确认消息是否接收
channel.basic_qos(prefetch_count=1)#根据权重发消息 一对一发
channel.basic_consume(#消费消息
callback,#如果收到消息，就调用CALLBACK函数来处理消息
queue='hello1',#从哪个队列里收消息
#no_ack=True,
)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()#启动 开始收消息 一直收，没有就卡主
生产者_send_广播模式
importpika
importsys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logging',exchange_type='fanout')
messages =' '.join(sys.argv[1:])or"info: Hello World!"
#messages="info: Hello World!"
channel.basic_publish(exchange='yyyy',
routing_key='',
body=messages)
print(" [x] Sent %r"% messages)
connection.close()
消费者_recive_广播模式
# _*_coding:utf-8_*_
__author__ ='Alex Li'
importpika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='yyyy',
exchange_type='fanout')
result = channel.queue_declare(exclusive=True) #不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
queue_name = result.method.queue#随机分配的queue
print("random queue name:",queue_name)
channel.queue_bind(exchange='yyyy',
queue=queue_name)#绑定转发器
print(' [*] Waiting for logs. To exit press CTRL+C')
defcallback(ch,method,properties, body):
print(" [x] %r"% body)
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()