如何在同步处理场景下使用Elasticsearch的BulkProcessor

解答 #

• 结论：将concurrentRequests设置为0完全可以实现同步处理的行为，是该场景下的标准解决方案。
• 参数逻辑说明：concurrentRequests用于配置BulkProcessor允许并行执行的bulk请求数，设为0时，同一时间仅会执行一个bulk请求，下一批次的请求会等待上一批请求处理完成、响应返回后才会触发提交，整个处理流程完全同步，不会出现异步并发的情况。
• 补充后可直接运行的Kotlin实现示例：

val processor = BulkProcessor.builder(
    { request, _ ->
        // 调用同步bulk方法完成请求提交
        client.bulk(request, RequestOptions.DEFAULT)
    },
    "custom-bulk-processor" // 可选配置，指定处理器名称方便问题排查
)
.setConcurrentRequests(0) // 核心配置，开启同步处理模式
.setBulkActions(1000) // 可选配置，每积累1000条数据触发一次bulk，可按需调整
.setBulkSize(ByteSizeValue.ofMb(5)) // 可选配置，每积累5MB数据触发一次bulk，可按需调整
.build()

• 同步模式注意事项：
  • 建议自行捕获client.bulk调用抛出的异常，或者在BulkProcessor.Listener的afterBulk回调中处理请求失败的情况，避免数据丢失
  • 进程退出前调用processor.awaitClose(超时时间, 时间单位)，会等待所有已提交的请求全部处理完成后再关闭处理器，不会丢失未执行的请求

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