基于Apache、MySQL、PHP的Web应用离线开发及CouchDB与MySQL自动同步咨询

Hey there! Let's tackle this bidirectional real-time sync problem between MySQL and CouchDB for your offline PHP app. I’ve worked through similar setups before, so here’s a structured approach to make it work smoothly:

First, we need to cover two critical flows:

1. MySQL data changes syncing to CouchDB (for serving offline clients via PouchDB)
2. CouchDB data changes (including PouchDB syncs from offline users) syncing back to MySQL

Let’s break down each flow with actionable solutions:

1. MySQL → CouchDB Real-Time Sync #

You have two reliable options here, depending on your scale and complexity:

Option A: MySQL Triggers + PHP Sync Worker #

This is a straightforward approach for smaller apps:

• Step 1: Add a sync log table in MySQL to track changes. Create a table like sync_log with columns: id (PK), table_name, primary_key_col, primary_key, operation (INSERT/UPDATE/DELETE), created_at, synced (boolean, default 0).
• Step 2: Create triggers for every table you need to sync. For example, an INSERT trigger on your users table would insert a row into sync_log with the operation details.
• Step 3: Build a PHP sync script that runs either via a cron job (for near-real-time) or a long-running process (for true real-time):
  • The script fetches unsynced rows from sync_log
  • For each row, it pulls the full record from the source MySQL table
  • It pushes the data to CouchDB using its REST API (use PUT for updates, POST for new docs; include a unique _id like mysql_users_123 to map to MySQL records)
  • Marks the log entry as synced once successful

Option B: Debezium CDC (Change Data Capture) #

For larger, high-throughput apps, this is a more robust solution:

• Use Debezium to listen to MySQL’s binlog and capture real-time changes without triggers
• Route these changes to a message broker like Kafka
• Build a PHP consumer service that reads from Kafka and pushes updates to CouchDB
• This avoids adding overhead to MySQL and handles high volumes of changes gracefully

2. CouchDB → MySQL Real-Time Sync #

CouchDB’s built-in Change Feed is the key here—it lets you listen to every document change in real time:

Build a PHP Long-Running Listener #

• Write a PHP script that maintains a continuous connection to CouchDB’s /_changes endpoint (use feed=continuous and include_docs=true to get full document data)
• The script tracks the last processed change sequence ID (store this in a file or MySQL table to resume after restarts)
• When a change is detected:
  • If the document was deleted, delete the corresponding record in MySQL
  • If it’s an update/insert, map the CouchDB document fields to your MySQL table and run an INSERT ... ON DUPLICATE KEY UPDATE query
• Add error handling and retry logic (e.g., if MySQL is down, queue the change for later processing)

Bonus: PouchDB Sync Hooks #

While the CouchDB change feed is the most reliable, you can also add hooks to PouchDB’s sync events (like pouchdb.sync().on('change', ...)) to trigger immediate syncs when an offline user pushes changes. However, always rely on the CouchDB-side listener as the single source of truth to avoid missing updates from multiple clients.

3. Conflict Resolution (Critical!) #

Bidirectional sync will inevitably cause conflicts—here’s how to handle them:

• Version tracking: Add a version integer field to your MySQL tables, and use CouchDB’s built-in _rev field. When syncing, only update the target if the source version is newer.
• Timestamp-based resolution: Add an updated_at field to both MySQL and CouchDB records, and keep the record with the most recent timestamp.
• Conflict logging: For unresolvable conflicts (e.g., two users modified the same field differently), log the conflict to a dedicated table/document for manual review later.
• Define business rules: Decide upfront which system takes priority for specific data types (e.g., user-generated content from PouchDB might override MySQL admin edits, or vice versa).

4. Simplified Code Examples #

MySQL → CouchDB Sync Script (Cron Job) #

<?php
// Initialize connections
$pdo = new PDO('mysql:host=localhost;dbname=your_app', 'db_user', 'db_pass');
$couchClient = new GuzzleHttp\Client(['base_uri' => 'http://couchdb:5984/']);

