Oracle跨库数据同步：从A库a_table查询插入B库b_table方案咨询

Alright, let's tackle your cross-Oracle database sync problem—no third-party scripts/programs required, and we can set it up to run automatically either on a schedule or when new data hits table A. Here are the two solid native Oracle approaches:

This uses Oracle's built-in DBMS_SCHEDULER combined with a database link (DB Link) to connect the two databases, plus a stored procedure to handle the data transfer.

Step 1: Create a Database Link from DB B to DB A #

First, in Database B, create a link pointing to Database A so it can query a_table:

CREATE DATABASE LINK db_link_to_A 
CONNECT TO your_A_username 
IDENTIFIED BY your_A_password 
USING 'tns_alias_for_DB_A'; -- Make sure this alias exists in B's tnsnames.ora

Note: Ensure the user in DB B has SELECT privileges on a_table in DB A, and the user in DB A has granted those privileges.

Step 2: Build a Sync Stored Procedure #

Create a procedure in DB B to fetch filtered records from a_table and insert them into b_table. We'll add optional logging to track syncs and errors:

CREATE OR REPLACE PROCEDURE sync_a_to_b AS
  PRAGMA AUTONOMOUS_TRANSACTION; -- Optional: Isolate sync transaction from other operations
BEGIN
  -- Insert only the records you need (adjust the WHERE clause to your filter rules)
  INSERT INTO b_table (col1, col2, col3, created_at)
  SELECT col1, col2, col3, created_at
  FROM a_table@db_link_to_A
  WHERE created_at > NVL(
    (SELECT MAX(last_sync_timestamp) FROM sync_log), 
    TO_DATE('1970-01-01', 'YYYY-MM-DD')
  ); -- Sync only new records since last run

  -- Log the sync details
  INSERT INTO sync_log (sync_timestamp, records_inserted)
  VALUES (SYSDATE, SQL%ROWCOUNT);

  COMMIT;
EXCEPTION
  WHEN OTHERS THEN
    ROLLBACK;
    -- Log errors for debugging
    INSERT INTO sync_error_log (error_timestamp, error_message)
    VALUES (SYSDATE, SQLERRM);
    COMMIT;
END;
/

Tip: If b_table has unique keys, use MERGE instead of INSERT to avoid duplicate records (e.g., MERGE INTO b_table b USING (SELECT ... FROM a_table@db_link_to_A) a ON (b.id = a.id) WHEN NOT MATCHED THEN INSERT (...) VALUES (...);).

Step 3: Schedule the Procedure with DBMS_SCHEDULER #

Set up a recurring job to run the sync every 30 minutes:

BEGIN
  DBMS_SCHEDULER.CREATE_JOB (
    job_name        => 'SYNC_A_TO_B_SCHEDULED',
    job_type        => 'STORED_PROCEDURE',
    job_action      => 'sync_a_to_b',
    start_date      => SYSTIMESTAMP,
    repeat_interval => 'FREQ=MINUTELY;INTERVAL=30', -- Runs every 30 minutes
    enabled         => TRUE,
    comments        => 'Sync filtered records from A.a_table to B.b_table every 30 minutes'
  );
END;
/

You can adjust the repeat_interval to fit your needs (e.g., FREQ=HOURLY;INTERVAL=1 for hourly runs).

If you want to sync records immediately when they're inserted into a_table, use an Oracle trigger combined with a DB Link. Note: Be cautious with performance for bulk inserts—we'll cover an optimized approach too.

Step 1: Create a Database Link from DB A to DB B #

In Database A, create a link to Database B so it can push data to b_table:

CREATE DATABASE LINK db_link_to_B 
CONNECT TO your_B_username 
IDENTIFIED BY your_B_password 
USING 'tns_alias_for_DB_B';

Ensure the user in DB A has INSERT privileges on b_table in DB B.

Step 2: Option A: Row-Level Trigger (Simple, for Low-Volume Inserts) #

Create an AFTER INSERT trigger on a_table that pushes qualifying records to b_table immediately:

CREATE OR REPLACE TRIGGER trg_a_table_sync_on_insert
AFTER INSERT ON a_table
FOR EACH ROW
WHEN (-- Filter the records you want to sync (adjust conditions as needed)
      :NEW.status = 'APPROVED' AND :NEW.data_type = 'CRITICAL')
DECLARE
  PRAGMA AUTONOMOUS_TRANSACTION; -- Isolate sync to avoid blocking the original insert
