yml
spring: data: elasticsearch: client: reactive: endpoints: 192.168.209.160:9200 connection-timeout: 10000#链接到es的超时时间，毫秒为单位，默认10秒（10000毫秒） socket-timeout: 10000#读取和写入的超时时间，单位为毫秒，默认5秒（5000毫秒） elasticsearch: rest: uris: 192.168.209.160:9200 # 这两个属性在新版本的springboot中已经不建议使用，9300属于elasticsearch各节点之间的通讯接口。 # 属于lowlevelclient。我们推荐使用9200的RestHighLevelClient去链接 # cluster-nodes: 127.0.0.1:9300 # cluster-name: helloElasticsearch
pom
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-captcha</artifactId>
<version>5.2.0</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>
Controller
package com.fwz.tproject.testfunction.controller; import java.util.Date; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import com.fwz.tproject.testfunction.service.ElasticSearchUtils; import com.fwz.tproject.testfunction.service.IdGeneratorSnowflake; import com.fwz.tproject.testfunction.service.OrderService; /** * * * @author 冯文哲 * @version 2018-06-11 */ @RestController @RequestMapping(value = "/test") public class MainController { @Autowired private IdGeneratorSnowflake idGenerator; @Autowired ElasticSearchUtils utilsService; @RequestMapping(value = "createIndex") public String elasticsearch() { if (utilsService.createIndex("fwztest_index", 5, 1, "")) { return "创建成功"; } else { return "创建失败"; } } @RequestMapping(value = "addDoc") public String addDoc() { for (int j = 0; j < 1000; j++) { Map<String, Object> map = new ConcurrentHashMap<String, Object>(); map.put("author_id", idGenerator.snowflakeId()); map.put("title", "这有" + j + "个中国人"); map.put("content", "其中有" + (j - 1) + "个老黑"); map.put("create_date", new Date()); utilsService.addDoc("fwztest_index", String.valueOf(idGenerator.snowflakeId()), map); } return "新增成功"; } @RequestMapping(value = "deleteDoc") public String deleteDoc(String id) { utilsService.deleteDoc("fwztest_index", id); return "删除成功"; } @RequestMapping(value = "updateDoc") public String updateDoc(String id) { utilsService.updateDoc("fwztest_index", id, ""); return "修改成功"; } @RequestMapping(value = "selectDoc") public Map<String, Object> selectDoc(String id) { return utilsService.getDoc("fwztest_index", id); } }
Utils
package com.fwz.tproject.testfunction.service; import java.io.IOException; import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import org.elasticsearch.action.ActionListener; import org.elasticsearch.action.DocWriteRequest; 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.search.SearchScrollRequest; 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.indices.CreateIndexRequest; import org.elasticsearch.client.indices.CreateIndexResponse; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.common.xcontent.XContentType; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHits; import org.elasticsearch.search.aggregations.AggregationBuilders; import org.elasticsearch.search.aggregations.Aggregations; import org.elasticsearch.search.aggregations.bucket.terms.Terms; import org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket; import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder; import org.elasticsearch.search.aggregations.metrics.Avg; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.annotation.AsyncResult; import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.stereotype.Service; @Service @EnableAsync public class ElasticSearchUtils { @Autowired private RestHighLevelClient restClient; Logger logger = LoggerFactory.getLogger(ElasticSearchUtils.class.getName()); /** * createIndex * * @param indexName //索引名称 * @param shards //主分片 * @param replicas //备份分片 * @param mapping //mapping配置 * @return */ public boolean createIndex(String indexName, Integer shards, Integer replicas, String mapping) { logger.info(restClient.toString()); CreateIndexRequest request = new CreateIndexRequest(indexName); request.settings(Settings.builder().put("number_of_shards", 5).put("number_of_replicas", 1)); request.mapping( "{"properties":{"author_id":{"type":"long"},"title":{"type":"text","analyzer":"standard","fields":{"keyword":{"type":"keyword","ignore_above":256}}},"content":{"type":"text","analyzer":"ik_max_word","fields":{"keyword":{"type":"keyword","ignore_above":256}}},"create_date":{"type":"date"}}}", XContentType.JSON); request.setTimeout(TimeValue.timeValueMinutes(1)); CreateIndexResponse createIndexResponse; try { createIndexResponse = restClient.indices().create(request, RequestOptions.DEFAULT); boolean acknowledged = createIndexResponse.isAcknowledged(); logger.info("是否获取ACK：" + acknowledged); return acknowledged; } catch (IOException e) { // TODO Auto-generated catch block logger.error(e.toString()); } return false; } /** * * addDocument * * @param index 索引名称 * @param id 数据ID（为空则使用es内部ID） * @param source 数据（json 或 Map） * @return * @author fwzz * @version 创建时间：2021年1月27日 下午5:10:42 * */ @Async public Future<Boolean> addDoc(String index, String id, Map<String, Object> source) { // 增, source 里对象创建方式可以是JSON字符串，或者Map，或者XContentBuilder 对象 IndexRequest indexRequest = new IndexRequest(index).source(source); if (id != null && !"".equals(id)) { indexRequest = indexRequest.id(id); } try { IndexResponse res = restClient.index(indexRequest, RequestOptions.DEFAULT); logger.info("新增数据成功，ID为： " + res.getId()); return new AsyncResult<Boolean>(true); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } return new AsyncResult<Boolean>(false); } /** * * deleteDocument * * @param index 索引名称 * @param id 数据ID * @return * @author fwzz * @version 创建时间：2021年1月27日 下午5:19:26 * */ public boolean deleteDoc(String index, String id) { // 删 DeleteRequest deleteRequest = new DeleteRequest(index, id); DeleteResponse res; try { res = restClient.delete(deleteRequest, RequestOptions.DEFAULT); logger.info(res.getResult().toString()); logger.info("删除数据成功，ID为： " + res.getId()); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } return false; } /** * * updateDocument * * @param index * @param id * @param source * @return * @author fwzz * @version 创建时间：2021年1月27日 下午5:25:43 * */ public boolean updateDoc(String index, String id, String source) { // 改, source 里对象创建方式可以是JSON字符串，或者Map，或者XContentBuilder 对象 UpdateRequest updateRequest = new UpdateRequest(index, id).doc(source); try { UpdateResponse res = restClient.update(updateRequest, RequestOptions.DEFAULT); logger.info("修改数据成功，ID为： " + res.getId()); return true; } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } return false; } /** * selectDocument * * @param index * @param id * @return * @author fwzz * @version 创建时间：2021年1月27日 下午5:27:33 * */ public Map<String, Object> getDoc(String index, String id) { // 查 GetRequest getRequest = new GetRequest(index, id); try { GetResponse res = restClient.get(getRequest, RequestOptions.DEFAULT); logger.info("查询数据成功，ID为： " + res.getId()); logger.info("查询数据成功，字符串数据为： " + res.getSourceAsString()); Map<String, Object> map = res.getSourceAsMap(); logger.info("查询数据成功，Map数据为： " + map.toString()); return map; } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } return null; } /** * bulkDemo * * @param index * @param id * @return * @author fwzz * @version 创建时间：2021年1月27日 下午7:35:34 * */ public Boolean bulkRequest(String index, String id) { BulkRequest request = new BulkRequest(); /** * map为更新或新增的数据 */ request.add(new IndexRequest(index).source(XContentType.JSON, new HashMap<String, Object>())); request.add(new DeleteRequest(index, id)); request.add(new UpdateRequest(index, id).doc(XContentType.JSON, new HashMap<String, Object>())); BulkResponse bulkResponse; try { bulkResponse = restClient.bulk(request, RequestOptions.DEFAULT); for (BulkItemResponse bulkItemResponse : bulkResponse) { if (bulkItemResponse.isFailed()) { BulkItemResponse.Failure failure = bulkItemResponse.getFailure(); logger.info(failure.getMessage()); continue; } DocWriteResponse itemResponse = bulkItemResponse.getResponse(); if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.INDEX || bulkItemResponse.getOpType() == DocWriteRequest.OpType.CREATE) { IndexResponse indexResponse = (IndexResponse) itemResponse; logger.info(indexResponse.getResult().toString()); } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) { UpdateResponse updateResponse = (UpdateResponse) itemResponse; logger.info(updateResponse.getResult().toString()); } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.DELETE) { DeleteResponse deleteResponse = (DeleteResponse) itemResponse; logger.info(deleteResponse.getResult().toString()); } } return true; } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } return false; } /** * searchQueryDemo 可完全取代getRequest * * @return * @author fwzz * @version 创建时间：2021年1月27日 下午7:43:57 * */ public Boolean searchQuery() { /** * 指定index */ SearchRequest searchRequest = new SearchRequest("gdp_tops*"); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); /** * 指定query */ sourceBuilder.query(QueryBuilders.termQuery("city", "北京市")); sourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS)); searchRequest.source(sourceBuilder); try { SearchResponse response = restClient.search(searchRequest, RequestOptions.DEFAULT); Arrays.stream(response.getHits().getHits()).forEach(i -> { System.out.println(i.getIndex()); System.out.println(i.getSourceAsMap()); }); logger.info(response.getHits().getTotalHits().toString()); return true; } catch (IOException e) { e.printStackTrace(); } return false; } /** * aggsSearchDemo * * @return * @author fwzz * @version 创建时间：2021年1月27日 下午7:46:31 * */ public Boolean aggsQuery() { /** * 指定index */ SearchRequest searchRequest = new SearchRequest("gdp_tops*"); SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); TermsAggregationBuilder aggregation = AggregationBuilders.terms("by_company").field("company.keyword"); aggregation.subAggregation(AggregationBuilders.avg("average_age").field("age")); searchSourceBuilder.aggregation(aggregation); searchSourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS)); /** * 分页查询 */ /* * searchSourceBuilder.from(0); searchSourceBuilder.size(5); */ searchRequest.source(searchSourceBuilder); try { /** * 处理方法1 （1 2 都尝试一下） */ SearchResponse response = restClient.search(searchRequest, RequestOptions.DEFAULT); Arrays.stream(response.getHits().getHits()).forEach(i -> { logger.info(i.getIndex()); logger.info(i.getSourceAsMap().toString()); }); /** * 处理方法2 （1 2 都尝试一下） */ Aggregations aggregations = response.getAggregations(); Terms byCompanyAggregation = aggregations.get("by_company"); Bucket elasticBucket = byCompanyAggregation.getBucketByKey("Elastic"); Avg averageAge = elasticBucket.getAggregations().get("average_age"); double avg = averageAge.getValue(); logger.info(response.getHits().getTotalHits().toString()); return true; } catch (IOException e) { e.printStackTrace(); } return false; } /** * searchAsyncDemo * * @return * @author fwzz * @version 创建时间：2021年1月27日 下午7:50:00 * */ public Boolean searchAsync() { /** * 指定index */ SearchRequest searchRequest = new SearchRequest("gdp_tops*"); restClient.searchAsync(searchRequest, RequestOptions.DEFAULT, new ActionListener<SearchResponse>() { @Override public void onResponse(SearchResponse searchResponse) { SearchHit[] searchHits = searchResponse.getHits().getHits(); for (SearchHit hit : searchHits) { // 结果的Index String index = hit.getIndex(); // 结果的ID String id = hit.getId(); // 结果的评分 float score = hit.getScore(); // 查询的结果 JSON字符串形式 String sourceAsString = hit.getSourceAsString(); // 查询的结果 Map的形式 Map<String, Object> sourceAsMap = hit.getSourceAsMap(); // Document的title String documentTitle = (String) sourceAsMap.get("title"); // 结果中的某个List List<Object> users = (List<Object>) sourceAsMap.get("user"); // 结果中的某个Map Map<String, Object> innerObject = (Map<String, Object>) sourceAsMap.get("innerObject"); } } @Override public void onFailure(Exception e) { logger.error(e.toString()); } }); return true; } /** * 有时候需要查询的数据太多，可以考虑使用SearchRequest.scroll()方法拿到scrollId;之后再使用SearchScrollRequest * 其用法如下： * * @return * @author fwzz * @version 创建时间：2021年1月27日 下午8:00:14 * */ public Boolean searchScroll() { SearchRequest searchRequest = new SearchRequest("posts"); SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); searchSourceBuilder.query(QueryBuilders.termQuery("city", "北京市")); searchSourceBuilder.size(5); searchRequest.source(searchSourceBuilder); searchRequest.scroll(TimeValue.timeValueMinutes(1L)); SearchResponse searchResponse; try { searchResponse = restClient.search(searchRequest, RequestOptions.DEFAULT); String scrollId = searchResponse.getScrollId(); SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId); scrollRequest.scroll(TimeValue.timeValueSeconds(30)); SearchResponse searchScrollResponse = restClient.scroll(scrollRequest, RequestOptions.DEFAULT); scrollId = searchScrollResponse.getScrollId(); SearchHits hits = searchScrollResponse.getHits(); logger.info(hits.getTotalHits().toString()); return true; } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } return false; } }
全局ID生成工具类
package com.fwz.tproject.testfunction.service; import javax.annotation.PostConstruct; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; import cn.hutool.core.lang.Snowflake; import cn.hutool.core.net.NetUtil; import cn.hutool.core.util.IdUtil; @Component public class IdGeneratorSnowflake { private long workerId = 0; private long datacenterId = 1; private Snowflake snowflake = IdUtil.createSnowflake(workerId, datacenterId); private static final Logger log = LoggerFactory.getLogger(IdGeneratorSnowflake.class.getName()); // 依赖注入完成后执行该方法，进行一些初始化工作 @PostConstruct public void init() { try { workerId = NetUtil.ipv4ToLong(NetUtil.getLocalhostStr()); log.info("当前机器的workerId: {}", workerId); } catch (Exception e) { e.printStackTrace(); log.warn("当前机器的workerId获取失败", e); // 释放ID workerId = NetUtil.getLocalhostStr().hashCode(); } } // 使用默认机房号获取ID public synchronized long snowflakeId() { return snowflake.nextId(); } // 自己制定机房号获取ID public synchronized long snowflakeId(long workerId, long datacenterId) { Snowflake snowflake = IdUtil.createSnowflake(workerId, datacenterId); return snowflake.nextId(); } /** * 生成的是不带-的字符审，类似于: 73a64edf935d4952a287739a66f96e06 * * @return */ public String simpleUUID() { return IdUtil.simpleUUID(); } /** * 生成的UUID是带-的字符串，类似于: b12b6401-6f9c-4351-b2b6-d8afc9ab9272 * * @return */ public String randomUUID() { return IdUtil.randomUUID(); } public static void main(String[] args) { IdGeneratorSnowflake f = new IdGeneratorSnowflake(); for (int i = 0; i < 1000; i++) { System.out.println(f.snowflakeId(0, 0)); } } }