// Fetch unsynced changes
$stmt = $pdo->query("SELECT * FROM sync_log WHERE synced = 0 ORDER BY created_at ASC");
while ($logEntry = $stmt->fetch(PDO::FETCH_ASSOC)) {
    $docId = "mysql_{$logEntry['table_name']}_{$logEntry['primary_key']}";
    try {
        switch ($logEntry['operation']) {
            case 'INSERT':
            case 'UPDATE':
                // Get full record from MySQL
                $dataStmt = $pdo->query("SELECT * FROM {$logEntry['table_name']} WHERE {$logEntry['primary_key_col']} = {$logEntry['primary_key']}");
                $docData = $dataStmt->fetch(PDO::FETCH_ASSOC);
                $docData['_id'] = $docId;
                // Push to CouchDB
                $couchClient->put("your_couch_db/{$docId}", ['json' => $docData]);
                break;
            case 'DELETE':
                // Delete from CouchDB (need current _rev first)
                $docResponse = $couchClient->get("your_couch_db/{$docId}");
                $doc = json_decode($docResponse->getBody(), true);
                $couchClient->delete("your_couch_db/{$docId}?rev={$doc['_rev']}");
                break;
        }
        // Mark log entry as synced
        $pdo->exec("UPDATE sync_log SET synced = 1 WHERE id = {$logEntry['id']}");
    } catch (Exception $e) {
        error_log("Sync failed for log ID {$logEntry['id']}: " . $e->getMessage());
    }
}
?>

CouchDB → MySQL Listener (Long-Running Process) #

<?php
use GuzzleHttp\Client;

$couchClient = new Client(['base_uri' => 'http://couchdb:5984/']);
$pdo = new PDO('mysql:host=localhost;dbname=your_app', 'db_user', 'db_pass');
$lastSeq = file_exists('last_seq.txt') ? (int)file_get_contents('last_seq.txt') : 0;

while (true) {
    try {
        $response = $couchClient->get("your_couch_db/_changes", [
            'query' => [
                'feed' => 'continuous',
                'since' => $lastSeq,
                'include_docs' => 'true'
            ],
            'timeout' => 300
        ]);

        $body = $response->getBody();
        while (!$body->eof()) {
            $line = trim($body->readLine());
            if (empty($line)) continue;
            
            $change = json_decode($line, true);
            $doc = $change['doc'];
            $lastSeq = $change['seq'];

            // Map CouchDB doc to MySQL table
            if (str_starts_with($doc['_id'], 'mysql_')) {
                list(, $tableName, $primaryKey) = explode('_', $doc['_id'], 3);
                $primaryKeyCol = 'id'; // Adjust to your table's PK column
                unset($doc['_id'], $doc['_rev'], $doc['_deleted']);

                if (isset($change['deleted']) && $change['deleted']) {
                    $pdo->exec("DELETE FROM {$tableName} WHERE {$primaryKeyCol} = {$primaryKey}");
                } else {
                    $columns = implode(', ', array_keys($doc));
                    $placeholders = ':' . implode(', :', array_keys($doc));
                    $updateClause = implode(', ', array_map(fn($col) => "{$col} = :{$col}", array_keys($doc)));
                    
                    $stmt = $pdo->prepare("INSERT INTO {$tableName} ({$columns}) VALUES ({$placeholders}) ON DUPLICATE KEY UPDATE {$updateClause}");
                    $stmt->execute($doc);
                }
            }

            // Save last processed sequence
            file_put_contents('last_seq.txt', $lastSeq);
        }
    } catch (Exception $e) {
        error_log("CouchDB feed error: " . $e->getMessage());
        sleep(5); // Retry after 5 seconds
    }
}
?>

5. Pro Tips #

• Test rigorously: Simulate offline edits, network drops, and concurrent changes in a staging environment before going live.
• Use a queue: For high-volume syncs, add a queue (like Redis Queue) to decouple sync tasks from your main app and handle retries more efficiently.
• Monitor sync health: Set up alerts for failed syncs, lag time, and conflict rates to catch issues early.

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