如何解决rabbitmq 上的消费者不消费消息
我正在使用 RabbitMQ 消息代理。我创建了一个发布者,该发布者可以正确发布一个不消耗任何消息的消费者。
我的消费者代码是这样的:
import sys
import json
import pika
from fastapi import FastAPI
from fastapi.logger import logger
from decouple import config
import logging
from botocore.exceptions import ClientError
import csv
sys.path.append('../lib')
from manager_mq import Manager
from s3 import S3
_RABBITMQ_DEFAULT_USER = config('RABBITMQ_DEFAULT_USER')
_RABBITMQ_DEFAULT_PASS = config('RABBITMQ_DEFAULT_PASS')
_HOST = config('HOST')
_AWS_DEFAULT_REGION = config('AWS_DEFAULT_REGION')
_AWS_ACCESS_KEY_ID = config('AWS_ACCESS_KEY_ID')
_AWS_SECRET_ACCESS_KEY = config('AWS_SECRET_ACCESS_KEY')
_BUCKET_NAME = config('BUCKET_NAME')
mng = Manager(_RABBITMQ_DEFAULT_USER,_RABBITMQ_DEFAULT_PASS,_HOST)
channel = mng.channel()
mng.declare_exchange('rico','topic')
mng.create_queue('rico')
mng.declare_exchange('rico','topic')
app = FastAPI(debug=True)
def callback(ch,method,properties,body):
notification = json.dumps(json.loads(body))
ch.basic_act(delivery_tag = method.delivery_tag)
with open('log.txt','a') as l:
l.writelines(str(notification))
s3 = S3('s3',_AWS_DEFAULT_REGION,_AWS_ACCESS_KEY_ID,_AWS_SECRET_ACCESS_KEY)
latest = s3.get_latest_created_file(_BUCKET_NAME)
last_id = int(latest['Key'].split('.')[0])
id = last_id + 1
name = id.join('json')
obj = s3.save_object(_BUCKET_NAME,name,notification)
return notification
mng.basic_qos(1)
mng.basic_consume(queue_name='rico',callback=callback)
mng.consuming()
我在rabbitmq中创建了一个管理所有进程的类。
import pika
class Manager:
def __init__(self,default_user,default_password,host) -> object:
self.__default_user = default_user
self.__default_password = default_password
self.__host = host
def connection(self):
credentials = pika.PlainCredentials(self.__default_user,self.__default_password)
parameters = pika.ConnectionParameters(self.__host,credentials=credentials)
connection = pika.BlockingConnection(parameters)
return connection
def channel(self):
connection = self.connection()
channel = connection.channel()
return channel
def create_queue(self,queue_name):
queue = self.channel()
return queue.queue_declare(queue=queue_name,durable=True)
def declare_exchange(self,exchange,types):
return self.channel().exchange_declare(exchange=exchange,exchange_type=types)
def publish_message(self,routing,body):
return self.channel().basic_publish(exchange=exchange,routing_key=routing,body=body,properties=pika.BasicProperties(
delivery_mode=2
))
def queue_bind(self,queue_name):
self.channel.queue_bind(exchange=exchange,queue=queue_name)
def basic_consume(self,queue_name,callback,auto_act=True):
self.channel().basic_consume(queue=queue_name,on_message_callback=callback)
def consuming(self):
self.channel().start_consuming()
def close_connection(self):
self.connection().close()
def basic_qos(self,prefetch_count):
self.connection().channel().basic_qos(prefetch_count=prefetch_count)
我的消费者应该获取消息并将其转换为 Json 以像 AWS s3 中的对象一样保存,但是,没有任何反应。没有错误,AWS 中没有保存数据。
这个过程的结果应该类似于下图:
有人可以帮我解决这个问题吗?
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。