您现在的位置是：首页 > 其他

当前栏目

rabbitmq-python使用教程

消息教程队列连接队列

2023-03-02 11:22:57 时间

我们将用Python编写两个小程序;

发送单个消息的生产者（发送者）

接收消息并将其打印出来的消费者（接收者）。

传递的消息就是 “Hello World”。

使用前，先安装pika

pip install pika

# **发送端** #

[这是我们的第一个程序send.py][send.py]，将向队列发送一条消息。

我们需要做的第一件事是建立与RabbitMQ服务器的连接。

import pika
    # 创建连接对象，连接本地
    connection = pika.BlockingConnection（pika.ConnectionParameters（'localhost'））
    # 在连接的基础上，创建一个频道
    channel = connection.channel()
    # 创建 消息队列，表明我们向消息队列为hello的发送消息
    channel.queue_declare（queue = 'hello'）
    
    """ # 如果远程连接，需要账号密码，使用pika.PlainCredentials生成 身份验证凭据 username,pwd = ("guest","guest") user_pwd = pika.PlainCredentials(username, pwd) # 远程连接192.168.56.102服务器，身份凭证为user_pwd对象 connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.56.102', credentials=user_pwd)) """

此时我们已经做好了发送准备

# 我们需要确保列队存在，所以先创建一个列队
    channel.queue_declare（queue = 'hello'）
    # routing_key为刚刚的消息队列名字，body为发送的字符串
    channel.basic_publish(exchange = '',
                          routing_key = 'hello',
                          body = 'hello world')
    print("[我 叶良辰]发送了 'Hello World！'")

[send.py][send.py 1] 返送端\[生产者\]的完整代码

import pika
    username,pwd = ("guest","guest")
    user_pwd = pika.PlainCredentials(username, pwd)
    
    connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.56.102', credentials=user_pwd))
    channel = connection.channel()
    channel.queue_declare(queue = 'hello')
    
    channel.basic_publish(exchange = '',
                          routing_key = 'hello',
                          body = 'hello world')
    print("[我 叶良辰]发送了 'Hello World！'")

# **接收端** #

[我们的第二个程序receive.py][receive.py]。

无论是发送端还是接收端，我们都需要连接rabbitmq-server,代码相同

import pika
    
    username, pwd = ("guest", "guest")
    user_pwd = pika.PlainCredentials(username, pwd)
    
    connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.56.102', credentials=user_pwd))
    channel = connection.channel()
    # 接收端，也需要创建这个列队
    # 重复在两个程序中重新声明队列是一个好习惯
    # 不论我们先运行哪一个程序，消息都不会被抛弃。
    channel.queue_declare(queue='hello')
    
    
    def callback(ch, method, properties, body):
        """ 这是回调函数"""
        print(" [x] Received %r" % body)
    
    
    channel.basic_consume(queue = 'hello',
        auto_ack = True,
        on_message_callback = callback)
    print('[*]等待消息。退出按CTRL + C')
    channel.start_consuming()
    
    """ 接收端永无止境的轮回(循环)，去接收列队的消息 """

上面我们介绍了简单的发送消息，我们经过实践和使用，会发现一些特点和问题。

""" 我们的发送一条消息，如果有两个接收端，那么只有其中一个会接收到消息。 这种轮询的机制，类似我们nginx服务器上的负载均衡。 它会平均的将任务 分配到我们的worker(接收端) 那么问题1，如果我们的回调函数执行到一半，接收端挂掉了，消息就会消失不见了。 我们不希望这样，我们希望的是，worker1处理一半挂掉了，将消息传递给worker2重新处理。 """
    
    def callback(ch, method, properties, body):
        print " [x] Received %r" % (body,)
        time.sleep( body.count('.') )
        print " [x] Done"
        # 确认消息
        ch.basic_ack(delivery_tag = method.delivery_tag)
    
    channel.basic_consume(queue='hello', on_message_callback=callback)
    
    """ 默认情况下，手动的消息确认已打开。 前面的例子，我们通过auto_ack = True 关闭了 如上面的代码，我们的消息只有被确认后，才会从列队中删除，如果没有处理完全中途退出，会有其他worker进行处理 """

解决了一个问题，还有一个问题，如果我们的rabbitmq死掉了，重启后，列队都没有了。

我们需要本地持久化的存储这个消息队列。

# 首先确保rabbitmq不会丢失我们的队列
    channel.queue_declare(queue = 'hello'，durable = True)
    """ 如果出现异常,让我们声明一个不同命的队列，因为这个队列已经存在，如果使用的参数和队列不匹配，会报错。 channel.queue_declare(queue = 'task_queue'，durable = True) 这里发送端，接收端也要更改响应的代码 """
    
    channel.basic_publish(exchange='',
                          routing_key="task_queue",
                          body=message,
                          properties=pika.BasicProperties(
                             delivery_mode = 2, # 使我们的消息也持久化
                          ))
    """ delivery_mode = 2 虽然告诉了rabbitmq持久化存储我们的消息，但是这中间有一个空档 但是一般的服务这样完全可以了，如果想确保消息不丢失，可以使用我们刚刚提到的确认的模式。 publisher confirms https://www.rabbitmq.com/confirms.html """

其他文档

https://www.rabbitmq.com/tutorials/tutorial-three-python.html

# 消费者端 设置分配规则
    channel.basic_qos(prefetch_count=1)
    """ 添加这一句，表示这个worker最多处理一个任务， 如果有过没处理完，不会接收其他任务请求. """

发给所有人，推送。

# 发送端 ，不需要声明Queue
    channel.queue_declare(exchange="logs",type="fanout")
    channel.basic_publish(exchange = 'logs',
                          routing_key = '',
                          body = 'hello world')
    """ 因为推送给所有人，是广播，所以不需要写Q exchange = '起个名字',所有绑定到这个的queue，都可以接收到消息 type = 类型 fanout,发送给所有人 direct ，接收自己想要的，选择。 topic 。 direct和topic还没有实践了解。 """

# 接收端
    #不指定Queue,随机生成一个，不用就删除，随机的名字
    result = channel.queue_declare(exclusive=True)
    queue_name = result.method.queue
    channel.queue_bind(exchange='logs',
    			queue=queue_name) 
    # 绑定到exchange,并使用刚刚生成的queue发送接收
    """ 类似收音机，这个消息不会存在列队中，如果当时不在，消息就过去了 """

[send.py]: http://xn--send-k84fuim2cb52e2uge5kdj9bx3ijudcr4h.py
[send.py 1]: http://send.py
[receive.py]: http://xn--receive-q33kn2ao1af27j11iq40ejlk17d.py

猜你喜欢

如何免费申请域名
怎样申请免费域名
基于相同颜色连通像素个数的统计进行图像字符识别
如何免费注册域名
天池大数据竞赛 Spaceack带你利用Pandas，趋势图与桑基图分析美国选民候选人喜好度
怎么申请免费域名
单机版的K8s环境搭建及部署Java Web应用Demo
如何申请免费域名
Python利用matplotlib万花尺画月饼
备案域名哪个网站好
阿里云飞天加速计划：ECS体验v
购买阿里云服务器一年多少钱
阿里云高校学生免费算力活动使用感受
企业级云服务器多少钱
阿里云服务器能承载多少人数
北京注册域名多少钱
云服务器买多少带宽
飞天加速计划yyds
www域名多少钱
云服务器带宽怎么选

zl程序教程

当前栏目

rabbitmq-python使用教程

相关文章