java 集成 elastic search 7.6.2/7.7.0

1
2
3
4
5
<dependency>
	<groupId>org.elasticsearch.client</groupId>
	<artifactId>elasticsearch-rest-high-level-client</artifactId>
	<version>7.6.2</version>
</dependency>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
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.impl.client.BasicCredentialsProvider;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.LinkedList;
import java.util.List;

@Configuration
public class ElasticSearchConfig {

    /**
     * 协议
     */
    @Value("${elasticsearch.schema:http}")
    private String schema;

    /**
     * 集群地址，如果有多个用“,”隔开
     */
    @Value("#{'${elasticsearch.address}'.split(',')}")
    private List<String> addresses;

    @Value("${elasticsearch.certificated:true}")
    private boolean certificated;

    @Value("${elasticsearch.username}")
    private String username;

    @Value("${elasticsearch.password}")
    private String password;

    /**
     * 连接超时时间
     */
    @Value("${elasticsearch.connectTimeout:5000}")
    private int connectTimeout;

    /**
     * Socket 连接超时时间
     */
    @Value("${elasticsearch.socketTimeout:5000}")
    private int socketTimeout;

    /**
     * 获取连接的超时时间
     */
    @Value("${elasticsearch.connectionRequestTimeout:5000}")
    private int connectionRequestTimeout;

    /**
     * 最大连接数
     */
    @Value("${elasticsearch.maxConnectNum:100}")
    private int maxConnectNum;

    /**
     * 最大路由连接数
     */
    @Value("${elasticsearch.maxConnectPerRoute:100}")
    private int maxConnectPerRoute;

    @Bean(name = "restHighLevelClient")
    public RestHighLevelClient restHighLevelClient() {
        // 拆分地址
        List<HttpHost> hostLists = new LinkedList<>();
        for (String address : addresses) {
            String host = address.split(":")[0];
            String port = address.split(":")[1];
            hostLists.add(new HttpHost(host, Integer.parseInt(port), schema));
        }
        // 设置密码
        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));

        // 转换成 HttpHost 数组
        HttpHost[] httpHost = hostLists.toArray(new HttpHost[]{});
        // 构建连接对象
        RestClientBuilder builder = RestClient.builder(httpHost);
        builder.setRequestConfigCallback(requestConfigBuilder -> {
            // 异步连接延时配置
            requestConfigBuilder.setConnectTimeout(connectTimeout);
            requestConfigBuilder.setSocketTimeout(socketTimeout);
            requestConfigBuilder.setConnectionRequestTimeout(connectionRequestTimeout);
            return requestConfigBuilder;
        });

        builder.setHttpClientConfigCallback(httpClientBuilder -> {
            // 异步连接数配置
            httpClientBuilder.setMaxConnTotal(maxConnectNum);
            httpClientBuilder.setMaxConnPerRoute(maxConnectPerRoute);
            // 密码设置
            if (certificated) {
                httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
            }
            return httpClientBuilder;
        });
        return new RestHighLevelClient(builder);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.client.core.CountResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.LinkedList;
import java.util.List;

@Component
public class ElasticSearchDao {

    private static final Logger LOGGER = LoggerFactory.getLogger(ElasticSearchDao.class);

    @Resource
    private RestHighLevelClient restHighLevelClient;

    /**
     * 创建document
     *
     * @param indexRequest index请求
     * @return 创建结果
     */
    public boolean index(IndexRequest indexRequest) {
        try {
            boolean created = true;
            IndexResponse indexResponse = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
            ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
            if (shardInfo.getFailed() > 0) {
                created = false;
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    LOGGER.error("index fail, cause: ", failure.getCause());
                }
            }
            return created;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 批量创建document
     *
     * @return 创建成功数量
     * @date 2020/6/18 10:00
     */
    public List<String> bulkIndex(BulkRequest bulkRequest) {
        List<String> list = new LinkedList<>();
        try {
            BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            for (BulkItemResponse bulkItemResponse : bulkResponse) {
                if (bulkItemResponse.isFailed()) {
                    BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                    LOGGER.error("index bulkIndex, cause: ", failure.getCause());
                } else {
                    list.add(bulkItemResponse.getId());
                }
            }
            return list;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 获取
     *
     * @param getRequest get请求
     * @return source
     */
    public String get(GetRequest getRequest) {
        try {
            GetResponse getResponse = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);
            if (!getResponse.isExists()) {
                return null;
            }
            return getResponse.getSourceAsString();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 计数
     *
     * @param countRequest count请求
     * @return 数量
     */
    public long count(CountRequest countRequest) {
        try {
            CountResponse countResponse = restHighLevelClient.count(countRequest, RequestOptions.DEFAULT);
            return countResponse.getCount();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 搜索
     *
     * @param searchRequest search请求
     * @return 文档
     */
    public SearchHits search(SearchRequest searchRequest) {
        try {
            SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            return searchResponse.getHits();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    /**
     * 删除API
     */
    public boolean delete(DeleteRequest deleteRequest) {
        boolean delete = false;
        try {
            DeleteResponse deleteResponse = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
            DocWriteResponse.Result result = deleteResponse.getResult();
            if (DocWriteResponse.Result.DELETED.equals(result)) {
                delete = true;
            }
            return delete;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 按查询删除
     */
    public void deleteByQuery(DeleteByQueryRequest deleteByQueryRequest) {
        try {
            restHighLevelClient.deleteByQuery(deleteByQueryRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 更新API
     */
    public boolean update(UpdateRequest updateRequest) {
        boolean update = false;
        try {
            UpdateResponse updateResponse = restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
            DocWriteResponse.Result result = updateResponse.getResult();
            if (DocWriteResponse.Result.UPDATED.equals(result) || DocWriteResponse.Result.CREATED.equals(result)) {
                update = true;
            }
            return update;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 高亮对象
     *
     * @param name fieldName
     * @return HighlightBuilder
     */
    public HighlightBuilder highlightBuilder(String name) {
        HighlightBuilder highlightBuilder = new HighlightBuilder();
        HighlightBuilder.Field highlightTitle = new HighlightBuilder.Field("title");
        highlightTitle.highlighterType("unified");
        highlightBuilder.field(highlightTitle);
        HighlightBuilder.Field highlightField = new HighlightBuilder.Field(name);
        highlightBuilder.field(highlightField);
        return highlightBuilder;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@Test
public void testBulkIndex() {
	String index = "";
	BulkRequest bulkRequest = new BulkRequest(index);
	
	Map<String, Object> map = new HashMap<>();
	map.put("key","value");
	
	IndexRequest indexRequest = new IndexRequest();
	indexRequest.source(JSON.toJSONString(map, SerializerFeature.WriteNullStringAsEmpty), XContentType.JSON);
	// 可添加多个indexRequest
	bulkRequest.add(indexRequest);
	elasticSearchDao.bulkIndex(bulkRequest);
}