东莞设计兼职网站建设,常州网络公司鼎豪网络网站建设,网页传奇发布网,中铁十六局集团门户登录文章目录 二、Python-flask-rabbitMQ-插件方式整合引言具体步骤1 安装依赖#xff1a;2 编写实体类#xff1a;3 编写消费者和生产者#xff1a;4 初始化消费者和生产者#xff1a;5 其他地方使用生产者 二、Python-flask-rabbitMQ-插件方式整合
引言
当今互联网应用的高… 文章目录 二、Python-flask-rabbitMQ-插件方式整合引言具体步骤1 安装依赖2 编写实体类3 编写消费者和生产者4 初始化消费者和生产者5 其他地方使用生产者 二、Python-flask-rabbitMQ-插件方式整合
引言
当今互联网应用的高并发场景下如何保证可靠的消息传递和高效的处理成为了一项重要的挑战。在这种情况下RabbitMQ作为一种可靠的消息队列系统被广泛应用于各个领域。
本文将介绍如何使用RabbitMQ插件整合Flask框架实现并发性生产者和消费者的兼容。通过这种方式我们可以利用RabbitMQ的优势来确保消息的可靠传递并提高应用程序的处理能力。
首先我们需要在Flask应用程序中引入RabbitMQ插件。可以使用Pika库来连接和操作RabbitMQ。通过创建一个连接池管理连接对象我们可以避免频繁地创建和销毁连接提高效率。
同时为了处理高并发的生产者我们可以使用批量发送消息的方式来减少通道创建和消息发布的开销。通过设置缓冲区来收集一定数量或一定时间间隔内的消息然后批量发送可以更有效地利用资源。
此外对于高并发的消费者我们可以考虑使用异步的方式来处理消息。通过将消息发送任务交给后台线程或异步任务队列处理可以避免请求的堵塞提高应用程序的并发能力。
通过以上的优化方案我们可以在Flask应用程序中充分利用RabbitMQ的功能并且兼容高并发的生产者和消费者。这将帮助我们构建更可靠、高效的消息队列系统应对日益增长的并发访问需求。
总之使用RabbitMQ插件整合Flask框架并采用优化方案来兼容并发性生产者和消费者是构建可靠、高效消息传递系统的关键一步。通过这种方式我们能够更好地应对高并发场景下的挑战提升应用程序的性能和稳定性。
具体步骤
1 安装依赖
使用pip安装pika库
pip install pika2 编写实体类
from queue import Queue
from threading import Lockimport pika# 定义交换机类型的枚举值
class ExchangeType:DEFAULT defaultDIRECT directFANOUT fanoutTOPIC topicclass RabbitMQ:def __init__(self, hostlocalhost, port5672, usernameguest, passwordguest, pool_size10):self.credentials pika.PlainCredentials(username, password)self.parameters pika.ConnectionParameters(hosthost, portport, credentialsself.credentials)self.connection_pool Queue(pool_size) # 连接池存储连接和信道self.lock Lock() # 互斥锁用于对连接池的访问进行同步for _ in range(pool_size):connection self._create_connection()channel connection.channel()self.connection_pool.put((connection, channel))def _create_connection(self):return pika.BlockingConnection(self.parameters)def get_channel(self):with self.lock:connection, channel self.connection_pool.get() # 从连接池获取连接和信道return connection, channeldef release_channel(self, connection, channel):with self.lock:self.connection_pool.put((connection, channel)) # 将连接和信道放回连接池def send_message(self, exchange, routing_key, message, exchange_typeExchangeType.DEFAULT):connection, channel self.get_channel()try:channel.exchange_declare(exchangeexchange, exchange_typeexchange_type) # 声明交换机并指定类型channel.basic_publish(exchangeexchange, routing_keyrouting_key, bodymessage) # 发布消息finally:self.release_channel(connection, channel)def receive_messages(self, queue, callback):connection, channel self.get_channel()try:channel.queue_declare(queuequeue, durableTrue) # 声明队列并标记为持久化# channel.queue_purge(queuequeue) # 清空队列以防之前的非持久化消息残留channel.basic_qos(prefetch_count10) # 每次从 RabbitMQ 获取 10 条消息channel.basic_consume(queuequeue, on_message_callbackcallback, auto_ackFalse) # 消费消息并设置回调函数channel.start_consuming() # 开始消费消息finally:self.release_channel(connection, channel)
3 编写消费者和生产者
def connect_mq(app):# 初始化 RabbitMQ 实例rabbitmq RabbitMQ(hostapp.config[RABBITMQ_HOST], port5672, usernameguest, passwordguest)# 在应用上下文中注册 RabbitMQ 实例app.config[RABBITMQ] rabbitmq# consume_mq(app)#thread threading.Thread(targetconsume_mq, args(app,))# 启动线程thread.start()def consume_mq(app):# 启动消费者程序开始接收和处理消息def callback(ch, method, properties, body):try:print(f消息队列内容 {body.decode()})# 处理rabbitMQ内容to_transcribe(body.decode())except Exception as e:print(str(e))ch.basic_ack(delivery_tagmethod.delivery_tag)# 启动消费者程序开始接收和处理消息with app.app_context():rabbitmq current_app.config[RABBITMQ]rabbitmq.receive_messages(audio_queue, callback)
4 初始化消费者和生产者
def create_app():app Flask(__name__)connect_mq_v1(app)5 其他地方使用生产者 class MessageHandler:处理存放音频将所有的任务都放在MQ里面def __init__(self, dir_name,uuid_str, back_url, file_url, request_type, file_name, *args, **kwargs):# 文件夹名称self.dir_name dir_name# 文件名称self.file_name file_name# 文件上传类型self.request_type request_type# 文件存储位置self.file_url file_url# 客户端回调地址self.back_url back_url# 唯一标识self.uuid_str uuid_strdef send(self)::param content_type:队列类型:param rpc:MQ对象:return:try:# 发送消息队列# rpc.send_expire(bodyjson.dumps(self.to_json()), exchangeaudio_queue, keyaudio_queue)rabbitmq current_app.config[RABBITMQ]rabbitmq.send_message(audio_queue, audio_queue, json.dumps(self.to_json()))print(发送消息到mq成功,用于存放音频信息)except Exception as e:print(f发送消息到mq服务失败请检查, {e})def to_json(self):_dict self.__dict__return _dict# 将请求体和uuid放到rabbitMQ中
MessageHandler(**dates).send()
