在REST服务中异步返回ElasticSearch数据的实现难题

我帮你排查下代码里的几个关键问题，这些应该就是导致你没法把ElasticSearch的数据返回给客户端的原因：

1. GetResponse无法直接序列化为JSON #

你现在直接把ElasticSearch的GetResponse对象传给Response.ok()，但JAX-RS默认的序列化器根本不认识这个ES原生对象，自然没法把它转成JSON返回给客户端。你需要先提取GetResponse里的实际业务数据，转换成JAX-RS能处理的格式（比如Vert.x的JsonObject）。

修改ElasticResource里的订阅逻辑：

single.subscribe(
    mex -> {
        GetResponse esResponse = mex.body();
        // 提取ES返回的source数据，转成JsonObject
        JsonObject data = new JsonObject(esResponse.getSourceAsString());
        inAsyncResponse.resume(Response.ok(data).build());
    },
    error -> {
        // 一定要处理错误，不然客户端会一直挂着等响应
        inAsyncResponse.resume(Response.serverError().entity("查询失败：" + error.getMessage()).build());
    }
);

2. Single缺少泛型声明，类型不安全 #

在QuarkusElasticServiceImpl的getReq方法里，你返回的是无泛型的Single，这会导致类型擦除，事件总线传递数据时可能出现类型匹配问题，进而影响后续的响应处理。

修改方法签名：

@Override
@ConsumeEvent("QuarkusElasticService.getReq")
public Single<GetResponse> getReq(JsonObject jsonObject) {
    // 原有逻辑不变
    GetRequest getRequest = new GetRequest(
            jsonObject.getString("index"),
            jsonObject.getString("id"));
    return esClient.rxGetAsync(getRequest, RequestOptions.DEFAULT);
}

3. 缺失错误处理逻辑 #

你原来的代码只处理了请求成功的情况，完全没考虑ES查询失败、事件总线传递出错这些异常场景。一旦出现错误，客户端的AsyncResponse会一直处于挂起状态，永远得不到响应。所以必须在subscribe方法里加上错误分支的处理，就像第一点里的代码那样。

4. （可选但推荐）添加ES客户端资源清理 #

在Quarkus应用中，最好加上@PreDestroy方法来关闭ElasticSearch客户端，避免应用关闭时出现资源泄漏：

import javax.annotation.PreDestroy;

// ... 其他代码 ...

@PreDestroy
public void cleanupEsClient() {
    if (esClient != null) {
        esClient.close();
    }
}

修改后的完整代码示例 #

ElasticResource #

import io.reactivex.Single;
import io.vertx.core.json.JsonObject;
import io.vertx.reactivex.core.eventbus.EventBus;
import io.vertx.reactivex.core.eventbus.Message;
import org.elasticsearch.action.get.GetResponse;
import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.container.AsyncResponse;
import javax.ws.rs.container.Suspended;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;

@Path("/elastic")
@ApplicationScoped
public class ElasticResource {
    @Inject
    EventBus eventBus;

    @GET
    @Produces(MediaType.APPLICATION_JSON)
    @Path("bank-es")
    public void greetingVertx(@Suspended final AsyncResponse inAsyncResponse) {
        Single<Message<GetResponse>> single = eventBus.<GetResponse>rxSend("QuarkusElasticService.getReq", 
            new JsonObject().put("index", "bank").put("id", "1"));
        
        single.subscribe(
            mex -> {
                GetResponse esResponse = mex.body();
                JsonObject data = new JsonObject(esResponse.getSourceAsString());
                inAsyncResponse.resume(Response.ok(data).build());
            },
            error -> {
                inAsyncResponse.resume(Response.serverError().entity("查询失败：" + error.getMessage()).build());
            }
        );
    }
}

QuarkusElasticServiceImpl #

import com.sourcesense.sisal.socialbetting.dev.example.elastic.service.QuarkusElasticService;
import io.quarkus.vertx.ConsumeEvent;
import io.reactiverse.elasticsearch.client.reactivex.RestHighLevelClient;
import io.reactivex.Single;
import io.vertx.core.json.JsonObject;
import io.vertx.reactivex.core.Vertx;
import org.apache.http.HttpHost;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.inject.Inject;
import java.util.concurrent.ExecutorService;

public class QuarkusElasticServiceImpl implements QuarkusElasticService {
    @Inject
    Vertx vertx;
    @Inject
    ExecutorService executor;
    private RestHighLevelClient esClient;

    @PostConstruct
    public void init() {
        RestClientBuilder builder = RestClient.builder(
                new HttpHost("localhost", 9200, "http"),
                new HttpHost("localhost", 9201, "http"));
        esClient = RestHighLevelClient.create(vertx, builder);
    }

    @Override
    @ConsumeEvent("QuarkusElasticService.getReq")
    public Single<GetResponse> getReq(JsonObject jsonObject) {
        GetRequest getRequest = new GetRequest(
                jsonObject.getString("index"),
                jsonObject.getString("id"));
        return esClient.rxGetAsync(getRequest, RequestOptions.DEFAULT);
    }

    @PreDestroy
    public void cleanupEsClient() {
        if (esClient != null) {
            esClient.close();
        }
    }
}

这些修改应该能解决你的问题，核心就是把ES的原生响应转换成可序列化的JSON结构，同时完善错误处理和类型安全，确保客户端能收到正常的响应或者错误提示。

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