统一消息系统

import pika

# 消息生产者

def send_message(message):

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

channel = connection.channel()

channel.queue_declare(queue='bid_center')

channel.basic_publish(exchange='',

routing_key='bid_center',

body=message)

print(f" [x] Sent '{message}'")

connection.close()

# 消息消费者

def receive_message():

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

channel = connection.channel()

channel.queue_declare(queue='bid_center')

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

print(f" [x] Received '{body.decode()}'")

channel.basic_consume(callback,

queue='bid_center',

no_ack=True)

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

channel.start_consuming()

# 投标书相关操作

def submit_bid_document(document_id):

message = f"Bid document {document_id} has been submitted."

send_message(message)

# 测试

if __name__ == '__main__':

# 模拟投标书提交

submit_bid_document("BID-2024-001")

# 接收消息

# receive_message() # 可以单独运行该函数来监听消息

import json

from pika import BlockingConnection, ConnectionParameters

# 定义消息处理器

class MessageHandler:

def handle(self, message):

data = json.loads(message)

if data['type'] == 'submit_bid':

self.handle_submit_bid(data['document_id'])

elif data['type'] == 'approve_bid':

self.handle_approve_bid(data['document_id'])

elif data['type'] == 'reject_bid':

self.handle_reject_bid(data['document_id'])

def handle_submit_bid(self, document_id):

print(f"Handling submission of bid document: {document_id}")

# 这里可以调用投标书提交的业务逻辑

def handle_approve_bid(self, document_id):

print(f"Approving bid document: {document_id}")

# 这里可以调用审批逻辑

def handle_reject_bid(self, document_id):

print(f"Rejecting bid document: {document_id}")

# 这里可以调用拒绝逻辑

# 消息消费者

def consume_messages():

connection = BlockingConnection(ConnectionParameters('localhost'))

channel = connection.channel()

channel.queue_declare(queue='bid_center')

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

message = body.decode()

handler = MessageHandler()

handler.handle(message)

channel.basic_consume(callback, queue='bid_center', no_ack=True)

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

channel.start_consuming()

# 示例消息

if __name__ == '__main__':

# 模拟发送消息

send_message(json.dumps({'type': 'submit_bid', 'document_id': 'BID-2024-001'}))

send_message(json.dumps({'type': 'approve_bid', 'document_id': 'BID-2024-001'}))

send_message(json.dumps({'type': 'reject_bid', 'document_id': 'BID-2024-001'}))

# 启动消费者

# consume_messages()