0
0

springboot 2.0集成elasticsearch 7.6.2(集群)

小伙伴们，你们好呀，我是老寇，跟我一起学习es 7.6.2

注：请点击我，获取源码

一、引入依赖配置pom.xml

  1.         <dependency>
  2.             <groupId>org.springframework.boot</groupId>
  3.             <artifactId>spring-boot-starter-web</artifactId>
  4.         </dependency>
  5.         <dependency>
  6.             <groupId>io.springfox</groupId>
  7.             <artifactId>springfox-swagger2</artifactId>
  8.         </dependency>
  9.         <dependency>
  10.             <groupId>io.springfox</groupId>
  11.             <artifactId>springfox-swagger-ui</artifactId>
  12.         </dependency> 
  13.        <dependency>
  14.             <groupId>org.elasticsearch.client</groupId>
  15.             <artifactId>elasticsearch-rest-high-level-client</artifactId>
  16.             <version>7.6.2</version>
  17.             <exclusions>
  18.                 <exclusion>
  19.                     <groupId>org.elasticsearch</groupId>
  20.                     <artifactId>elasticsearch</artifactId>
  21.                 </exclusion>
  22.                 <exclusion>
  23.                     <groupId>org.elasticsearch.client</groupId>
  24.                     <artifactId>elasticsearch-rest-client</artifactId>
  25.                 </exclusion>
  26.             </exclusions>
  27.         </dependency>
  28.         <dependency>
  29.             <groupId>org.elasticsearch.client</groupId>
  30.             <artifactId>elasticsearch-rest-client</artifactId>
  31.             <version>7.6.2</version>
  32.         </dependency>
  33.         <dependency>
  34.             <groupId>org.elasticsearch</groupId>
  35.             <artifactId>elasticsearch</artifactId>
  36.             <version>7.6.2</version>
  37.         </dependency>

二、配置application-dev.yml(生产就克隆application-dev改成生产配置)

  1. elasticsearch:
  2.   host: 192.168.1.1:9200,192.1.2.133:9200,192.168.1.3:9200
  3.   cluster-name: laokou-elasticsearch
  4.   username:
  5.   password:
  6.   synonym:
  7.     path: http://192.168.1.1:9048/laokou-service/synonym

问题思考：比如说，一条文章记录，它有标题，内容，阅读量，在数据存入es时，我需要对es配置分词器，并且能够通过阅读量来筛选数据，你怎么做？

三、配置ES注解

注解可以修饰属性或方法（前提是先配置）

type > 需要在es配置什么类型

participle > 需要配置什么分词器

  1. /**
  2.  * @author Kou Shenhai
  3.  */
  4. @Target({ElementType.FIELD,ElementType.METHOD})
  5. @Retention(RetentionPolicy.RUNTIME)
  6. @Documented
  7. @Inherited
  8. public @interface FieldInfo {
  10.     /**
  11.      * 默认 keyword
  12.      * @return
  13.      */
  14.     String type() default "keyword";
  16.     /**
  17.      * 0 not_analyzed 1 ik_smart 2.ik_max_word 3.ik-index(自定义分词器)
  18.      * @return
  19.      */
  20.     int participle() default 0;
  22. }

拼接属性对应的类型及分词器

  1. /**
  2.  * 属性、类型、分词器
  3.  * @author Kou Shenhai 2413176044@leimingtech.com
  4.  * @version 1.0
  5.  * @date 2021/2/9 0009 上午 10:20
  6.  */
  7. @Data
  8. @NoArgsConstructor
  9. public class FieldMapping {
  11.     private String field;
  13.     private String type;
  15.     private Integer participle;
  17.     public FieldMapping(String field, String type, Integer participle) {
  18.         this.field = field;
  19.         this.type = type;
  20.         this.participle = participle;
  21.     }
  22. }

组装每个属性对应的类型及分词器 => List<FieldMapping>

  1. /**
  2.  * 每个属性对应的类型及分词器
  3.  * @author Kou Shenhai 2413176044@leimingtech.com
  4.  * @version 1.0
  5.  * @date 2021/1/24 0024 下午 7:51
  6.  */
  7. @Slf4j
  8. public class FieldMappingUtil {
  10.     public static List<FieldMapping> getFieldInfo(Class clazz) {
  11.         return getFieldInfo(clazz, null);
  12.     }
  14.     public static List<FieldMapping> getFieldInfo(Class clazz, String fieldName) {
  15.         //返回class中的所有字段（包括私有字段）
  16.         Field[] fields = clazz.getDeclaredFields();
  17.         //创建FieldMapping集合
  18.         List<FieldMapping> fieldMappingList = new ArrayList<>();
  19.         for (Field field : fields) {
  20.             //获取字段上的FieldInfo对象
  21.             boolean annotationPresent = field.isAnnotationPresent(FieldInfo.class);
  22.             if (annotationPresent) {
  23.                 FieldInfo fieldInfo = field.getAnnotation(FieldInfo.class);
  24.                 //获取字段名称
  25.                 String name = field.getName();
  26.                 fieldMappingList.add(new FieldMapping(name, fieldInfo.type(), fieldInfo.participle()));
  27.             } else {
  28.                 continue;
  29.             }
  30.         }
  31.         return fieldMappingList;
  32.     }
  33. }

