elasticsearch - 等待elasticsearch完成bulkAsync保存

Question

我正在写写到Elasticsearch的应用程序。我从某个目录读取文件，处理并保存输出。文件很大(数百万行)，因此我需要尽快执行，因此在循环中，我创建了具有1,000个indexRequests的批量请求，并使用bulkAsync方法保存了它们。处理完成后，我要退出应用程序。但是，当我退出时，似乎并非所有处理都已完成，最终我丢失了许多文档。申请完成后，我得到:

2019-10-15T20:00:31,343 ERROR [I/O dispatcher 3] o.m.g.e.i.RequestSender$esBulkSaveListener$1:
org.apache.http.ConnectionClosedException: Connection closed unexpectedly
        at org.apache.http.nio.protocol.HttpAsyncRequestExecutor.closed(HttpAsyncRequestExecutor.java:146)
        at org.apache.http.impl.nio.client.InternalIODispatch.onClosed(InternalIODispatch.java:71)
        at org.apache.http.impl.nio.client.InternalIODispatch.onClosed(InternalIODispatch.java:39)
        at org.apache.http.impl.nio.reactor.AbstractIODispatch.disconnected(AbstractIODispatch.java:100)
        at org.apache.http.impl.nio.reactor.BaseIOReactor.sessionClosed(BaseIOReactor.java:277)
        at org.apache.http.impl.nio.reactor.AbstractIOReactor.processClosedSessions(AbstractIOReactor.java:449)
        at org.apache.http.impl.nio.reactor.AbstractIOReactor.execute(AbstractIOReactor.java:283)
        at org.apache.http.impl.nio.reactor.BaseIOReactor.execute(BaseIOReactor.java:104)
        at org.apache.http.impl.nio.reactor.AbstractMultiworkerIOReactor$Worker.run(AbstractMultiworkerIOReactor.java:591)
        at java.base/java.lang.Thread.run(Thread.java:834)

如果我没有在处理后立即退出应用程序，则看不到此消息，但是看起来我根本没有办法退出它。
有什么方法可以退出前等待Elasticsearch完成吗？

Answer 1

您为什么需要为此编写自定义应用程序？

Logstash或Filebeat确实可以读取文件并与Elasticsearch很好地集成，Logstash可以处理故障和背压。

查看您的错误，可能是您向集群​​发送了太多数据，它无法处理该负载并且崩溃了。我还建议您检查Elasticsearch日志，它们可能会很有用，并提示您不正确的地方。