统一消息系统

Python

# 获取客户信息

import requests

def get_customer_info(customer_id):

url = f"https://api.customer-system.com/customers/{customer_id}"

response = requests.get(url)

if response.status_code == 200:

return response.json()

else:

return None

# 同步到订单系统

def sync_order_system(customer_data):

url = "https://api.order-system.com/orders"

payload = {

"customer_name": customer_data["name"],

"email": customer_data["email"],

"phone": customer_data["phone"]

}

response = requests.post(url, json=payload)

return response.status_code == 201

# 主函数

if __name__ == "__main__":

customer_id = 12345

customer_data = get_customer_info(customer_id)

if customer_data:

result = sync_order_system(customer_data)

print("同步成功" if result else "同步失败")

else:

print("无法获取客户信息")

Python

# 生产者：发送客户信息到队列

import pika

def send_to_queue(customer_data):

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

channel = connection.channel()

channel.queue_declare(queue='customer_sync')

channel.basic_publish(

exchange='',

routing_key='customer_sync',

body=str(customer_data)

)

print(" [x] Sent customer data to queue")

connection.close()

# 消费者：接收客户信息并同步到订单系统

def receive_from_queue():

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

channel = connection.channel()

channel.queue_declare(queue='customer_sync')

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

customer_data = eval(body.decode())

print(f" [x] Received customer data: {customer_data}")

# 这里调用同步逻辑

# sync_order_system(customer_data)

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

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

channel.start_consuming()

# 主函数

if __name__ == "__main__":

# 示例：发送客户信息

customer_data = {"id": 12345, "name": "张三", "email": "zhangsan@example.com"}

send_to_queue(customer_data)

# 示例：接收消息

# receive_from_queue()

Java

// Canal客户端配置

public class CanalClient {

public static void main(String[] args) {

String destination = "example";

String ip = "127.0.0.1";

int port = 11111;

// 创建Canal连接

CanalConnector connector = CanalConnectors.newSingleConnector(

new InetSocketAddress(ip, port), destination, "", "");

connector.connect();

connector.subscribe(".*\\..*");

connector.rollback();

while (true) {

Message message = connector.get(1000);

List rowDatas = message.getRowDatas();

for (RowData rowData : rowDatas) {

System.out.println("Table: " + message.getTable());

System.out.println("Old Values: " + rowData.getOldMap());

System.out.println("New Values: " + rowData.getNewMap());

// 这里可以添加同步逻辑

}

}

}

}