BEGIN
  INSERT INTO b_table@db_link_to_B (col1, col2, col3, source_id)
  VALUES (:NEW.col1, :NEW.col2, :NEW.col3, :NEW.id);
  
  COMMIT;
EXCEPTION
  WHEN OTHERS THEN
    ROLLBACK;
    -- Log errors without failing the original insert
    INSERT INTO a_sync_error_log (error_time, source_record_id, error_msg)
    VALUES (SYSDATE, :NEW.id, SQLERRM);
    COMMIT;
END;
/

Step 2: Option B: Asynchronous Queue-Based Trigger (High-Volume Inserts) #

For bulk insert scenarios, a row-level trigger can cause performance bottlenecks. Instead, use Oracle Advanced Queuing (AQ) to queue sync requests and process them in the background:

1. Create a queue in DB A:

   BEGIN
     DBMS_AQADM.CREATE_QUEUE_TABLE(
       queue_table        => 'sync_queue_table',
       queue_payload_type => 'SYS.XMLTYPE'
     );
     DBMS_AQADM.CREATE_QUEUE(
       queue_name         => 'sync_queue',
       queue_table        => 'sync_queue_table'
     );
     DBMS_AQADM.START_QUEUE(queue_name => 'sync_queue');
   END;
   /
   

2. Trigger to enqueue messages:

   CREATE OR REPLACE TRIGGER trg_a_table_enqueue_sync
   AFTER INSERT ON a_table
   FOR EACH ROW
   WHEN (:NEW.status = 'APPROVED' AND :NEW.data_type = 'CRITICAL')
   DECLARE
     l_payload SYS.XMLTYPE;
   BEGIN
     l_payload := SYS.XMLTYPE.CREATEXML(
       '<sync_record>
          <col1>' || :NEW.col1 || '</col1>
          <col2>' || :NEW.col2 || '</col2>
          <col3>' || :NEW.col3 || '</col3>
        </sync_record>'
     );
     DBMS_AQ.ENQUEUE(
       queue_name => 'sync_queue',
       enqueue_options => DBMS_AQ.ENQUEUE_OPTIONS_T(),
       message_properties => DBMS_AQ.MESSAGE_PROPERTIES_T(),
       payload => l_payload,
       msgid => NULL
     );
   END;
   /
   

3. Create a consumer procedure to process the queue:

   CREATE OR REPLACE PROCEDURE process_sync_queue AS
     dequeue_options DBMS_AQ.DEQUEUE_OPTIONS_T;
     message_properties DBMS_AQ.MESSAGE_PROPERTIES_T;
     msgid RAW(16);
     l_payload SYS.XMLTYPE;
     PRAGMA AUTONOMOUS_TRANSACTION;
   BEGIN
     LOOP
       DBMS_AQ.DEQUEUE(
         queue_name => 'sync_queue',
         dequeue_options => dequeue_options,
         message_properties => message_properties,
         payload => l_payload,
         msgid => msgid
       );
       
       -- Insert into B's b_table
       INSERT INTO b_table@db_link_to_B (col1, col2, col3)
       VALUES (
         l_payload.EXTRACT('/sync_record/col1/text()').GETSTRINGVAL(),
         l_payload.EXTRACT('/sync_record/col2/text()').GETSTRINGVAL(),
         l_payload.EXTRACT('/sync_record/col3/text()').GETSTRINGVAL()
       );
       
       COMMIT;
     END LOOP;
   EXCEPTION
     WHEN NO_DATA_FOUND THEN
       COMMIT;
     WHEN OTHERS THEN
       ROLLBACK;
       INSERT INTO a_sync_error_log (error_time, error_msg)
       VALUES (SYSDATE, SQLERRM);
       COMMIT;
   END;
   /
   

4. Schedule the consumer procedure to run continuously:
   Use DBMS_SCHEDULER to start the procedure and keep it running (or set it to run at short intervals).

• Distributed Transactions: Cross-database operations use distributed transactions by default. If you want to avoid tying the sync transaction to the original insert/update, use PRAGMA AUTONOMOUS_TRANSACTION to isolate it.
• Security: Restrict DB Link permissions to only what's necessary (e.g., don't use a superuser account for the link).
• Error Handling: Always log sync failures—this is critical for debugging when records don't sync as expected.
• Performance: For large datasets, prioritize incremental syncs (using timestamps or auto-increment IDs) over full table scans. For trigger-based syncs, use queues for high-volume inserts to avoid blocking.

内容的提问来源于stack exchange，提问作者Ishara Kularatna