初次接触es，可能对增删改查很熟悉，以为能为得心应手，本次应用场景为 数据库变更一条记录，会触发更新es中的数据，每秒并发大概30条左右，测试环境一切工作正常（数据量较少），上线后发现日志中很多类似于下面的错误：
队列满了
org.elasticsearch.common.util.concurrent.EsRejectedExecutionException: rejected execution (queue capacity 200) on org.elasticsearch.search.action.SearchServiceTransportAction$23@5f804c60
at org.elasticsearch.common.util.concurrent.EsAbortPolicy.rejectedExecution(EsAbortPolicy.java:62)
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:821)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1372)
at org.elasticsearch.search.action.SearchServiceTransportAction.execute(SearchServiceTransportAction.java:509)
at org.elasticsearch.search.action.SearchServiceTransportAction.sendExecuteScan(SearchServiceTransportAction.java:441)
at org.elasticsearch.action.search.type.TransportSearchScanAction$AsyncAction.sendExecuteFirstPhase(TransportSearchScanAction.java:68)
at org.elasticsearch.action.search.type.TransportSearchTypeAction$BaseAsyncAction.performFirstPhase(TransportSearchTypeAction.java:171)
at org.elasticsearch.action.search.type.TransportSearchTypeAction$BaseAsyncAction.start(TransportSearchTypeAction.java:153)
at org.elasticsearch.action.search.type.TransportSearchScanAction.doExecute(TransportSearchScanAction.java:52)
at org.elasticsearch.action.search.type.TransportSearchScanAction.doExecute(TransportSearchScanAction.java:42)
...
更新版本错误：
Caused by: org.elasticsearch.index.engine.VersionConflictEngineException: [kpi][4] [opportunity][1442415600000]: version conflict, current [5933], provided [5932]
at org.elasticsearch.index.engine.internal.InternalEngine.innerIndex(InternalEngine.java:582) [elasticsearch-1.4.4.jar:]
at org.elasticsearch.index.engine.internal.InternalEngine.index(InternalEngine.java:522) [elasticsearch-1.4.4.jar:]
at org.elasticsearch.index.shard.service.InternalIndexShard.index(InternalIndexShard.java:425) [elasticsearch-1.4.4.jar:]
at org.elasticsearch.action.index.TransportIndexAction.shardOperationOnPrimary(TransportIndexAction.java:193) [elasticsearch-1.4.4.jar:]
经过高手指点，从单次的实时操作改为批量操作，这样做的好处有，减少网路开销，从消息大小，时间，消息数量三个维度来衡量 批量操作的维度，如果数据不是要求非常实时的操作（非常实时的存储应该也不会选择es）,改为批量操作后，错误均修复，大概配置如下。
private BulkProcessor bulkProcessor;
@PostConstruct
public void init() {
this.bulkProcessor = BulkProcessor.builder(
esTransportMainClient.getClient(),
new BulkProcessor.Listener() {
@Override
public void beforeBulk(long executionId, BulkRequest request) {
logger.info("---尝试插入{}条数据---", request.numberOfActions());
}
@Override
public void afterBulk(long executionId,
BulkRequest request, BulkResponse response) {
logger.info("---尝试插入{}条数据成功---", request.numberOfActions());
}
@Override
public void afterBulk(long executionId,
BulkRequest request, Throwable failure) {
logger.error("[es错误]---尝试插入数据失败---", failure);
}
})
.setBulkActions(1000)
.setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB))
.setFlushInterval(TimeValue.timeValueSeconds(5))
.setConcurrentRequests(2)
.build();
}
我这里全局保持一个bulkProcesser就可以维持正常业务了。
每次使用的方法：
bulkProcessor.add(updateRequestBuilder.request());
此processer的含义为如果消息数量到达1000 或者消息大小到大5M 或者时间达到5s 任意条件满足，客户端就会把当前的数据提交到服务端处理。效率很高。