Flask 项目中实现 PostgreSQL 和 Elasticsearch 高效同步-CFANZ编程社区

在 Flask 项目中，要同时更新 PostgreSQL 数据库和 Elasticsearch，通常需要确保数据同步的可靠性和操作的高效性。这涉及到数据一致性管理、批量操作处理，以及对异步任务框架的利用。以下是一些提高效率的关键步骤和方法：

1. 使用异步任务队列

对于需要同时更新多个数据存储的场景，使用异步任务队列可以显著提高效率。可以考虑使用 Celery 配合消息队列（如 Redis 或 RabbitMQ）来异步更新 Elasticsearch（ES）。

方案流程

在业务逻辑中更新 PostgreSQL 数据库。
在提交数据库事务后，触发异步任务，将更新请求发送到 Elasticsearch。
通过 Celery 实现异步任务的分发与重试机制，以保证任务可靠完成。

示例代码

from flask import Flask, request, jsonify
from celery import Celery
from sqlalchemy import create_engine
from elasticsearch import Elasticsearch

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:password@localhost/dbname'
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

# 设置数据库和 ES 连接
db_engine = create_engine(app.config['SQLALCHEMY_DATABASE_URI'])
es = Elasticsearch()

# 配置 Celery
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

@celery.task
def update_elasticsearch(doc_id, data):
    es.index(index='my_index', id=doc_id, body=data)

@app.route('/update', methods=['POST'])
def update_record():
    # 更新数据库
    with db_engine.begin() as conn:
        data = request.json
        # 假设 'id' 和 'content' 是更新的字段
        conn.execute("UPDATE my_table SET content = %s WHERE id = %s", (data['content'], data['id']))
    
    # 提交事务后，异步更新 ES
    update_elasticsearch.delay(data['id'], data)

    return jsonify({'status': 'Update queued for Elasticsearch'})

这种方式确保数据库操作是同步的，而 ES 更新是异步执行的，不会阻塞主业务逻辑。

2. 使用批量操作减少开销

对于需要频繁更新的场景，批量更新操作比逐条更新更高效。Elasticsearch 提供了 _bulk API，可以实现批量索引数据，从而减少网络开销。对于 PostgreSQL，可以使用批量插入或更新操作。

批量操作示例

使用批量操作处理多个更新请求：

收集多条待更新的数据，达到一定数量时批量发送到 PostgreSQL 和 Elasticsearch。
在 Celery 中实现批量任务，定期或当达到指定量时发送批量请求。

示例代码

# 批量任务示例
@celery.task
def batch_update_elasticsearch(records):
    actions = [
        {
            "_op_type": "index",
            "_index": "my_index",
            "_id": record['id'],
            "_source": record
        }
        for record in records
    ]
    es.bulk(actions)

对于 PostgreSQL 可以考虑 execute_batch 或 execute_values，提高批量插入性能：

from psycopg2.extras import execute_values

def batch_update_postgres(records):
    with db_engine.begin() as conn:
        sql = "INSERT INTO my_table (id, content) VALUES %s ON CONFLICT (id) DO UPDATE SET content = excluded.content"
        execute_values(conn, sql, records)

3. 使用事务和消息队列确保数据一致性

为了确保 PostgreSQL 和 Elasticsearch 数据的一致性，可以通过事务和消息队列进行操作。使用事务保证数据库更新成功后，再发送异步更新请求到 Elasticsearch。若事务失败，则不会发出更新请求。

事务和消息队列示例

在 Flask 应用中，开启事务。
当 PostgreSQL 更新完成并成功提交事务后，将更新事件推送到消息队列。
消费者从消息队列中取出任务，并更新 Elasticsearch。

from flask_sqlalchemy import SQLAlchemy
from kombu import Connection, Exchange, Queue, Producer

db = SQLAlchemy(app)
exchange = Exchange('updates', type='direct')
queue = Queue(name='es_updates', exchange=exchange, routing_key='es.update')

@app.route('/update', methods=['POST'])
def update_record():
    data = request.json
    with db.session.begin():
        # 更新 PostgreSQL 数据
        record = MyModel.query.get(data['id'])
        record.content = data['content']
        db.session.commit()
    
        # 事务成功，向消息队列发送任务
        with Connection('redis://localhost:6379/0') as conn:
            producer = Producer(conn)
            producer.publish(data, exchange=exchange, routing_key='es.update', declare=[queue])
    
    return jsonify({'status': 'Update queued for Elasticsearch'})

消费者可以是一个独立的 Python 脚本，处理消息队列的任务，保证事务成功后才更新 Elasticsearch。

4. 数据一致性策略：双写与最终一致性

为了确保 PostgreSQL 和 Elasticsearch 数据的一致性，通常需要考虑以下策略：

双写：在应用逻辑中同时更新 PostgreSQL 和 Elasticsearch，确保两者一致，但会增加代码复杂性。
最终一致性：在业务中允许短时间的不一致性，在后台进程中通过同步脚本定期修复数据偏差。

例如，每隔一定时间对 PostgreSQL 数据库和 Elasticsearch 进行一致性检查，确保数据无偏差。

@app.route('/update', methods=['POST'])
def update_record():
    data = request.json
    with db_engine.begin() as conn:
        conn.execute("UPDATE my_table SET content = %s WHERE id = %s", (data['content'], data['id']))
    
    # 同步更新 Elasticsearch
    es.index(index='my_index', id=data['id'], body=data)
    return jsonify({'status': 'Update applied to PostgreSQL and Elasticsearch'})

5. 定期数据同步：异步或定时器校准

对于数据量较大的场景，即使采用事务和队列机制，依然可能出现一致性偏差。可以定期执行同步作业，确保 PostgreSQL 和 Elasticsearch 数据的一致性。

使用 APScheduler 或 Celery 定期任务执行一次全量同步，将 PostgreSQL 数据与 Elasticsearch 数据进行比对更新：

from apscheduler.schedulers.background import BackgroundScheduler

def sync_pg_to_es():
    with db_engine.connect() as conn:
        result = conn.execute("SELECT id, content FROM my_table")
        actions = [
            {
                "_op_type": "index",
                "_index": "my_index",
                "_id": row['id'],
                "_source": {"content": row['content']}
            }
            for row in result
        ]
    es.bulk(actions)

scheduler = BackgroundScheduler()
scheduler.add_job(sync_pg_to_es, 'interval', hours=24)  # 每天同步一次
scheduler.start()