四、配置es及swagger

  1. /**
  2.  * es配置文件
  3.  * @author Kou Shenhai 2413176044@leimingtech.com
  4.  * @version 1.0
  5.  * @date 2020/8/9 0009 下午 2:01
  6.  */
  7. @Configuration
  8. public class ElasticsearchConfig {
  9.     
  10.     private static final String HTTP_SCHEME = "http";
  12.     private static final Logger LOGGER = LoggerFactory.getLogger(ElasticsearchConfig.class);
  13.     /**
  14.      * 权限验证
  15.      */
  16.     final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
  17.     /**
  18.      * es主机
  19.      */
  20.     @Value("${elasticsearch.host}")
  21.     private String[] host;
  23.     @Value("${elasticsearch.username}")
  24.     private String username;
  26.     @Value("${elasticsearch.password}")
  27.     private String password;
  29.     @Bean
  30.     public RestClientBuilder restClientBuilder() {
  31.         HttpHost[] hosts = Arrays.stream(host)
  32.                 .map(this::makeHttpHost)
  33.                 .filter(Objects::nonNull)
  34.                 .toArray(HttpHost[]::new);
  35.         LOGGER.info("host:{}",Arrays.toString(hosts));
  36.         //配置权限验证
  37.         credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));
  38.         return RestClient.builder(hosts).setHttpClientConfigCallback(httpClientBuilder ->
  39.                 httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)
  40.                 .setMaxConnPerRoute(100)
  41.                 //最大连接数
  42.                 .setMaxConnTotal(100)
  43.         ).setRequestConfigCallback(builder -> {
  44.             builder.setConnectTimeout(-1);
  45.             builder.setSocketTimeout(60000);
  46.             builder.setConnectionRequestTimeout(-1);
  47.             return builder;
  48.         });
  49.     }
  51.     /**
  52.      * 处理请求地址
  53.      * @param address
  54.      * @return
  55.      */
  56.     private HttpHost makeHttpHost(String address) {
  57.         assert StringUtils.isNotEmpty(address);
  58.         String[] hostAddress = address.split(":");
  59.         if (hostAddress.length == 2) {
  60.             String ip = hostAddress[0];
  61.             Integer port = Integer.valueOf(hostAddress[1]);
  62.             return new HttpHost(ip, port, HTTP_SCHEME);
  63.         } else {
  64.             return null;
  65.         }
  66.     }
  68.     /**
  69.      * 配置highLevelClient bean
  70.      * @param restClientBuilder
  71.      * @return
  72.      */
  73.     @Bean(name = "restHighLevelClient")
  74.     public RestHighLevelClient restHighLevelClient(@Autowired RestClientBuilder restClientBuilder) {
  75.         return new RestHighLevelClient(restClientBuilder);
  76.     }
  77. }
  1. /**
  2.  * @author Kou Shenhai
  3.  */
  4. @Configuration
  5. @EnableSwagger2
  6. public class SwaggerConfig {
  7.     @Bean
  8.     public Docket createRestApi() {
  9.         return new Docket(DocumentationType.SWAGGER_2)
  10.                 .apiInfo(apiInfo())
  11.                 .select()
  12.                 .apis(RequestHandlerSelectors.withMethodAnnotation(ApiOperation.class))
  13.                 .paths(PathSelectors.any())
  14.                 .build();
  15.     }
  16.     private ApiInfo apiInfo() {
  17.         return new ApiInfoBuilder()
  18.                 .title("API文档")
  19.                 .version("2.0.0")
  20.                 .description("API文档 - Elasticsearch服务")
  21.                 //作者信息
  22.                 .contact(new Contact("寇申海", "https://blog.csdn.net/qq_39893313", "2413176044@qq.com"))
  23.                 .build();
  24.     }
  25. }

