可控批量写入需用channel缓冲+固定worker协程：channel容量设为bulk上限的2–3倍，worker数不超过es bulk线程数，并加超时避免阻塞。
goroutine + channel 实现可控批量写入
直接用 goroutine 起一堆 client.Index() 是最常见也最危险的做法：没节制的并发会打爆 ES 连接数，还可能触发 bulk queue rejected。真正可控的异步批量，得靠 channel 做缓冲 + goroutine 消费者模型。
核心是建一个带缓冲的 channel 存文档，再起固定数量的 worker 从 channel 取数据、攒够一批就调 Bulk()：
- channel 容量建议设为单次 bulk 上限的 2–3 倍（比如 bulk 最多 1000 条，channel 缓存 2000–3000）
- worker 数量别超过 ES 集群的可用 bulk 线程数（
thread_pool.bulk.queue_size默认 200，别盲目开 50 个 goroutine） - 必须加超时控制：channel 写入不能无限阻塞，用
select { case ch
手动拼接 Bulk 请求体时的 NDJSON 格式陷阱
Go 自带 json.Encoder 默认每条输出都换行，但 Elasticsearch 的 /_bulk 接口要求 action line 和 document line 必须严格交替、且中间不能有空行——否则直接返回 400 Bad Request 并提示 parsing_exception。
正确做法是用 bytes.Buffer 手动控制换行：
var buf bytes.Buffer
for _, doc := range docs {
// action line（不换行）
json.NewEncoder(&buf).Encode(map[string]map[string]string{
"index": {"_index": "logs", "_id": doc.ID},
})
// document line（不换行）
json.NewEncoder(&buf).Encode(doc.Data)
}
// 整体作为 body 提交
resp, err := es.Bulk(bytes.NewReader(buf.Bytes())).Do(ctx)
注意：json.Encoder.Encode() 自带换行，所以这里不需要额外 buf.WriteString("\n")；但如果用 json.Marshal()，就必须自己补 \n。
批量写入后必须检查 errors 字段
HTTP 状态码 200 完全不可信。Elasticsearch 的 /_bulk 接口只要请求格式合法、网络通，就一定返回 200 —— 即使里面 1000 条有 999 条失败了，errors: true 也会藏在 response body 里。
必须完整解析返回 JSON，并逐项检查 items 中每个操作的 status：
- 状态码 200/201 表示成功
- 400/409/500 等表示失败，需看
error.reason判断是 mapping conflict、id 冲突还是磁盘满 - 特别警惕
429 Too Many Requests：说明 bulk queue 已满，该降速或扩容
漏掉这步，就是静默丢数据，线上很难排查。
写入性能瓶颈往往不在 Go 侧，而在 ES 索引设置
Go 端压测看着很流畅，ES 却持续 reject，大概率是索引配置没调优。两个关键点必须改：
- 写入前临时关闭刷新：
PUT /my_index/_settings { "index.refresh_interval": "-1" }，写完再设回"1s" - 写入前把副本数设为 0：
PUT /my_index/_settings { "index.number_of_replicas": 0 }，写完再恢复
这两步能让写入吞吐翻倍以上。但注意：它们只对新索引或冷索引有效，正在被高频查询的热索引慎用。另外，refresh_interval 设为 -1 后，新写入文档不会自动出现在搜索结果中，得手动调 POST /my_index/_refresh 或等你设回去。
golang免费学习笔记（深入）：立即使用
在学习笔记中，你将探索golang的核心概念和高级技巧！
