Elixir Phoenix中SQL数据增改后同步写入Elasticsearch的最佳方案

1 通用同步流程（兼容任意SQL/NoSQL到ES的同步场景） #

• 事务优先：所有业务数据变更（新增/修改/删除帖子、评论）首先在SQL事务内完成落地，绝对不要把ES同步逻辑放在事务内执行，避免ES调用失败导致正常业务事务回滚、阻塞用户请求。
• 异步触发：事务提交成功后再触发异步ES同步任务，不要用同步调用占用用户请求链路。
• 全量聚合：同步任务执行时，直接从SQL查询当前帖子的完整数据+所有关联的评论数据，组装成ES要求的嵌套文档结构，不要依赖变更事件的增量字段拼接，ES底层的嵌套文档更新本质也是重写整个父文档，全量组装的成本和增量更新几乎一致，还能避免多次并发变更导致的文档数据不一致。
• 幂等更新：调用ES的index接口时，带上SQL中帖子的最新更新时间戳作为版本号，ES侧开启外部版本校验，避免旧的变更请求覆盖新的变更结果。
• 容错兜底：异步任务设置指数退避自动重试，同时每天低峰期跑一次离线全量校验同步任务，批量修复ES和SQL不一致的记录，覆盖任务丢失、ES宕机等极端场景。

2 Elixir Phoenix中的逻辑放置位置 #

2.1 变更事件触发层 #

使用Ecto Schema的生命周期钩子实现，不要把触发逻辑散在Controller中，避免多接口修改同一张表时漏触发同步：

• 给Post和CommentSchema添加after_insert、after_update、after_delete钩子，钩子中仅处理事务判断和任务投递逻辑。

注意：Ecto钩子默认在事务内执行，需要搭配Ecto.Repo.after_transaction/2使用，确保只有事务提交成功后才投递任务，避免事务回滚后触发无效同步请求。
示例代码：

schema "comments" do
  belongs_to :post, Post
  # 其他字段定义
  after_insert :dispatch_post_sync_job
  after_update :dispatch_post_sync_job
  after_delete :dispatch_post_sync_job
end

defp dispatch_post_sync_job(comment, _opts) do
  Ecto.Repo.after_transaction(fn _status ->
    # 仅投递帖子ID，不要传递其他业务数据，避免任务执行时读取到旧数据
    Oban.insert(PostESSyncWorker.new(%{post_id: comment.post_id}))
  end)
end

PostSchema也添加同款钩子，自身变更时投递自身的同步任务即可。

2.2 同步逻辑实现层 #

所有ES同步逻辑收敛到独立的异步任务Worker中，不要散在各个业务模块，方便统一维护：
示例代码（以Oban任务队列为例）：

defmodule MyApp.Workers.PostESSyncWorker do
  use Oban.Worker, queue: :es_sync, max_attempts: 5

  @impl true
  def perform(%Oban.Job{args: %{"post_id" => post_id}}) do
    post = MyApp.Posts.get_post_with_all_comments(post_id)

    if post do
      # 组装符合ES索引结构的完整文档
      es_doc = %{
        id: post.id,
        title: post.title,
        content: post.content,
        updated_at: post.updated_at,
        comments: Enum.map(post.comments, fn c -> 
          %{id: c.id, content: c.content, author_id: c.author_id, created_at: c.created_at} 
        end)
      }
      # 调用ES客户端，用帖子的更新时间戳做乐观锁
      MyApp.ESClient.index(%{
        index: "posts", 
        id: post.id, 
        body: es_doc, 
        version: DateTime.to_unix(post.updated_at), 
        version_type: "external"
      })
    else
      # 帖子已删除时同步删除ES文档
      MyApp.ESClient.delete(%{index: "posts", id: post_id})
    end
  end
end

2.3 兜底校验层 #

单独写一个定时任务，每天低峰期遍历所有帖子记录，对比SQL和ES的更新时间，不一致的直接触发同步任务，解决极端场景的一致性问题。

3 方案优势 #

• 逻辑收敛：无论帖子还是评论变更，最终都走同一套同步逻辑，维护成本低，不会出现漏同步情况
• 性能损耗小：异步执行不阻塞用户请求，单次同步仅需一次关联查询就能拿到全量数据，没有多余的IO开销
• 一致性保障：事务后投递+幂等更新+兜底校验三层防护，数据一致性风险极低
• 通用性强：关联结构调整、更换为其他NoSQL数据库时，只需修改组装文档的逻辑即可，整体流程不需要大改

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