五、ES工具类 （索引相关配置不懂的，请查看elasticsearch 7.6.2 - 索引管理）

  1. /**
  2.  * Elasticsearch工具类-用于操作ES
  3.  * @author Kou Shenhai 2413176044@leimingtech.com
  4.  * @version 1.0
  5.  * @date 2021/1/24 0024 下午 5:42
  6.  */
  7. @Slf4j
  8. @Component
  9. public class ElasticsearchUtil {
  11.     private static final String PRIMARY_KEY_NAME = "id";
  13.     @Value("${elasticsearch.synonym.path}")
  14.     private String synonymPath;
  16.     @Autowired
  17.     private RestHighLevelClient restHighLevelClient;
  19.     /**
  20.      * 批量同步数据到ES
  21.      * @param indexName 索引名称
  22.      * @param indexAlias 别名
  23.      * @param jsonDataList 数据列表
  24.      * @param clazz 类型
  25.      * @return
  26.      * @throws IOException
  27.      */
  28.     public boolean saveDataBatch(String indexName,String indexAlias,String jsonDataList,Class clazz) throws IOException {
  29.         //判空
  30.         if (StringUtils.isBlank(jsonDataList)) {
  31.             return false;
  32.         }
  33.         if (syncIndex(indexName, indexAlias, clazz)) {
  34.             return false;
  35.         }
  36.         //3.批量操作Request
  37.         BulkRequest bulkRequest = packBulkIndexRequest(indexName, jsonDataList);
  38.         if (bulkRequest.requests().isEmpty()) {
  39.             return false;
  40.         }
  41.         final BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
  42.         if (bulk.hasFailures()) {
  43.             for (BulkItemResponse item : bulk.getItems()) {
  44.                 log.error("索引[{}],主键[{}]更新操作失败，状态为:[{}],错误信息:{}",indexName,item.getId(),item.status(),item.getFailureMessage());
  45.             }
  46.             return false;
  47.         }
  48.         // 记录索引新增与修改数量
  49.         Integer createdCount = 0;
  50.         Integer updatedCount = 0;
  51.         for (BulkItemResponse item : bulk.getItems()) {
  52.             if (DocWriteResponse.Result.CREATED.equals(item.getResponse().getResult())) {
  53.                 createdCount++;
  54.             } else if (DocWriteResponse.Result.UPDATED.equals(item.getResponse().getResult())){
  55.                 updatedCount++;
  56.             }
  57.         }
  58.         log.info("索引[{}]批量同步更新成功，共新增[{}]个，修改[{}]个",indexName,createdCount,updatedCount);
  59.         return true;
  60.     }
  62.     /**
  63.      * ES修改数据
  64.      * @param indexName 索引名称
  65.      * @param id 主键
  66.      * @param paramJson 参数JSON
  67.      * @return
  68.      */
  69.     public boolean updateData(String indexName,String id,String paramJson) {
  70.         UpdateRequest updateRequest = new UpdateRequest(indexName, id);
  71.         //如果修改索引中不存在则进行新增
  72.         updateRequest.docAsUpsert(true);
  73.         //立即刷新数据
  74.         updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
  75.         updateRequest.doc(paramJson,XContentType.JSON);
  76.         try {
  77.             UpdateResponse updateResponse = restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
  78.             log.info("索引[{}],主键：【{}】操作结果:[{}]",indexName,id,updateResponse.getResult());
  79.             if (DocWriteResponse.Result.CREATED.equals(updateResponse.getResult())) {
  80.                 //新增
  81.                 log.info("索引：【{}】,主键：【{}】新增成功",indexName,id);
  82.                 return true;
  83.             } else if (DocWriteResponse.Result.UPDATED.equals(updateResponse.getResult())) {
  84.                 //修改
  85.                 log.info("索引：【{}】，主键：【{}】修改成功",indexName, id);
  86.                 return true;
  87.             } else if (DocWriteResponse.Result.NOOP.equals(updateResponse.getResult())) {
  88.                 //无变化
  89.                 log.info("索引:[{}],主键:[{}]无变化",indexName, id);
  90.                 return true;
  91.             }
  92.         } catch (IOException e) {
  93.             e.printStackTrace();
  94.             log.error("索引：[{}],主键：【{}】，更新异常:[{}]",indexName, id,e);
  95.             return false;
  96.         }
  97.         return false;
  98.     }
  100.     /**
  101.      * 批量修改ES
  102.      * @param indexName 索引名称
  103.      * @param indexAlias 别名
  104.      * @param jsonDataList 数据列表
  105.      * @param clazz 类型
  106.      * @return
  107.      * @throws IOException
  108.      */
  109.     public boolean updateDataBatch(String indexName,String indexAlias, String jsonDataList,Class clazz) throws IOException {
  110.         //1.创建索引
  111.         boolean createIndexFlag = createIndex(indexName,indexAlias, clazz);
  112.         if (!createIndexFlag) {
  113.             return false;
  114.         }
  115.         return this.updateDataBatch(indexName,jsonDataList);
  116.     }
  118.     /**
  119.      * 删除数据
  120.      * @param indexName 索引名称
  121.      * @param id 主键
  122.      * @return
  123.      */
  124.     public boolean deleteData(String indexName,String id) {
  125.         DeleteRequest deleteRequest = new DeleteRequest(indexName);
  126.         deleteRequest.id(id);
  127.         deleteRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
  128.         try {
  129.             DeleteResponse deleteResponse = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
  130.             if (DocWriteResponse.Result.NOT_FOUND.equals(deleteResponse.getResult())) {
  131.                 log.error("索引：【{}】，主键：【{}】删除失败",indexName, id);
  132.                 return false;
  133.             } else {
  134.                 log.info("索引【{}】主键【{}】删除成功",indexName, id);
  135.                 return true;
  136.             }
  137.         } catch (IOException e) {
  138.             e.printStackTrace();
  139.             log.error("删除索引【{}】出现异常[{}]",indexName,e);
  140.             return false;
  141.         }
  142.     }
  144.     /**
  145.      * 批量删除ES
  146.      * @param indexName 索引名称
  147.      * @param ids id列表
  148.      * @return
  149.      */
  150.     public boolean deleteDataBatch(String indexName,List<String> ids) {
  151.         if (CollectionUtils.isEmpty(ids)) {
  152.             return false;
  153.         }
  154.         BulkRequest bulkRequest = packBulkDeleteRequest(indexName, ids);
  155.         try {
  156.             BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
  157.             if (bulk.hasFailures()) {
  158.                 for (BulkItemResponse item : bulk.getItems()) {
  159.                     log.error("删除索引:[{}],主键：{}失败，信息：{}",indexName,item.getId(),item.getFailureMessage());
  160.                 }
  161.                 return false;
  162.             }
  163.             //记录索引新增与修改数量
  164.             Integer deleteCount = 0;
  165.             for (BulkItemResponse item : bulk.getItems()) {
  166.                 if (DocWriteResponse.Result.DELETED.equals(item.getResponse().getResult())) {
  167.                     deleteCount++;
  168.                 }
  169.             }
  170.             log.info("批量删除索引[{}]成功，共删除[{}]个",indexName,deleteCount);
  171.             return true;
  172.         } catch (IOException e) {
  173.             e.printStackTrace();
  174.             log.error("删除索引：【{}】出现异常:{}",indexName,e);
  175.             return false;
  176.         }
  177.     }
  179.     /**
  180.      * 组装删除操作
  181.      * @param indexName 索引名称
  182.      * @param ids id列表
  183.      * @return
  184.      */
  185.     private BulkRequest packBulkDeleteRequest(String indexName, List<String> ids) {
  186.         BulkRequest bulkRequest = new BulkRequest();
  187.         bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
  188.         ids.forEach(id -> {
  189.             DeleteRequest deleteRequest = new DeleteRequest(indexName);
  190.             deleteRequest.id(id);
  191.             bulkRequest.add(deleteRequest);
  192.         });
  193.         return bulkRequest;
  194.     }
  196.     /**
  197.      * 批量修改ES
  198.      * @param indexName 索引名称
  199.      * @param jsonDataList json数据列表
  200.      * @return
  201.      */
  202.     public boolean updateDataBatch(String indexName, String jsonDataList) {
  203.         //判空
  204.         if (StringUtils.isBlank(jsonDataList)) {
  205.             return false;
  206.         }
  207.         BulkRequest bulkRequest = packBulkUpdateRequest(indexName, jsonDataList);
  208.         if (bulkRequest.requests().isEmpty()) {
  209.             return false;
  210.         }
  211.         try {
  212.             //同步执行
  213.             BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
  214.             if (bulk.hasFailures()) {
  215.                 for (BulkItemResponse item : bulk.getItems()) {
  216.                     log.error("索引【{}】,主键[{}]修改操作失败，状态为【{}】,错误信息：{}",indexName,item.status(),item.getFailureMessage());
  217.                 }
  218.                 return false;
  219.             }
  220.             //记录索引新增与修改数量
  221.             Integer createCount = 0;
  222.             Integer updateCount = 0;
  223.             for (BulkItemResponse item : bulk.getItems()) {
  224.                 if (DocWriteResponse.Result.CREATED.equals(item.getResponse().getResult())) {
  225.                     createCount++;
  226.                 } else if (DocWriteResponse.Result.UPDATED.equals(item.getResponse().getResult())) {
  227.                     updateCount++;
  228.                 }
  229.             }
  230.             log.info("索引【{}】批量修改更新成功，共新增[{}]个，修改[{}]个",indexName,createCount,updateCount);
  231.         } catch (IOException e) {
  232.             e.printStackTrace();
  233.             log.error("索引【{}】批量修改更新出现异常",indexName);
  234.             return false;
  235.         }
  236.         return true;
  237.     }
  239.     /**
  240.      * 组装bulkUpdate
  241.      * @param indexName 索引名称
  242.      * @param jsonDataList 数据列表
  243.      * @return
  244.      */
  245.     private BulkRequest packBulkUpdateRequest(String indexName,String jsonDataList) {
  246.         BulkRequest bulkRequest = new BulkRequest();
  247.         bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
  248.         JSONArray jsonArray = JSONArray.parseArray(jsonDataList);
  249.         if (jsonArray.isEmpty()) {
  250.             return bulkRequest;
  251.         }
  252.         jsonArray.forEach(o -> {
  253.             Map<String, String> map = (Map<String, String>) o;
  254.             UpdateRequest updateRequest = new UpdateRequest(indexName,map.get(PRIMARY_KEY_NAME));
  255.             // 修改索引中不存在就新增
  256.             updateRequest.docAsUpsert(true);
  257.             updateRequest.doc(JSON.toJSONString(o),XContentType.JSON);
  258.             bulkRequest.add(updateRequest);
  259.         });
  260.         return bulkRequest;
  261.     }
  263.     /**
  264.      * 删除索引、新建索引
  265.      * @param indexName 索引名称
  266.      * @param indexAlias 别名
  267.      * @param clazz 类型
  268.      * @return
  269.      * @throws IOException
  270.      */
  271.     private boolean syncIndex(String indexName, String indexAlias, Class clazz) throws IOException {
  272.         //1.删除索引
  273.         boolean deleteAllFlag = deleteIndex(indexName);
  274.         if (!deleteAllFlag) {
  275.             return true;
  276.         }
  277.         //2.创建索引
  278.         boolean createIndexFlag = createIndex(indexName, indexAlias, clazz);
  279.         if (!createIndexFlag) {
  280.             return true;
  281.         }
  282.         return false;
  283.     }
  285.     /**
  286.      * 根据主键查询ES
  287.      * @param indexName 索引名称
  288.      * @param id 主键
  289.      * @return
  290.      */
  291.     public String getDataById(String indexName,String id) {
  292.         //判断索引是否存在
  293.         //1.判断索引是否存在
  294.         boolean result = isIndexExists(indexName);
  295.         if (!result) {
  296.             return "";
  297.         }
  298.         GetRequest getRequest = new GetRequest(indexName, id);
  299.         try {
  300.             GetResponse getResponse = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);
  301.             String resultJson = getResponse.getSourceAsString();
  302.             log.info("索引【{}】主键【{}】，查询结果：【{}】",indexName,id,resultJson);
  303.             return resultJson;
  304.         } catch (IOException e) {
  305.             e.printStackTrace();
  306.             log.error("索引【{}】主键[{}]，查询异常：{}",indexName,id,e);
  307.             return "";
  308.         }
  310.     }
  312.     /**
  313.      * 清空索引内容
  314.      * @param indexName 索引名称
  315.      * @return
  316.      */
  317.     public boolean deleteAll(String indexName) {
  318.         //1.判断索引是否存在
  319.         boolean result = isIndexExists(indexName);
  320.         if (!result) {
  321.             log.error("索引【{}】不存在，删除失败",indexName);
  322.             return false;
  323.         }
  324.         DeleteRequest deleteRequest = new DeleteRequest(indexName);
  325.         deleteRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
  326.         try {
  327.             DeleteResponse deleteResponse = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
  328.             if (DocWriteResponse.Result.NOT_FOUND.equals(deleteResponse.getResult())) {
  329.                 log.error("索引【{}】删除失败",indexName);
  330.                 return false;
  331.             }
  332.             log.info("索引【{}】删除成功",indexName);
  333.             return true;
  334.         } catch (IOException e) {
  335.             e.printStackTrace();
  336.             log.error("删除索引[{}]，出现异常[{}]",indexName,e);
  337.             return false;
  338.         }
  339.     }
  341.     /**
  342.      * 批量数据保存到ES-异步
  343.      * @param indexName 索引名称
  344.      * @param indexAlias 索引别名
  345.      * @param jsonDataList 数据列表
  346.      * @param clazz 类型
  347.      * @return
  348.      * @throws IOException
  349.      */
  350.     public boolean saveDataBatchSync(String indexName,String indexAlias,String jsonDataList,Class clazz) throws IOException {
  351.         //判空
  352.         if (StringUtils.isBlank(jsonDataList)) {
  353.             return false;
  354.         }
  355.         if (syncIndex(indexName, indexAlias, clazz)) {
  356.             return false;
  357.         }
  358.         //3.批量操作Request
  359.         BulkRequest bulkRequest = packBulkIndexRequest(indexName, jsonDataList);
  360.         if (bulkRequest.requests().isEmpty()) {
  361.             return false;
  362.         }
  363.         //4.异步执行
  364.         ActionListener<BulkResponse> listener = new ActionListener<BulkResponse>() {
  365.             @Override
  366.             public void onResponse(BulkResponse bulkItemResponses) {
  367.                 if (bulkItemResponses.hasFailures()) {
  368.                     for (BulkItemResponse item : bulkItemResponses.getItems()) {
  369.                         log.error("索引【{}】,主键【{}】更新失败，状态【{}】，错误信息：{}",indexName,item.getId(),
  370.                                 item.status(),item.getFailureMessage());
  371.                     }
  372.                 }
  373.             }
  374.             //失败操作
  375.             @Override
  376.             public void onFailure(Exception e) {
  377.                 log.error("索引【{}】批量异步更新出现异常:{}",indexName,e);
  378.             }
  379.         };
  380.         restHighLevelClient.bulkAsync(bulkRequest,RequestOptions.DEFAULT,listener);
  381.         log.info("索引批量更新索引【{}】中",indexName);
  382.         return true;
  383.     }
  385.     /**
  386.      * 删除索引
  387.      * @param indexName 索引名称
  388.      * @return
  389.      * @throws IOException
  390.      */
  391.     public boolean deleteIndex(String indexName) throws IOException {
  392.         //1.判断索引是否存在
  393.         boolean result = isIndexExists(indexName);
  394.         if (!result) {
  395.             log.error("索引【{}】不存在，删除失败",indexName);
  396.             return false;
  397.         }
  398.         //2.删除操作Request
  399.         DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(indexName);
  400.         deleteIndexRequest.indicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN);
  401.         AcknowledgedResponse acknowledgedResponse = restHighLevelClient.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
  402.         if (!acknowledgedResponse.isAcknowledged()) {
  403.             log.error("索引【{}】删除失败",indexName);
  404.             return false;
  405.         }
  406.         log.info("索引【{}】删除成功",indexName);
  407.         return true;
  408.     }
  410.     /**
  411.      * 批量操作的Request
  412.      * @param indexName 索引名称
  413.      * @param jsonDataList json数据列表
  414.      * @return
  415.      */
  416.     private BulkRequest packBulkIndexRequest(String indexName,String jsonDataList) {
  417.         BulkRequest bulkRequest = new BulkRequest();
  418.         //IMMEDIATE > 请求向es提交数据，立即进行数据刷新<实时性高，资源消耗高>
  419.         //WAIT_UNTIL >  请求向es提交数据，等待数据完成刷新<实时性高，资源消耗低>
  420.         //NONE > 默认策略<实时性低>
  421.         bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
  422.         JSONArray jsonArray = JSONArray.parseArray(jsonDataList);
  423.         if (jsonArray.isEmpty()) {
  424.             return bulkRequest;
  425.         }
  426.         //循环数据封装bulkRequest
  427.         jsonArray.forEach(obj ->{
  428.             final Map<String, String> map = (Map<String, String>) obj;
  429.             IndexRequest indexRequest = new IndexRequest(indexName);
  430.             indexRequest.source(JSON.toJSONString(obj),XContentType.JSON);
  431.             indexRequest.id(map.get(PRIMARY_KEY_NAME));
  432.             bulkRequest.add(indexRequest);
  433.         });
  434.         return bulkRequest;
  435.     }
  437.     /**
  438.      * 创建索引
  439.      * @param indexName 索引名称
  440.      * @param indexAlias 别名
  441.      * @param clazz 类型
  442.      * @return
  443.      * @throws IOException
  444.      */
  445.     public boolean createIndex(String indexName,String indexAlias,Class clazz) throws IOException {
  446.         //判断索引是否存在
  447.         boolean result = isIndexExists(indexName);
  448.         if (!result) {
  449.             boolean createResult = createIndexAndCreateMapping(indexName,indexAlias, FieldMappingUtil.getFieldInfo(clazz));
  450.             if (!createResult) {
  451.                 log.info("索引【{}】创建失败",indexName);
  452.                 return false;
  453.             }
  454.         }
  455.         log.info("索引：[{}]创建成功",indexName);
  456.         return true;
  457.     }
  459.     /**
  460.      * 数据同步到ES
  461.      * @param id 主键
  462.      * @param indexName 索引名称
  463.      * @param jsonData json数据
  464.      * @param clazz 类型
  465.      * @return
  466.      */
  467.     public boolean saveData(String id,String indexName,String indexAlias,String jsonData,Class clazz) throws IOException {
  468.         //1.创建索引
  469.         boolean createIndexFlag = createIndex(indexName,indexAlias, clazz);
  470.         if (!createIndexFlag) {
  471.             return false;
  472.         }
  473.         //2.创建操作Request
  474.         IndexRequest indexRequest = new IndexRequest(indexName);
  475.         //3.配置相关信息
  476.         indexRequest.source(jsonData, XContentType.JSON);
  477.         //IMMEDIATE > 立即刷新
  478.         indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
  479.         indexRequest.id(id);
  480.         IndexResponse response = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
  481.         //4.判断索引是新增还是修改
  482.         if (IndexResponse.Result.CREATED.equals(response.getResult())) {
  483.             log.info("索引【{}】保存成功",indexName);
  484.             return true;
  485.         } else if (IndexResponse.Result.UPDATED.equals(response.getResult())) {
  486.             log.info("索引【{}】修改成功",indexName);
  487.             return true;
  488.         }
  489.         return false;
  490.     }
  492.     /**
  493.      * 判断索引是否存在
  494.      * @param indexName 索引名称
  495.      * @return
  496.      */
  497.     public boolean isIndexExists(String indexName) {
  498.         try {
  499.             GetIndexRequest getIndexRequest = new GetIndexRequest(indexName);
  500.             return restHighLevelClient.indices().exists(getIndexRequest, RequestOptions.DEFAULT);
  501.         }catch (Exception e) {
  502.             e.printStackTrace();
  503.         }
  504.         return false;
  505.     }
  507.     /**
  508.      * 创建索引设置相关配置信息
  509.      * @param indexName 索引名称
  510.      * @param indexAlias 索引别名
  511.      * @param fieldMappingList 数据列表
  512.      * @return
  513.      * @throws IOException
  514.      */
  515.     private boolean createIndexAndCreateMapping(String indexName,String indexAlias, List<FieldMapping> fieldMappingList) throws IOException {
  516.         //封装es索引的mapping
  517.         XContentBuilder mapping = packEsMapping(fieldMappingList, null);
  518.         mapping.endObject().endObject();
  519.         mapping.close();
  520.         //进行索引的创建
  521.         CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);
  522.         //配置分词器
  523.         XContentBuilder settings = packSettingMapping();
  524.         XContentBuilder aliases = packEsAliases(indexAlias);
  525.         log.info("索引配置脚本：{}",settings);
  526.         log.info("索引字段内容：{}",mapping);
  527.         createIndexRequest.settings(settings);
  528.         createIndexRequest.mapping("_doc", mapping);
  529.         createIndexRequest.aliases(aliases);
  530.         //同步方式创建索引
  531.         CreateIndexResponse createIndexResponse = restHighLevelClient.indices().create(createIndexRequest,RequestOptions.DEFAULT);
  532.         boolean acknowledged = createIndexResponse.isAcknowledged();
  533.         if (acknowledged) {
  534.             log.info("索引:{}创建成功", indexName);
  535.             return true;
  536.         } else {
  537.             log.error("索引:{}创建失败", indexName);
  538.             return false;
  539.         }
  540.     }
  542.     /**
  543.      * 配置ES别名
  544.      * @author Kou Shenhai
  545.      * @param alias 别名
  546.      * @return
  547.      * @throws IOException
  548.      */
  549.     private XContentBuilder packEsAliases(String alias) throws IOException{
  550.         XContentBuilder aliases = XContentFactory.jsonBuilder().startObject()
  551.                 .startObject(alias).endObject();
  552.         aliases.endObject();
  553.         aliases.close();
  554.         return aliases;
  555.     }
  557.     /**
  558.      * 配置Mapping
  559.      * @param fieldMappingList 组装的实体类信息
  560.      * @param mapping
  561.      * @return
  562.      * @throws IOException
  563.      */
  564.     private XContentBuilder packEsMapping(List<FieldMapping> fieldMappingList,XContentBuilder mapping) throws IOException {
  565.         if (mapping == null) {
  566.             //如果对象是空，首次进入，设置开始结点
  567.             mapping = XContentFactory.jsonBuilder().startObject()
  568.                     .field("dynamic",true)
  569.                     .startObject("properties");
  570.         }
  571.         //循环实体对象的类型集合封装ES的Mapping
  572.         for (FieldMapping fieldMapping : fieldMappingList) {
  573.             String field = fieldMapping.getField();
  574.             String dataType = fieldMapping.getType();
  575.             Integer participle = fieldMapping.getParticiple();
  576.             //设置分词规则
  577.             if (Constant.NOT_ANALYZED.equals(participle)) {
  578.                 if (FieldTypeEnum.DATE.getValue().equals(dataType)) {
  579.                     mapping.startObject(field)
  580.                             .field("type", dataType)
  581.                             .field("format","yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis")
  582.                             .endObject();
  583.                 } else {
  584.                     mapping.startObject(field)
  585.                             .field("type", dataType)
  586.                             .endObject();
  587.                 }
  588.             } else if (Constant.IK_INDEX.equals(participle)) {
  589.                 mapping.startObject(field)
  590.                         .field("type",dataType)
  591.                         .field("eager_global_ordinals",true)
  592.                         //fielddata=true 用来解决text字段不能进行聚合操作
  593.                         .field("fielddata",true)
  594.                         .field("boost",100.0)
  595.                         .field("analyzer","ik-index-synonym")
  596.                         .field("search_analyzer","ik-search-synonym")
  597.                         .startObject("fields").startObject("data-pinyin")
  598.                         .field("term_vector", "with_positions_offsets")
  599.                         .field("analyzer","ik-search-pinyin")
  600.                         .field("type",dataType)
  601.                         .field("boost",100.0)
  602.                         .endObject().endObject()
  603.                         .endObject();
  604.             }
  605.         }
  606.         return mapping;
  607.     }
  609.     /**
  610.      * 配置Settings
  611.      * @return
  612.      * @throws IOException
  613.      */
  614.     private XContentBuilder packSettingMapping() throws IOException {
  615.         XContentBuilder setting = XContentFactory.jsonBuilder().startObject()
  616.                 .startObject("index")
  617.                 .field("number_of_shards",5)
  618.                 .field("number_of_replicas",1)
  619.                 .field("refresh_interval","120s")
  620.                 .endObject()
  621.                 .startObject("analysis");
  622.         //ik分词 同义词 拼音
  623.         setting.startObject("analyzer")
  624.                 .startObject("ik-search-pinyin")
  625.                 .field("type","custom")
  626.                 .field("tokenizer","ik_smart")
  627.                 .field("char_filter",new String[] {"html_strip"})
  628.                 .field("filter", new String[]{"laokou-pinyin","word_delimiter","lowercase", "asciifolding"})
  629.                 .endObject();
  630.         setting.startObject("ik-index-synonym")
  631.                 .field("type","custom")
  632.                 .field("tokenizer","ik_max_word")
  633.                 .field("char_filter",new String[] {"html_strip"})
  634.                 .field("filter", new String[]{"laokou-remote-synonym"})
  635.                 .endObject();
  636.         setting.startObject("ik-search-synonym")
  637.                 .field("type","custom")
  638.                 .field("tokenizer","ik_smart")
  639.                 .field("char_filter",new String[] {"html_strip"})
  640.                 .field("filter", new String[]{"laokou-remote-synonym"})
  641.                 .endObject();
  642.         setting.endObject();
  643.         //设置拼音分词器 同义词分词
  644.         setting.startObject("filter")
  645.                 .startObject("laokou-pinyin")
  646.                 .field("type", "pinyin")
  647.                 .field("keep_first_letter", false)
  648.                 .field("keep_separate_first_letter", false)
  649.                 .field("keep_full_pinyin", true)
  650.                 .field("keep_original", false)
  651.                 .field("keep_joined_full_pinyin",true)
  652.                 .field("limit_first_letter_length", 16)
  653.                 .field("lowercase", true)
  654.                 .field("remove_duplicated_term", true)
  655.                 .endObject()
  656.                 .startObject("laokou-remote-synonym")
  657.                 .field("type","dynamic_synonym")
  658.                 .field("synonyms_path", synonymPath)
  659.                 .field("interval",120)
  660.                 .field("dynamic_reload",true)
  661.                 .endObject()
  662.                 .endObject();
  663.         setting.endObject().endObject();
  664.         setting.close();
  665.         return setting;
  666.     }
  667. }

