我正在尝试使用JEST添加对Titan ES 客户端的 Elasticsearch HTTP 访问。Titan-es 仅支持 ES 的本地和传输 (TCP) 模式。但我想支持通过 ES 的 HTTP 接口进行通信。这将允许客户端库喜欢使用AWS Elasticsearch作为仅提供 HTTP(S) 接口的索引后端。有关更多信息,请参阅此帖子。titan-es
我正在寻找一些关于我目前正在考虑的方法的反馈:
- 创建一个
ElasticsearchHttpClient
实现org.elasticache.client.Client
接口的新类。新类将使用它JestClient
作为它的内部客户端。这样它将通过 HTTP 与 ES 通信。新类可能会扩展 ES'AbstractClient
以减少必须实现的方法:admin()
、settings()
、execute()
、threadPool()
和close()
。 - 添加一个新的枚举
HTTP_CLIENT
到ElasticSearchSetup
- 确保
connect()
on 方法HTTP_CLIENT
返回一个实例,该实例包含和Connection
的正确值。该成员将是新类的一个实例。node
client
client
ElasticsearchHttpClient
- 如果配置为 ,请确保该方法检索(包含新的)
ElasticSearchIndex.interfaceConfiguration()
的正确实例。从那时起,其余代码应该继续在新协议上工作。Connection
ElasticsearchHttpClient
INTERFACE
HTTP_CLIENT
这听起来应该有效吗?第一步是我最关心的问题——我不相信我可以使用 JestClient 实现所有客户端方法。
package com.thinkaurelius.titan.diskstorage.es;
import io.searchbox.client.JestClient;
import io.searchbox.client.JestClientFactory;
import io.searchbox.client.JestResult;
import io.searchbox.client.config.HttpClientConfig;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.*;
import org.elasticsearch.client.AdminClient;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.support.AbstractClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.threadpool.ThreadPool;
import java.io.IOException;
public class ElasticsearchHttpClient extends AbstractClient {
private final JestClient internalClient;
private final ThreadPool pool;
public ElasticsearchHttpClient(String hostname, int port) {
JestClientFactory factory = new JestClientFactory();
factory.setHttpClientConfig(new HttpClientConfig
.Builder(String.format("http://%s:%d", hostname, port))
.multiThreaded(true)
.build());
JestClient client = factory.getObject();
this.pool = new ThreadPool("jest");
this.internalClient = client;
}
@Override
public AdminClient admin() {
return null;
}
@Override
public Settings settings() {
return null;
}
@Override
public <Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder, Client>> ActionFuture<Response> execute(Action<Request, Response, RequestBuilder, Client> action, Request request) {
try {
JestResult response = internalClient.execute(convertRequest(action, request));
return convertResponse(response);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
@Override
public <Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder, Client>> void execute(Action<Request, Response, RequestBuilder, Client> action, Request request, ActionListener<Response> listener) {
execute(action, request);
}
private <Response extends ActionResponse> ActionFuture<Response> convertResponse(JestResult result) {
// TODO How to convert a JestResult a Elasticsearch ActionResponse/ActionFuture?
return null;
}
private <Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder, Client>> io.searchbox.action.Action<JestResult> convertRequest(Action<Request, Response, RequestBuilder, Client> action, Request request) {
// TODO How to convert an Elasticsearch Action<..> and Request to a Jest Action<JestResult>?
return null;
}
@Override
public ThreadPool threadPool() {
return pool;
}
@Override
public void close() throws ElasticsearchException {
pool.shutdownNow();
}
}
[我也在Titan 邮件列表和Elasticsearch 论坛上问过这个问题。]