包装弹力搜索的异步批量索引
我正在为一个新项目开发原型。其想法是在Elasticsearch中提供一个反应性Spring微服务来批量索引文档。Elasticsearch提供了一个高级Rest客户端，它提供了一个异步方法来批量处理索引请求。异步传递使用侦听器的回调被提到这里。回调以批方式接收索引响应(每个请求)。我试图将此响应作为Flux发送回客户端。我想出了一些基于这篇博客文章的东西。
控制器
@RestController
public class AppController {
@SuppressWarnings("unchecked")
@RequestMapping(value = "/test3", method = RequestMethod.GET)
public Flux<String> index3() {
ElasticAdapter es = new ElasticAdapter();
JSONObject json = new JSONObject();
json.put("TestDoc", "Stack123");
Flux<String> fluxResponse = es.bulkIndex(json);
return fluxResponse;
}ElasticAdapter
@Component
class ElasticAdapter {
String indexName = "test2";
private final RestHighLevelClient client;
private final ObjectMapper mapper;
private int processed = 1;
Flux<String> bulkIndex(JSONObject doc) {
return bulkIndexDoc(doc)
.doOnError(e -> System.out.print("Unable to index {}" + doc+ e));
}
private Flux<String> bulkIndexDoc(JSONObject doc) {
return Flux.create(sink -> {
try {
doBulkIndex(doc, bulkListenerToSink(sink));
} catch (JsonProcessingException e) {
sink.error(e);
}
});
}
private void doBulkIndex(JSONObject doc, BulkProcessor.Listener listener) throws JsonProcessingException {
System.out.println("Going to submit index request");
BiConsumer<BulkRequest, ActionListener<BulkResponse>> bulkConsumer =
(request, bulkListener) ->
client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener);
BulkProcessor.Builder builder =
BulkProcessor.builder(bulkConsumer, listener);
builder.setBulkActions(10);
BulkProcessor bulkProcessor = builder.build();
// Submitting 5,000 index requests ( repeating same JSON)
for (int i = 0; i < 5000; i++) {
IndexRequest indexRequest = new IndexRequest(indexName, "person", i+1+"");
String json = doc.toJSONString();
indexRequest.source(json, XContentType.JSON);
bulkProcessor.add(indexRequest);
}
System.out.println("Submitted all docs
}
private BulkProcessor.Listener bulkListenerToSink(FluxSink<String> sink) {
return new BulkProcessor.Listener() {
@Override
public void beforeBulk(long executionId, BulkRequest request) {
}
@SuppressWarnings("unchecked")
@Override
public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
for (BulkItemResponse bulkItemResponse : response) {
JSONObject json = new JSONObject();
json.put("id", bulkItemResponse.getResponse().getId());
json.put("status", bulkItemResponse.getResponse().getResult
sink.next(json.toJSONString());
processed++;
}
if(processed >= 5000) {
sink.complete();
}
}
@Override
public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
failure.printStackTrace();
sink.error(failure);
}
};
}
public ElasticAdapter() {
// Logic to initialize Elasticsearch Rest Client
}
}我使用FluxSink创建响应流，以便将其发送回客户端。在这一点上，我不知道这是否正确。
我的期望是，调用客户端应该在批处理10 (因为批量处理器处理它的批处理10 - builder.setBulkActions(10); )的响应。我尝试使用Spring客户机来使用端点。但无法解决。这就是我试过的
WebClient
public class FluxClient {
public static void main(String[] args) {
WebClient client = WebClient.create("http://localhost:8080");
Flux<String> responseFlux = client.get()
.uri("/test3")
.retrieve()
.bodyToFlux(String.class);
responseFlux.subscribe(System.out::println);
}
}没有任何东西像我预期的那样打印在控制台上。我试着使用System.out.println(responseFlux.blockFirst());。它在结束时将所有响应打印为一批，而不是在最后的批处理中打印。
如果我的方法是正确的，正确的消费方式是什么？对于我心目中的解决方案，这个客户端将驻留在另一个Webapp中。
注:我对反应堆API的理解是有限的。使用的elasticsearch版本为6.8。
回答 1
Stack Overflow用户
发布于 2020-04-03 01:13:46
因此，对您的代码进行了以下更改。
在ElasticAdapter中，
public Flux<Object> bulkIndex(JSONObject doc) {
return bulkIndexDoc(doc)
.subscribeOn(Schedulers.elastic(), true)
.doOnError(e -> System.out.print("Unable to index {}" + doc+ e));
}在磁通上调用subscribeOn(Scheduler，requestOnSeparateThread)，从https://github.com/spring-projects/spring-framework/issues/21507那里了解到
在FluxClient中，
Flux<String> responseFlux = client.get()
.uri("/test3")
.headers(httpHeaders -> {
httpHeaders.set("Accept", "text/event-stream");
})
.retrieve()
.bodyToFlux(String.class);
responseFlux.delayElements(Duration.ofSeconds(1)).subscribe(System.out::println);添加了“接受”标题作为“文本/事件流”和延迟的Flux元素。
通过以上的更改，可以从服务器实时获得响应。
https://stackoverflow.com/questions/60981954
复制相似问题
腾讯云开发者
