统一消息系统

# 消息生产者（producer.py）

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

channel = connection.channel()

channel.queue_declare(queue='task_queue')

message = 'Hello, this is a task message!'

channel.basic_publish(exchange='',

routing_key='task_queue',

body=message)

print(" [x] Sent %r" % message)

connection.close()

# 消息消费者（consumer.py）

import pika

def callback(ch, method, properties, body):

print(" [x] Received %r" % body)

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

channel = connection.channel()

channel.queue_declare(queue='task_queue')

channel.basic_consume(callback,

queue='task_queue',

no_ack=True)

print(' [*] Waiting for messages. To exit press CTRL+C')

channel.start_consuming()

# 修改后的生产者（支持不同消息类型）

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

channel = connection.channel()

channel.exchange_declare(exchange='logs',

type='fanout')

messages = {

'email': 'This is an email notification.',

'sms': 'This is an SMS alert.',

'log': 'This is a system log entry.'

}

for msg_type, content in messages.items():

channel.basic_publish(exchange='logs',

routing_key='',

body=content)

print(f" [x] Sent {msg_type} message: {content}")

connection.close()

# 多个消费者，分别监听不同类型的消息

import pika

def callback_email(ch, method, properties, body):

print(f" [Email] Received: {body}")

def callback_sms(ch, method, properties, body):

print(f" [SMS] Received: {body}")

def callback_log(ch, method, properties, body):

print(f" [Log] Received: {body}")

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

channel = connection.channel()

channel.exchange_declare(exchange='logs', type='fanout')

result = channel.queue_declare(queue='', exclusive=True)

queue_name = result.method.queue

channel.queue_bind(exchange='logs', queue=queue_name)

# 假设我们有两个消费者，分别监听不同的消息类型

# 这里简化处理，实际中可以通过标签或路由键区分

channel.basic_consume(queue=queue_name, on_message_callback=callback_email, auto_ack=True)

channel.basic_consume(queue=queue_name, on_message_callback=callback_sms, auto_ack=True)

channel.basic_consume(queue=queue_name, on_message_callback=callback_log, auto_ack=True)

print(' [*] Waiting for logs. To exit press CTRL+C')

channel.start_consuming()

# 启用消息确认机制

def callback(ch, method, properties, body):

print(" [x] Received %r" % body)

ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(callback, queue='task_queue')

# 禁用自动确认，改为手动确认

channel.basic_consume(callback, queue='task_queue', no_ack=False)