统一消息系统

小明：最近我们公司要上线一个消息管理平台，但同时也需要和现有的投标文件系统对接，你有什么建议吗？

小李：这个问题很常见。我们可以考虑使用API来实现两者的集成。消息管理平台通常会提供RESTful API接口，而投标文件系统也可以通过类似方式暴露数据。

小明：那具体的实现步骤是怎样的呢？有没有什么需要注意的地方？

小李：首先，我们需要确定两个系统的通信方式。比如，投标文件系统可能需要在生成新文件时，向消息管理平台发送通知。这时候，我们可以用Webhook或者MQTT等协议。

小明：Webhook听起来不错，但我对它的具体实现不太清楚，你能举个例子吗？

小李：当然可以。假设投标文件系统在上传文件后，会向消息管理平台的某个端点发送POST请求，里面包含文件信息。我们可以用Python来写一个简单的Web服务接收这些请求。

小明：好的，那这个Web服务的代码应该怎么写呢？

小李：下面是一个简单的Flask应用示例，用于接收来自投标文件系统的Webhook请求，并将相关信息保存到消息管理平台中。

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    data = request.json
    if not data:
        return jsonify({"error": "No JSON data received"}), 400

    # 假设data中包含文件名、用户ID、时间戳等信息
    file_name = data.get('file_name')
    user_id = data.get('user_id')
    timestamp = data.get('timestamp')

    # 这里可以调用消息管理平台的API
    # 比如：发送一条消息给用户
    send_message_to_platform(user_id, f"您提交了新的投标文件: {file_name}，时间: {timestamp}")

    return jsonify({"status": "success"}), 200

def send_message_to_platform(user_id, message):
    # 示例：调用消息管理平台的API
    print(f"向用户 {user_id} 发送消息: {message}")
    # 实际中可以通过HTTP POST请求发送到消息平台的API端点

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

    

小明：这段代码看起来挺清晰的。那如果消息管理平台没有现成的API怎么办？是不是需要自己搭建一个？

小李：如果是这样，那我们就需要在消息管理平台内部实现一个接收Webhook的接口。或者，可以使用中间件，比如RabbitMQ或Kafka，作为消息队列，来解耦两个系统。

小明：那这种情况下，我该怎么设计系统架构呢？

小李：我们可以采用事件驱动架构。投标文件系统在发生某些事件（如文件上传）时，发布一个事件到消息队列。消息管理平台订阅该事件，并做出相应处理。

小明：听起来不错，那能给我一个这样的架构图示吗？

小李：虽然不能画图，但我可以用文字描述一下。投标文件系统 → 消息队列（如RabbitMQ） → 消息管理平台。当投标文件系统有新文件时，它会向消息队列发送一条消息，消息管理平台监听该队列并处理。

小明：那消息队列的代码怎么写呢？

小李：这里有一个简单的RabbitMQ生产者示例，用于将投标文件的信息发送到消息队列。

import pika

def send_to_queue(file_name, user_id, timestamp):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()

    channel.queue_declare(queue='bid_files')

    message = f'{{"file_name": "{file_name}", "user_id": "{user_id}", "timestamp": "{timestamp}"}}'
    channel.basic_publish(exchange='', routing_key='bid_files', body=message)

    print(f" [x] Sent: {message}")
    connection.close()

# 示例调用
send_to_queue("project1_bid.pdf", "user123", "2025-04-05T14:30:00Z")

    

小明：那消息管理平台这边怎么消费这些消息呢？

小李：我们可以编写一个消费者程序，从消息队列中读取消息，并将其转发到消息管理平台的API。

import pika
import requests

def callback(ch, method, properties, body):
    print(f" [x] Received: {body}")
    data = eval(body)  # 简单解析JSON字符串，实际应使用json模块
    file_name = data['file_name']
    user_id = data['user_id']
    timestamp = data['timestamp']

    # 调用消息管理平台API
    url = 'http://message-platform-api.com/api/messages'
    payload = {
        "user_id": user_id,
        "message": f"您提交了新的投标文件: {file_name}，时间: {timestamp}"
    }
    response = requests.post(url, json=payload)
    print(f" [x] Message sent to platform: {response.status_code}")

def consume_messages():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()

    channel.queue_declare(queue='bid_files')

    channel.basic_consume(queue='bid_files', on_message_callback=callback, auto_ack=True)

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

if __name__ == '__main__':
    consume_messages()

    

小明：看来整个流程已经比较清晰了。不过，有没有什么安全方面的问题需要注意？

小李：确实需要注意。比如，Webhook的请求应该验证来源，防止恶意攻击。同时，消息队列也应设置访问权限，确保只有授权的服务才能发送或接收消息。

小明：明白了。那如果消息管理平台本身也需要推送消息给投标文件系统呢？比如，提醒用户审核文件？

小李：这种情况下，可以反过来设计。消息管理平台在需要通知时，调用投标文件系统的API，发送消息或更新状态。

小明：那是不是需要在投标文件系统中也添加一个接收消息的接口？

小李：没错。比如，投标文件系统可以有一个REST API，允许外部系统发送通知。消息管理平台在检测到某些事件（如文件被审核通过）后，调用这个API。

小明：听起来整个系统就形成了一个闭环，互相协作。这真是一个高效的集成方式。

小李：是的，通过API和消息队列的结合，可以实现灵活、可扩展的系统集成。这也是现代微服务架构中常见的做法。

小明：非常感谢你的讲解，我对整个流程有了更深入的理解。

小李：不客气！如果你还有其他问题，随时问我。