如何在PostgreSQL中创建触发器向MySQL插入数据以同步用户？

Absolutely feasible, but you shouldn’t directly establish a MySQL connection inside a PostgreSQL trigger—this can lead to performance bottlenecks, transaction consistency issues, and wasted connection resources. Let’s break down the best approaches to solve your user data sync problem.

Why not a direct trigger-to-MySQL connection? #

• Performance overhead: Triggers run synchronously. Any delay or slowness in MySQL will block the original PostgreSQL write operation, hurting your main system’s responsiveness.
• Transaction consistency risks: If the MySQL insert fails (e.g., network glitch, constraint violation), do you roll back the PostgreSQL transaction? This creates complex edge cases that are hard to handle reliably.
• Connection resource drain: Every trigger execution would spin up a new MySQL connection, which can quickly exhaust your database’s connection limits under high load.

Recommended Solutions #

Option 1: Use PostgreSQL’s mysql_fdw (Lightweight, Strong Consistency) #

The MySQL Foreign Data Wrapper (mysql_fdw) lets you treat MySQL tables as if they’re native PostgreSQL tables. You can use this in a trigger to sync data seamlessly.

Step-by-Step Implementation:

1. Install the mysql_fdw extension:

   CREATE EXTENSION mysql_fdw;
   

2. Define a server pointing to your MySQL instance:

   CREATE SERVER mysql_user_db
   FOREIGN DATA WRAPPER mysql_fdw
   OPTIONS (host 'your_mysql_host', port '3306', dbname 'your_mysql_database');
   

3. Create a user mapping for MySQL authentication:

   CREATE USER MAPPING FOR your_postgres_user
   SERVER mysql_user_db
   OPTIONS (user 'your_mysql_username', password 'your_mysql_password');
   

4. Map the MySQL users table as a PostgreSQL foreign table:

   CREATE FOREIGN TABLE mysql_users (
       id INT PRIMARY KEY,
       username VARCHAR(50) NOT NULL,
       email VARCHAR(100) UNIQUE NOT NULL,
       created_at TIMESTAMP NOT NULL
   ) SERVER mysql_user_db
   OPTIONS (table_name 'users');
   

5. Create a trigger function to sync inserts:

   CREATE OR REPLACE FUNCTION sync_user_to_mysql()
   RETURNS TRIGGER AS $$
   BEGIN
       -- Handle duplicates with MySQL's ON DUPLICATE KEY UPDATE logic
       INSERT INTO mysql_users (id, username, email, created_at)
       VALUES (NEW.id, NEW.username, NEW.email, NEW.created_at)
       ON CONFLICT (id) DO UPDATE SET
           username = EXCLUDED.username,
           email = EXCLUDED.email,
           created_at = EXCLUDED.created_at;
       RETURN NEW;
   END;
   $$ LANGUAGE plpgsql;
   

6. Attach the trigger to your PostgreSQL users table:

   CREATE TRIGGER trigger_sync_new_user
   AFTER INSERT ON users
   FOR EACH ROW
   EXECUTE FUNCTION sync_user_to_mysql();
   

Notes:

• This enforces strong consistency: if the MySQL sync fails, the original PostgreSQL insert will roll back.
• Best suited for low-to-moderate write volumes—high concurrency may still impact PostgreSQL performance.

Option 2: Asynchronous Message Queue + Consumer (Scalable, High Performance) #

For high-traffic systems, decouple the sync process using a message queue. The PostgreSQL trigger sends a sync message to a queue, and an independent consumer service handles inserting into MySQL.

Step-by-Step Implementation:

1. Use PostgreSQL’s built-in pg_notify for a lightweight queue:
   • Create a trigger function to send sync notifications:

     CREATE OR REPLACE FUNCTION queue_user_sync()
     RETURNS TRIGGER AS $$
     BEGIN
         PERFORM pg_notify(
             'user_sync_channel',
             json_build_object(
                 'id', NEW.id,
                 'username', NEW.username,
                 'email', NEW.email,
                 'created_at', NEW.created_at
             )::TEXT
         );
         RETURN NEW;
     END;
     $$ LANGUAGE plpgsql;
     

   • Attach the trigger:

     CREATE TRIGGER trigger_queue_sync
     AFTER INSERT ON users
     FOR EACH ROW
     EXECUTE FUNCTION queue_user_sync();
     

2. Build a consumer script (example in Python):
   This script listens for notifications and syncs data to MySQL:

   import psycopg2
   from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
   import mysql.connector
   import json
   import logging
   
   # Configure logging for failed syncs
   logging.basicConfig(filename='sync_errors.log', level=logging.ERROR)
   
   # Connect to PostgreSQL
   pg_conn = psycopg2.connect(
       dbname='your_postgres_db',
       user='your_postgres_user',
       password='your_postgres_pass',
       host='your_postgres_host'
   )
   pg_conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
   pg_cursor = pg_conn.cursor()
   pg_cursor.execute("LISTEN user_sync_channel;")
   
   # Connect to MySQL
   mysql_conn = mysql.connector.connect(
       host='your_mysql_host',
       user='your_mysql_user',
       password='your_mysql_pass',
       database='your_mysql_db'
   )
   mysql_cursor = mysql_conn.cursor()
   
   print("Waiting for user sync messages...")
   while True:
       pg_conn.poll()
       while pg_conn.notifies:
           notify = pg_conn.notifies.pop(0)
           try:
               user_data = json.loads(notify.payload)
               # Insert/update MySQL record
               query = """
               INSERT INTO users (id, username, email, created_at)
               VALUES (%s, %s, %s, %s)
               ON DUPLICATE KEY UPDATE
                   username=%s, email=%s, created_at=%s
               """
               values = (
                   user_data['id'], user_data['username'], user_data['email'], user_data['created_at'],
                   user_data['username'], user_data['email'], user_data['created_at']
               )
               mysql_cursor.execute(query, values)
               mysql_conn.commit()
               print(f"Synchronized user {user_data['id']} successfully")
           except Exception as e:
               logging.error(f"Failed to sync user {user_data.get('id')}: {str(e)}")
               # Optional: Add retry logic or requeue the message
   

Advantages:

• No impact on PostgreSQL performance: Triggers finish instantly without waiting for MySQL.
• Resilience: If the consumer goes down, messages stay in the queue (or use a persistent queue like Kafka/RabbitMQ for better durability) and can be processed when it restarts.
• Scalability: You can run multiple consumers to handle high sync volumes, or add more targets (e.g., another database) without modifying PostgreSQL.

Final Recommendations #

• Use Option 1 if you need strict consistency between the two databases and have low write volumes.
• Use Option 2 for high-traffic systems where main database performance is a priority, or if you need to sync to multiple targets.
• Regardless of the approach, add error logging and retry logic to handle edge cases like network outages or constraint violations.

内容的提问来源于stack exchange，提问作者Mariano Serravalle