python--使用pika库操作rabbitmq实现需求

Author: wencoo
Blog：https://wencoo.blog.csdn.net/
Date: 22/04/2024
Email: jianwen056@aliyun.com
Wechat：wencoo824
QQ：1419440391
Details:

文章目录

目录
- 正文或背景
- pika链接mq
- pika指定消费数量
- pika自动消费实现
- pika获取队列任务数量
- pika主动获取任务消费
- 根据需求实现的完整示例代码
- 参考
技术交流
音视频领域其他技术文章的链接
- opengl相关文章
- ffmpeg相关文章
- - ffmpeg原理相关文章
  - ffmpeg源码分析相关文章
  - ffmpeg指令相关文章
  - ffmpeg报错相关文章
- libass相关文章
- c/c++相关文章
- linux相关文章
- 其他文章
后面都是一些废话，不用看，刷分的
- 推广一个AI学习网站
- 中国软件行业倡议书
- 作者有话说
- 关于内卷

在这里插入图片描述

有这样业务场景，算法部的同事有一个算法需要集成，有国内和国外两条链路，使用rabbitmq对应的试两个队列，但是他的算法只能开启一个实例，不能开两个进程，否则计算资源不足会崩溃（此处我想吐槽，做算法这帮人，工资又高，结果工程能力太差啦，所谓的算法也不过把github的开源库拿来改改参数，怎么好意思叫算法，搞不懂现在的中国互联网环境了），基于以上原因，需求则是需要在只开一个实力的情况下，消费两个队列里的任务。

由于算法模型都是python写的，所以要使用python来处理mq，python处理mq的驱动库是pika，下面来学习一下pika一些使用操作。

pika链接mq

import pika
import timecredentials = pika.PlainCredentials('admin', 'Hasx9527')
connection = pika.BlockingConnection(pika.ConnectionParameters(host = '180.184.35.67',port = 5672,virtual_host = '/',credentials = credentials))
channel = connection.channel()

pika指定消费数量

# 可选：指定消费者处理队列消息的数量
channel.basic_qos(prefetch_count=1)  # 实现消费者负载均衡，表示同时只能处理1条消息，待处理完成后在接收下一条消息

pika自动消费实现

def callback(ch, method, properties, body):ch.basic_ack(delivery_tag = method.delivery_tag)print(body.decode())time.sleep(10)# 可选：指定消费者处理队列消息的数量
channel.basic_qos(prefetch_count=1)  # 实现消费者负载均衡，表示同时只能处理1条消息，待处理完成后在接收下一条消息# 告诉rabbitmq，用callback来接收消息
channel.basic_consume('test.python.pika.1',callback)
# 开始接收信息，并进入阻塞状态，队列里有信息才会调用callback进行处理
channel.start_consuming()

pika获取队列任务数量

queue_info = channel.queue_declare(queue,passive=True)
message_count = queue_info.method.message_count
print(f"队列任务数量为:{message_count}")

pika主动获取任务消费

method_frame, header_frame, body = channel.basic_get(queue)
if method_frame:# print(method_frame, header_frame, body)  # 收到的全部数据  就要bodyprint(body)          # 是b'xxx'的格式channel.basic_ack(method_frame.delivery_tag)  #用自动应答 这个应该不要
else:print('消费单条 没有收到消息')

根据需求实现的完整示例代码

import pika
import timecredentials = pika.PlainCredentials('admin', 'Hasx9527')
connection = pika.BlockingConnection(pika.ConnectionParameters(host = '180.184.35.67',port = 5672,virtual_host = '/',credentials = credentials))
channel = connection.channel()queue1 = 'test.python.pika.1'
queue2 = 'test.python.pika.2'
def queueConsumer(queueName):while True:# 申明消息队列，消息在这个队列传递，如果不存在，则创建队列queue = queueNamechannel.queue_declare(queue , durable = True)# 可选：指定消费者处理队列消息的数量channel.basic_qos(prefetch_count=1)  # 实现消费者负载均衡，表示同时只能处理1条消息，待处理完成后在接收下一条消息method_frame, header_frame, body = channel.basic_get(queue)if method_frame:print(body)          # 是b'xxx'的格式time.sleep(10)channel.basic_ack(method_frame.delivery_tag)  #用自动应答 这个应该不要else:time.sleep(1)breakdef queueConsumer2(queueName,queueName2):while True:# 申明消息队列，消息在这个队列传递，如果不存在，则创建队列queue = queueName2channel.queue_declare(queue , durable = True)queue_info = channel.queue_declare(queueName,durable = True,passive=True)message_count = queue_info.method.message_countif message_count != 0:break# 可选：指定消费者处理队列消息的数量channel.basic_qos(prefetch_count=1)  # 实现消费者负载均衡，表示同时只能处理1条消息，待处理完成后在接收下一条消息method_frame, header_frame, body = channel.basic_get(queue)if method_frame:print(body)          # 是b'xxx'的格式time.sleep(10)channel.basic_ack(method_frame.delivery_tag)  #用自动应答 这个应该不要else:time.sleep(1)breakwhile True:print("开始运行队列1================\n")queueConsumer(queue1)print("开始运行队列2================\n")queueConsumer2(queue1,queue2)