1.引入依赖
1<dependency> 2 <groupId>org.elasticsearch</groupId> 3 <artifactId>elasticsearch</artifactId> 4 <version>${elasticsearch.version}</version> 5 </dependency> 6 <dependency> 7 <groupId>org.elasticsearch.client</groupId> 8 <artifactId>transport</artifactId> 9 <version>${elasticsearch.version}</version> 10 </dependency> 11 <dependency> 12 <groupId>org.elasticsearch.plugin</groupId> 13 <artifactId>transport-netty4-client</artifactId> 14 <version>${elasticsearch.version}</version> 15 </dependency> 16 <dependency> 17 <groupId>org.elasticsearch.client</groupId> 18 <artifactId>elasticsearch-rest-high-level-client</artifactId> 19 <version>${elasticsearch.version}</version> 20 </dependency> 21 <dependency> 22 <groupId>org.scala-lang</groupId> 23 <artifactId>scala-library</artifactId> 24 <version>2.11.0</version> 25 </dependency>
2.配置信息：
1/** 2 * 读取client配置信息 3 * @author 4 * 5 */ 6@Configuration 7@Getter 8@Setter 9public class ClientConfig { 10 11 /** 12 * elk集群地址 13 */ 14 @Value("${elasticsearch.ip}") 15 private String esHostName; 16 /** 17 * 端口 18 */ 19 @Value("${elasticsearch.port}") 20 private Integer esPort; 21 /** 22 * 集群名称 23 */ 24 @Value("${elasticsearch.cluster.name}") 25 private String esClusterName; 26 27 /** 28 * 连接池 29 */ 30 @Value("${elasticsearch.pool}") 31 private Integer esPoolSize; 32 33 34 /** 35 * 是否服务启动时重新创建索引 36 */ 37 @Value("${elasticsearch.regenerateIndexEnabled}") 38 private Boolean esRegenerateIndexFlag; 39 40 41 /** 42 * 是否服务启动时索引数据同步 43 */ 44 @Value("${elasticsearch.syncDataEnabled}") 45 private Boolean esSyncDataEnabled; 46}
3.es配置启动类：
1import org.elasticsearch.client.transport.TransportClient; 2import org.elasticsearch.common.settings.Settings; 3import org.elasticsearch.common.transport.TransportAddress; 4import org.elasticsearch.transport.client.PreBuiltTransportClient; 5import org.slf4j.Logger; 6import org.slf4j.LoggerFactory; 7import org.springframework.beans.factory.annotation.Autowired; 8import org.springframework.context.annotation.Bean; 9import org.springframework.context.annotation.Configuration; 10 11import java.net.InetAddress; 12 13/** 14 * es配置启动类 15 * @author 16 * 17 */ 18@Configuration 19public class ElasticsearchConfig { 20 private static final Logger LOGGER = LoggerFactory.getLogger(ElasticsearchConfig.class); 21 22 @Autowired 23 ClientConfig clientConfig; 24 25 @Bean 26 public TransportClient init() { 27 LOGGER.info("初始化开始。。。。。"); 28 TransportClient transportClient = null; 29 30 try { 31 /** 32 * 配置信息 33 * client.transport.sniff 增加嗅探机制，找到ES集群 34 * thread_pool.search.size 增加线程池个数，暂时设为5 35 */ 36 Settings esSetting = Settings.builder() 37 .put("client.transport.sniff", true) 38 .put("thread_pool.search.size", clientConfig.getEsPoolSize()) 39 .build(); 40 //配置信息Settings自定义 41 transportClient = new PreBuiltTransportClient(esSetting); 42 TransportAddress transportAddress = new TransportAddress(InetAddress.getByName(clientConfig.getEsHostName()), clientConfig.getEsPort()); 43 transportClient.addTransportAddresses(transportAddress); 44 45 46 } catch (Exception e) { 47 LOGGER.error("elasticsearch TransportClient create error!!!", e); 48 } 49 50 return transportClient; 51 } 52 53 54}
4.操作工具类：
1import com.alibaba.fastjson.JSON; 2import com.alibaba.fastjson.JSONObject; 3import org.apache.commons.lang3.StringUtils; 4import org.elasticsearch.action.admin.indices.create.CreateIndexResponse; 5import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse; 6import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest; 7import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse; 8import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse; 9import org.elasticsearch.action.bulk.BackoffPolicy; 10import org.elasticsearch.action.bulk.BulkProcessor; 11import org.elasticsearch.action.bulk.BulkRequest; 12import org.elasticsearch.action.bulk.BulkResponse; 13import org.elasticsearch.action.get.GetRequestBuilder; 14import org.elasticsearch.action.get.GetResponse; 15import org.elasticsearch.action.index.IndexRequest; 16import org.elasticsearch.action.index.IndexResponse; 17import org.elasticsearch.action.search.SearchRequestBuilder; 18import org.elasticsearch.action.search.SearchResponse; 19import org.elasticsearch.action.update.UpdateRequest; 20import org.elasticsearch.client.transport.TransportClient; 21import org.elasticsearch.common.settings.Settings; 22import org.elasticsearch.common.text.Text; 23import org.elasticsearch.common.transport.TransportAddress; 24import org.elasticsearch.common.unit.TimeValue; 25import org.elasticsearch.common.xcontent.XContentType; 26import org.elasticsearch.index.query.BoolQueryBuilder; 27import org.elasticsearch.search.SearchHit; 28import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder; 29import org.elasticsearch.search.fetch.subphase.highlight.HighlightField; 30import org.elasticsearch.search.sort.SortOrder; 31import org.elasticsearch.transport.client.PreBuiltTransportClient; 32import org.slf4j.Logger; 33import org.slf4j.LoggerFactory; 34import org.springframework.beans.factory.annotation.Autowired; 35import org.springframework.beans.factory.annotation.Value; 36import org.springframework.core.io.ClassPathResource; 37import org.springframework.stereotype.Component; 38 39import javax.annotation.PostConstruct; 40import java.io.InputStream; 41import java.lang.reflect.Method; 42import java.util.ArrayList; 43import java.util.List; 44import java.util.Map; 45import java.util.UUID; 46 47 48public class ElasticsearchUtils { 49 50 private static final Logger LOGGER = LoggerFactory.getLogger(ElasticsearchUtils.class); 51 52 @Autowired 53 private TransportClient transportClient; 54 55 private static TransportClient client; 56 57 @PostConstruct 58 public void init() { 59 client = this.transportClient; 60 } 61 62 /** 63 * 创建索引以及设置其内容 64 * @param index 65 * @param indexType 66 * @param filePath:json文件路径 67 */ 68 public static void createIndex(String index,String indexType,String filePath) throws RuntimeException { 69 try { 70 StringBuffer strBuf = new StringBuffer(); 71 //解析json配置 72 ClassPathResource resource = new ClassPathResource(filePath); 73 InputStream inputStream = resource.getInputStream(); 74 75 int len = 0; 76 byte[] buf = new byte[1024]; 77 while((len=inputStream.read(buf)) != -1) { 78 strBuf.append(new String(buf, 0, len, "utf-8")); 79 } 80 inputStream.close(); 81 //创建索引 82 createIndex(index); 83 //设置索引元素 84 putMapping(index, indexType, strBuf.toString()); 85 86 }catch(Exception e){ 87 throw new RuntimeException(e.getMessage()); 88 } 89 } 90 91 92 /** 93 * 创建索引 94 * 95 * @param index 索引名称 96 * @return 97 */ 98 public static boolean createIndex(String index){ 99 100 try { 101 if (isIndexExist(index)) { 102 //索引库存在则删除索引 103 deleteIndex(index); 104 } 105 CreateIndexResponse indexresponse = client.admin().indices().prepareCreate(index).setSettings(Settings.builder().put("index.number_of_shards", 5) 106 .put("index.number_of_replicas", 1) 107 ) 108 .get(); 109 LOGGER.info("创建索引 {} 执行状态 {}", index , indexresponse.isAcknowledged()); 110 111 return indexresponse.isAcknowledged(); 112 }catch (Exception e) { 113 throw new RuntimeException(e.getMessage()); 114 } 115 116 } 117 118 119 /** 120 * 创建索引 121 * 122 * @param index 索引名称 123 * @param indexType 索引类型 124 * @param mapping 创建的mapping结构 125 * @return 126 */ 127 public static boolean putMapping(String index,String indexType,String mapping) throws RuntimeException { 128 if (!isIndexExist(index)) { 129 throw new RuntimeException("创建索引库"+index+"mapping"+mapping+"结构失败,索引库不存在!"); 130 } 131 try { 132 PutMappingResponse indexresponse = client.admin().indices().preparePutMapping(index).setType(indexType).setSource(mapping, XContentType.JSON).get(); 133 134 LOGGER.info("索引 {} 设置 mapping {} 执行状态 {}", index ,indexType, indexresponse.isAcknowledged()); 135 136 return indexresponse.isAcknowledged(); 137 }catch (Exception e) { 138 throw new RuntimeException(e.getMessage()); 139 } 140 141 142 } 143 144 /** 145 * 判断索引是否存在 146 * 147 * @param index 148 * @return 149 */ 150 public static boolean isIndexExist(String index) { 151 IndicesExistsResponse inExistsResponse = client.admin().indices().exists(new IndicesExistsRequest(index)) 152 .actionGet(); 153 return inExistsResponse.isExists(); 154 } 155 156 157 /** 158 * 删除索引 159 * 160 * @param index 161 * @return 162 */ 163 public static boolean deleteIndex(String index) throws RuntimeException{ 164 if (!isIndexExist(index)) { 165 return true; 166 } 167 try { 168 DeleteIndexResponse dResponse = client.admin().indices().prepareDelete(index).execute().actionGet(); 169 if (dResponse.isAcknowledged()) { 170 LOGGER.info("delete index " + index + " successfully!"); 171 } else { 172 LOGGER.info("Fail to delete index " + index); 173 } 174 return dResponse.isAcknowledged(); 175 } catch (Exception e) { 176 177 throw new RuntimeException(e.getMessage()); 178 } 179 } 180 181 182 /** 183 * 数据添加 184 * 185 * @param jsonObject 186 * 要增加的数据 187 * @param index 188 * 索引，类似数据库 189 * @param type 190 * 类型，类似表 191 * @return 192 */ 193 public static String addData(JSONObject jsonObject, String index, String type) { 194 return addData(jsonObject, index, type, UUID.randomUUID().toString().replaceAll("-", "").toUpperCase()); 195 } 196 197 /** 198 * 数据添加，正定ID 199 * 200 * @param jsonObject 201 * 要增加的数据 202 * @param index 203 * 索引，类似数据库 204 * @param type 205 * 类型，类似表 206 * @param id 207 * 数据ID 208 * @return 209 */ 210 public static String addData(JSONObject jsonObject, String index, String type, String id)throws RuntimeException { 211 try { 212 IndexResponse response = client.prepareIndex(index, type, id).setSource(jsonObject).get(); 213 214 LOGGER.info("addData response status:{},id:{}", response.status().getStatus(), response.getId()); 215 216 return response.getId(); 217 } catch (Exception e) { 218 throw new RuntimeException(e.getMessage()); 219 } 220 } 221 222 223 /** 224 * 批量数据添加， 225 * 226 * @param list 227 * 要增加的数据 228 * @param pkName 229 * 主键id 230 * @param index 231 * 索引，类似数据库 232 * @param type 233 * 类型，类似表 234 * @return 235 */ 236 public static <T> void addBatchData(List<T> list, String pkName, String index, String type) { 237 if(list == null || list.isEmpty()) { 238 return; 239 } 240 // 创建BulkPorcessor对象 241 BulkProcessor bulkProcessor = BulkProcessor.builder(client, new BulkProcessor.Listener() { 242 @Override 243 public void beforeBulk(long paramLong, BulkRequest paramBulkRequest) { 244 // TODO Auto-generated method stub 245 } 246 247 // 执行出错时执行 248 @Override 249 public void afterBulk(long paramLong, BulkRequest paramBulkRequest, Throwable paramThrowable) { 250 // TODO Auto-generated method stub 251 } 252 @Override 253 public void afterBulk(long paramLong, BulkRequest paramBulkRequest, BulkResponse paramBulkResponse) { 254 // TODO Auto-generated method stub 255 } 256 }) 257 // 1w次请求执行一次bulk 258 .setBulkActions(1000) 259 // 1gb的数据刷新一次bulk 260 // .setBulkSize(new ByteSizeValue(1, ByteSizeUnit.GB)) 261 // 固定5s必须刷新一次 262 .setFlushInterval(TimeValue.timeValueSeconds(5)) 263 // 并发请求数量, 0不并发, 1并发允许执行 264 .setConcurrentRequests(1) 265 // 设置退避, 100ms后执行, 最大请求3次 266 .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3)).build(); 267 268 for (T vo : list) { 269 if(getPkValueByName(vo, pkName)!= null) { 270 String id = getPkValueByName(vo, pkName).toString(); 271 bulkProcessor.add(new IndexRequest(index, type, id).source(JSON.toJSONString(vo), XContentType.JSON)); 272 } 273 274 } 275 bulkProcessor.close(); 276 } 277 278 /** 279 * 根据主键名称获取实体类主键属性值 280 * 281 * @param clazz 282 * @param pkName 283 * @return 284 */ 285 private static Object getPkValueByName(Object clazz, String pkName) { 286 try { 287 String firstLetter = pkName.substring(0, 1).toUpperCase(); 288 String getter = "get" + firstLetter + pkName.substring(1); 289 Method method = clazz.getClass().getMethod(getter, new Class[] {}); 290 Object value = method.invoke(clazz, new Object[] {}); 291 return value; 292 } catch (Exception e) { 293 return null; 294 } 295 } 296 297 298 /** 299 * 通过ID 更新数据 300 * 301 * @param jsonObject 302 * 要增加的数据 303 * @param index 304 * 索引，类似数据库 305 * @param type 306 * 类型，类似表 307 * @param id 308 * 数据ID 309 * @return 310 */ 311 public static void updateDataById(JSONObject jsonObject, String index, String type, String id) throws RuntimeException { 312 313 try{ 314 UpdateRequest updateRequest = new UpdateRequest(); 315 316 updateRequest.index(index).type(type).id(id).doc(jsonObject); 317 318 client.update(updateRequest); 319 } catch (Exception e) { 320 throw new RuntimeException(e.getMessage()); 321 } 322 } 323 324 /** 325 * 批量数据更新， 326 * 327 * @param list 328 * 要增加的数据 329 * @param pkName 330 * 主键id 331 * @param index 332 * 索引，类似数据库 333 * @param type 334 * 类型，类似表 335 * @return 336 */ 337 public static <T> void updateBatchData(List<T> list, String pkName, String index, String type) { 338 // 创建BulkPorcessor对象 339 BulkProcessor bulkProcessor = BulkProcessor.builder(client, new BulkProcessor.Listener() { 340 @Override 341 public void beforeBulk(long paramLong, BulkRequest paramBulkRequest) { 342 // TODO Auto-generated method stub 343 } 344 345 // 执行出错时执行 346 @Override 347 public void afterBulk(long paramLong, BulkRequest paramBulkRequest, Throwable paramThrowable) { 348 // TODO Auto-generated method stub 349 } 350 @Override 351 public void afterBulk(long paramLong, BulkRequest paramBulkRequest, BulkResponse paramBulkResponse) { 352 // TODO Auto-generated method stub 353 } 354 }) 355 // 1w次请求执行一次bulk 356 .setBulkActions(1000) 357 // 1gb的数据刷新一次bulk 358 // .setBulkSize(new ByteSizeValue(1, ByteSizeUnit.GB)) 359 // 固定5s必须刷新一次 360 .setFlushInterval(TimeValue.timeValueSeconds(5)) 361 // 并发请求数量, 0不并发, 1并发允许执行 362 .setConcurrentRequests(1) 363 // 设置退避, 100ms后执行, 最大请求3次 364 .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3)).build(); 365 366 for (T vo : list) { 367 String id = getPkValueByName(vo, pkName).toString(); 368 bulkProcessor.add(new UpdateRequest(index, type, id).doc(JSON.toJSONString(vo), XContentType.JSON)); 369 } 370 bulkProcessor.close(); 371 } 372 373 374 /** 375 * 通过ID获取数据 376 * 377 * @param index 378 * 索引，类似数据库 379 * @param type 380 * 类型，类似表 381 * @param id 382 * 数据ID 383 * @param fields 384 * 需要显示的字段，逗号分隔（缺省为全部字段） 385 * @return 386 */ 387 public static Map<String, Object> searchDataById(String index, String type, String id, String fields) { 388 389 GetRequestBuilder getRequestBuilder = client.prepareGet(index, type, id); 390 391 if (StringUtils.isNotEmpty(fields)) { 392 getRequestBuilder.setFetchSource(fields.split(","), null); 393 } 394 395 GetResponse getResponse = getRequestBuilder.execute().actionGet(); 396 397 return getResponse.getSource(); 398 } 399 400 /** 401 * 使用分词查询 402 * 403 * @param index 404 * 索引名称 405 * @param type 406 * 类型名称,可传入多个type逗号分隔 407 * @param clz 408 * 数据对应实体类 409 * @param fields 410 * 需要显示的字段，逗号分隔（缺省为全部字段） 411 * @param boolQuery 412 * 查询条件 413 * @return 414 */ 415 public static <T> List<T> searchListData(String index, String type, Class<T> clz, String fields,BoolQueryBuilder boolQuery) { 416 return searchListData(index, type, clz, 0, fields, null, null,boolQuery); 417 } 418 419 /** 420 * 使用分词查询 421 * 422 * @param index 423 * 索引名称 424 * @param type 425 * 类型名称,可传入多个type逗号分隔 426 * @param clz 427 * 数据对应实体类 428 * @param size 429 * 文档大小限制 430 * @param fields 431 * 需要显示的字段，逗号分隔（缺省为全部字段） 432 * @param sortField 433 * 排序字段 434 * @param highlightField 435 * 高亮字段 436 * @param boolQuery 437 * 查询条件 438 * @return 439 */ 440 public static <T> List<T> searchListData(String index, String type, Class<T> clz, 441 Integer size, String fields, String sortField, String highlightField,BoolQueryBuilder boolQuery) throws RuntimeException{ 442 443 SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index); 444 if (StringUtils.isNotEmpty(type)) { 445 searchRequestBuilder.setTypes(type.split(",")); 446 } 447 // 高亮（xxx=111,aaa=222） 448 if (StringUtils.isNotEmpty(highlightField)) { 449 HighlightBuilder highlightBuilder = new HighlightBuilder(); 450 // 设置高亮字段 451 highlightBuilder.field(highlightField); 452 searchRequestBuilder.highlighter(highlightBuilder); 453 } 454 searchRequestBuilder.setQuery(boolQuery); 455 if (StringUtils.isNotEmpty(fields)) { 456 searchRequestBuilder.setFetchSource(fields.split(","), null); 457 } 458 searchRequestBuilder.setFetchSource(true); 459 460 if (StringUtils.isNotEmpty(sortField)) { 461 searchRequestBuilder.addSort(sortField, SortOrder.DESC); 462 } 463 if (size != null && size > 0) { 464 searchRequestBuilder.setSize(size); 465 } 466 searchRequestBuilder.setScroll(new TimeValue(1000)); 467 searchRequestBuilder.setSize(10000); 468 // 打印的内容 可以在 Elasticsearch head 和 Kibana 上执行查询 469 LOGGER.info("\n{}", searchRequestBuilder); 470 471 SearchResponse searchResponse = searchRequestBuilder.execute().actionGet(); 472 473 long totalHits = searchResponse.getHits().totalHits; 474 if(LOGGER.isDebugEnabled()) { 475 long length = searchResponse.getHits().getHits().length; 476 477 LOGGER.info("共查询到[{}]条数据,处理数据条数[{}]", totalHits, length); 478 } 479 480 481 if (searchResponse.status().getStatus() ==200) { 482 // 解析对象 483 return setSearchResponse(clz, searchResponse, highlightField); 484 } 485 486 return null; 487 } 488 489 490 /** 491 * 高亮结果集 特殊处理 492 * 493 * @param clz 494 * 数据对应实体类 495 * @param searchResponse 496 * 497 * @param highlightField 498 * 高亮字段 499 */ 500 private static <T> List<T> setSearchResponse(Class<T> clz, SearchResponse searchResponse, String highlightField) { 501 List<T> sourceList = new ArrayList<T>(); 502 for (SearchHit searchHit : searchResponse.getHits().getHits()) { 503 searchHit.getSourceAsMap().put("id", searchHit.getId()); 504 StringBuffer stringBuffer = new StringBuffer(); 505 if (StringUtils.isNotEmpty(highlightField)) { 506 507 // System.out.println("遍历 高亮结果集，覆盖 正常结果集" + searchHit.getSourceAsMap()); 508 HighlightField highlight = searchHit.getHighlightFields().get(highlightField); 509 if(highlight == null) { 510 continue; 511 } 512 Text[] text = highlight.getFragments(); 513 if (text != null) { 514 for (Text str : text) { 515 stringBuffer.append(str.string()); 516 } 517 // 遍历 高亮结果集，覆盖 正常结果集 518 searchHit.getSourceAsMap().put(highlightField, stringBuffer.toString()); 519 } 520 } 521 522 T t = JSON.parseObject(JSON.toJSONString(searchHit.getSourceAsMap()), clz); 523 sourceList.add(t); 524 } 525 526 return sourceList; 527 } 528 529}