elasticsearch rest客户端连接工具类

package cn.hong.elastic.util; import cn.hong.elastic.model.BulkVo; import net.sf.json.JSON; import net.sf.json.JSONObject; import org.apache.commons.lang3.StringUtils; import org.apache.http.HttpEntity; import org.apache.http.HttpHost; import org.apache.http.auth.AuthScope; import org.apache.http.auth.UsernamePasswordCredentials; import org.apache.http.client.CredentialsProvider; import org.apache.http.client.config.RequestConfig; import org.apache.http.entity.ContentType; import org.apache.http.impl.client.BasicCredentialsProvider; import org.apache.http.impl.nio.client.HttpAsyncClientBuilder; import org.apache.http.nio.entity.NStringEntity; import org.apache.http.util.EntityUtils; import org.elasticsearch.client.Response; import org.elasticsearch.client.ResponseException; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestClientBuilder; import java.io.IOException; import java.util.*; /** * * elasticsearch rest 工具类 * * @author hong */ public class RestUtil { // RestClient对象，用于es链接 private static volatile RestClient client; public static final String PUT = "PUT"; public static final String POST = "POST"; public static final String GET = "GET"; public static final String HEAD = "HEAD"; public static final String DELETE = "DELETE"; public static final String HANDLE_INDEX = "index"; public static final String HANDLE_UPDATE = "update"; public static final String HANDLE_DELETE = "delete"; public static final String HANDLE_CREATE = "create"; /** * 获取Rest Client 对象 */ public static RestClient getClient() { if(null != client) { return client; } try { // credentialsProvider为用户密码验证， 如阿里云的elasticsearch 需要用户密码验证， // 若不需要则注释掉credentialsProvider对象及setHttpClientConfigCallback final CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("elastic", "elasic")); // HttpHost可多个，直接逗号隔开即可 client = RestClient.builder(new HttpHost("localhost", 9200)) .setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() { @Override public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) { // 设置建立连接超时时间 客户端从服务器读取数据的timeout requestConfigBuilder.setConnectTimeout(5000); requestConfigBuilder.setSocketTimeout(40000); requestConfigBuilder.setConnectionRequestTimeout(3000); return requestConfigBuilder; } }) .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() { @Override public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) { return httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider); } }) .setMaxRetryTimeoutMillis(60000) .build(); } catch (Exception e) { e.printStackTrace(); } return client; } /** * 创建查询字符串 * @param fields 要显示的字段名数组 null为显示全部 * @param query 查询规则 * @param pageNo 页码 * @param pageSize 每页数量 * @param sort 排序规则 * @return */ public static String createQuery(List<String> fields, Object query, Integer pageNo, Integer pageSize, Object sort) { Map<String, Object> map = new HashMap<>(); if(null != fields && !fields.isEmpty()) { map.put("_source", fields); } if(null != query) { map.put("query", JSONObject.fromObject(query)); } if(null != pageNo && null != pageSize) { map.put("from", (pageNo - 1) * pageSize); map.put("size", pageSize); } if(null != sort) { map.put("sort", JSONObject.fromObject(sort)); } return JSONObject.fromObject(map).toString(); } /** * 根据sql直接搜索 仅限于 select查询 * 若要使用sql直接select查询，需安装sql插件 https://github.com/NLPchina/elasticsearch-sql * 另 http://essql.nlpcn.org/ 该网址可将select sql语句转化为 elastic 写法的查询 */ public static String searchSql(String sql) { try { Response response = getClient().performRequest(POST, "/_sql", Collections.singletonMap("sql", sql)); return EntityUtils.toString(response.getEntity()); } catch (ResponseException e) { e.printStackTrace(); } catch (IOException e) { e.printStackTrace(); } return null; } /** * 单索引查询 */ public static String search(String index, String type, String query) { List<String> indexs = new ArrayList<>(); indexs.add(index); List<String> types = new ArrayList<>(); types.add(type); return search(indexs, types, query); } /** * 查询 可多索引 多类型 查询 * @param indexs * @param types * @param query 查询条件json字符串 * @return */ public static String search(List<String> indexs, List<String> types, String query) { try { String index = StringUtils.join(indexs, ","); String type = "/"; if(null != types && !types.isEmpty()) { type += StringUtils.join(types, ",") + "/"; } System.out.println("/" + index + type + "_search"); System.out.println(query); HttpEntity entity = new NStringEntity(query, ContentType.APPLICATION_JSON); Response response = getClient().performRequest(POST, "/" + index + type + "_search", Collections.singletonMap("pretty", "true"), entity); return EntityUtils.toString(response.getEntity()); } catch (ResponseException e) { e.printStackTrace(); } catch (IOException e) { e.printStackTrace(); } return null; } /** * 批量操作 最后一行也要带\n * create 当文档不存在时自动创建 * index 创建新文档或更新已有文档 * update 局部更新文档 * delete 删除一个文档 * * POST /_bulk * { "delete": { "_index": "website", "_type": "blog", "_id": "123" }} * { "create": { "_index": "website", "_type": "blog", "_id": "123" }} * { "title": "My first blog post" } * { "index": { "_index": "website", "_type": "blog" }} * { "title": "My second blog post" } * { "update": { "_index": "website", "_type": "blog", "_id": "123", "_retry_on_conflict" : 3} } * { "doc" : {"title" : "My updated blog post"} } * */ public static String sendBulk(List<BulkVo> list) { try { List<String> bodys = new ArrayList<>(); for (BulkVo bean : list) { bodys.add("{ \"" + bean.getHandle() + "\": {\"_index\": \"" + bean.getIndexName() + "\", \"_type\": \"" + bean.getType() + "\", \"_id\": \"" + bean.getId() + "\"} }"); if(HANDLE_CREATE.equals(bean.getHandle()) || HANDLE_INDEX.equals(bean.getHandle())) { bodys.add