mysql消息推送

DELIMITER $$
CREATE TRIGGER after_user_insert
AFTER INSERT ON users
FOR EACH ROW
BEGIN
    INSERT INTO user_insert_queue (user_id, created_at) VALUES (NEW.id, NOW());
END$$
DELIMITER ;

import pika
import mysql.connector

# 连接MySQL
db = mysql.connector.connect(
    host="localhost",
    user="user",
    password="password",
    database="testdb"
)

cursor = db.cursor()

# 连接RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 声明队列
channel.queue_declare(queue='user_insert_queue')

# 从MySQL中读取数据并推送到RabbitMQ
cursor.execute("SELECT * FROM user_insert_queue")
rows = cursor.fetchall()

for row in rows:
    channel.basic_publish(exchange='', routing_key='user_insert_queue', body=str(row))

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

# 关闭连接
cursor.close()
db.close()
connection.close()

import pika

# 连接RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 声明队列
channel.queue_declare(queue='user_insert_queue')

def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)

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

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