rabbitmq_消息持久化_消息公平分发_消息广播

发布时间:2020-07-20 14:47:50 作者:leiwenbin627
来源:网络 阅读:471

生产者_procudure_send_消息持久化
import pika

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

 

消费者_consumer_recive_消息公平分发
# _*_coding:utf-8_*_
__author__ = 'Alex Li'
import pika

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 消费持久化的消息


def callback(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_广播模式
import pika
import sys

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'
import pika

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


def callback(ch, method, properties, body):
   
print(" [x] %r" % body)


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

channel.start_consuming()
推荐阅读:
  1. RabbitMQ消息分发轮询
  2. rabbitmq_有选择发送接收_消息过滤

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

消息 持久化 消息广播

上一篇:用代码详解在Java中如何操作Zookeeper

下一篇:Python如何读取文件内容为字符串

相关阅读

您好,登录后才能下订单哦!

密码登录
登录注册
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》