问题思考：比如说，我有几条记录，文章记录，聊天记录，订单记录，它们是不同的索引，需要单独建立索引，怎么根据不同的数据类型来创建不同的索引？你会怎么做？

六、索引管理工具类

  1. /**
  2.  * 索引管理
  3.  * @author Kou Shenhai 2413176044@leimingtech.com
  4.  * @version 1.0
  5.  * @date 2021/10/31 0031 上午 10:11
  6.  */
  7. public class FieldUtil {
  9.     public static final String MESSAGE_INDEX = "message";
  11.     private static final Map<String,Class<?>> classMap = new HashMap<>(16);
  13.     static {
  14.         classMap.put(FieldUtil.MESSAGE_INDEX, MessageIndex.class);
  15.     }
  17.     public static Class<?> getClazz(final String indexName) {
  18.         return classMap.getOrDefault(indexName,Object.class);
  19.     }
  20. }

七、测试es

  1. /**
  2.  * Elasticsearch API 服务
  3.  * @author Kou Shenhai 2413176044@leimingtech.com
  4.  * @version 1.0
  5.  * @date 2021/2/8 0008 下午 6:33
  6.  */
  7. @RestController
  8. @RequestMapping("/api")
  9. @Api(tags = "Elasticsearch API 服务")
  10. public class ElasticsearchController {
  12.     @Autowired
  13.     private ElasticsearchUtil elasticsearchUtil;
  15.     @PostMapping("/sync")
  16.     @ApiOperation("同步数据到ES")
  17.     @CrossOrigin
  18.     public void syncIndex(@RequestBody final ElasticsearchModel model) throws IOException {
  19.         String id = model.getId();
  20.         String indexName = model.getIndexName();
  21.         String indexAlias = model.getIndexAlias();
  22.         String jsonData = model.getData();
  23.         Class<?> clazz = FieldUtil.getClazz(indexAlias);
  24.         elasticsearchUtil.saveData(id,indexName,indexAlias,jsonData,clazz);
  25.     }
  27.     @PostMapping("/batchSync")
  28.     @ApiOperation("批量数据保存到ES-异步")
  29.     @CrossOrigin
  30.     public void batchSyncIndex(@RequestBody final ElasticsearchModel model) throws IOException {
  31.         String indexName = model.getIndexName();
  32.         String indexAlias = model.getIndexAlias();
  33.         String jsonDataList = model.getData();
  34.         Class<?> clazz = FieldUtil.getClazz(indexAlias);
  35.         elasticsearchUtil.saveDataBatchSync(indexName,indexAlias,jsonDataList,clazz);
  36.     }
  38.     @PostMapping("/batch")
  39.     @ApiOperation("批量同步数据到ES")
  40.     @CrossOrigin
  41.     public void saveBatchIndex(@RequestBody final ElasticsearchModel model) throws IOException {
  42.         String indexName = model.getIndexName();
  43.         String indexAlias = model.getIndexAlias();
  44.         String jsonDataList = model.getData();
  45.         Class<?> clazz = FieldUtil.getClazz(indexAlias);
  46.         elasticsearchUtil.saveDataBatch(indexName,indexAlias,jsonDataList,clazz);
  47.     }
  49.     @GetMapping("/get")
  50.     @ApiOperation("根据主键获取ES")
  51.     @CrossOrigin
  52.     @ApiImplicitParams({
  53.             @ApiImplicitParam(name = "indexName",value = "索引名称",required = true,paramType = "query",dataType = "String"),
  54.             @ApiImplicitParam(name = "id",value = "主键",required = true,paramType = "query",dataType = "String")
  55.     })
  56.     public HttpResultUtil<String> getDataById(@RequestParam("indexName")String indexName,@RequestParam("id")String id) {
  57.         return new HttpResultUtil<String>().ok(elasticsearchUtil.getDataById(indexName,id));
  58.     }
  60.     @PutMapping("/batch")
  61.     @ApiOperation("批量修改ES")
  62.     @CrossOrigin
  63.     public void updateDataBatch(@RequestBody final ElasticsearchModel model) throws IOException {
  64.         String indexName = model.getIndexName();
  65.         String indexAlias = model.getIndexAlias();
  66.         String jsonDataList = model.getData();
  67.         Class<?> clazz = FieldUtil.getClazz(indexAlias);
  68.         elasticsearchUtil.updateDataBatch(indexName,indexAlias,jsonDataList,clazz);
  69.     }
  71.     @PutMapping("/sync")
  72.     @ApiOperation("同步修改ES")
  73.     @CrossOrigin
  74.     public void updateData(@RequestBody final ElasticsearchModel model) {
  75.         String id = model.getId();
  76.         String indexName = model.getIndexName();
  77.         String paramJson = model.getData();
  78.         elasticsearchUtil.updateData(indexName,id,paramJson);
  79.     }
  81.     @DeleteMapping("/batch")
  82.     @ApiOperation("批量删除ES")
  83.     @CrossOrigin
  84.     public void deleteDataBatch(@RequestParam("indexName")String indexName,@RequestParam("ids")List<String> ids) {
  85.         elasticsearchUtil.deleteDataBatch(indexName,ids);
  86.     }
  88.     @DeleteMapping("/sync")
  89.     @ApiOperation("同步删除ES")
  90.     @CrossOrigin
  91.     public void deleteData(@RequestParam("indexName")String indexName,@RequestParam("id")String id) {
  92.         elasticsearchUtil.deleteData(indexName,id);
  93.     }
  95.     @DeleteMapping("/all")
  96.     @ApiOperation("清空ES")
  97.     @CrossOrigin
  98.     public void deleteAll(@RequestParam("indexName")String indexName) {
  99.         elasticsearchUtil.deleteAll(indexName);
  100.     }
  101. }

大功告成

补充：可根据自己的业务进行数据分区

标签： elasticsearch springboot
本文转载自: https://blog.csdn.net/qq_39893313/article/details/123013305
版权归原作者 k↑ 所有， 如有侵权，请联系我们删除。
登录后发布评论

“springboot 2.0集成elasticsearch 7.6.2(集群)”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航
没有找到文章内容分段。。。