草庐IT

springboot 2.0集成elasticsearch 7.6.2(集群)

k↑ 2024-02-19 原文

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

注：请点击我，获取源码

目录

一、引入依赖配置pom.xml

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

三、配置ES注解

四、配置es及swagger

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

七、测试es

一、引入依赖配置pom.xml

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger2</artifactId>
        </dependency>
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger-ui</artifactId>
        </dependency> 
       <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>7.6.2</version>
            <exclusions>
                <exclusion>
                    <groupId>org.elasticsearch</groupId>
                    <artifactId>elasticsearch</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.elasticsearch.client</groupId>
                    <artifactId>elasticsearch-rest-client</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-client</artifactId>
            <version>7.6.2</version>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch</artifactId>
            <version>7.6.2</version>
        </dependency>

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

elasticsearch:
  host: 192.168.1.1:9200,192.1.2.133:9200,192.168.1.3:9200
  cluster-name: laokou-elasticsearch
  username:
  password:
  synonym:
    path: http://192.168.1.1:9048/laokou-service/synonym

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

三、配置ES注解

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

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

participle > 需要配置什么分词器

/**
 * @author Kou Shenhai
 */
@Target({ElementType.FIELD,ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
public @interface FieldInfo {

    /**
     * 默认 keyword
     * @return
     */
    String type() default "keyword";

    /**
     * 0 not_analyzed 1 ik_smart 2.ik_max_word 3.ik-index(自定义分词器)
     * @return
     */
    int participle() default 0;

}

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

/**
 * 属性、类型、分词器
 * @author Kou Shenhai 2413176044@leimingtech.com
 * @version 1.0
 * @date 2021/2/9 0009 上午 10:20
 */
@Data
@NoArgsConstructor
public class FieldMapping {

    private String field;

    private String type;

    private Integer participle;

    public FieldMapping(String field, String type, Integer participle) {
        this.field = field;
        this.type = type;
        this.participle = participle;
    }
}

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

/**
 * 每个属性对应的类型及分词器
 * @author Kou Shenhai 2413176044@leimingtech.com
 * @version 1.0
 * @date 2021/1/24 0024 下午 7:51
 */
@Slf4j
public class FieldMappingUtil {

    public static List<FieldMapping> getFieldInfo(Class clazz) {
        return getFieldInfo(clazz, null);
    }

    public static List<FieldMapping> getFieldInfo(Class clazz, String fieldName) {
        //返回class中的所有字段（包括私有字段）
        Field[] fields = clazz.getDeclaredFields();
        //创建FieldMapping集合
        List<FieldMapping> fieldMappingList = new ArrayList<>();
        for (Field field : fields) {
            //获取字段上的FieldInfo对象
            boolean annotationPresent = field.isAnnotationPresent(FieldInfo.class);
            if (annotationPresent) {
                FieldInfo fieldInfo = field.getAnnotation(FieldInfo.class);
                //获取字段名称
                String name = field.getName();
                fieldMappingList.add(new FieldMapping(name, fieldInfo.type(), fieldInfo.participle()));
            } else {
                continue;
            }
        }
        return fieldMappingList;
    }
}

四、配置es及swagger

/**
 * es配置文件
 * @author Kou Shenhai 2413176044@leimingtech.com
 * @version 1.0
 * @date 2020/8/9 0009 下午 2:01
 */
@Configuration
public class ElasticsearchConfig {
    
    private static final String HTTP_SCHEME = "http";

    private static final Logger LOGGER = LoggerFactory.getLogger(ElasticsearchConfig.class);
    /**
     * 权限验证
     */
    final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
    /**
     * es主机
     */
    @Value("${elasticsearch.host}")
    private String[] host;

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

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

    @Bean
    public RestClientBuilder restClientBuilder() {
        HttpHost[] hosts = Arrays.stream(host)
                .map(this::makeHttpHost)
                .filter(Objects::nonNull)
                .toArray(HttpHost[]::new);
        LOGGER.info("host:{}",Arrays.toString(hosts));
        //配置权限验证
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));
        return RestClient.builder(hosts).setHttpClientConfigCallback(httpClientBuilder ->
                httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)
                .setMaxConnPerRoute(100)
                //最大连接数
                .setMaxConnTotal(100)
        ).setRequestConfigCallback(builder -> {
            builder.setConnectTimeout(-1);
            builder.setSocketTimeout(60000);
            builder.setConnectionRequestTimeout(-1);
            return builder;
        });
    }

    /**
     * 处理请求地址
     * @param address
     * @return
     */
    private HttpHost makeHttpHost(String address) {
        assert StringUtils.isNotEmpty(address);
        String[] hostAddress = address.split(":");
        if (hostAddress.length == 2) {
            String ip = hostAddress[0];
            Integer port = Integer.valueOf(hostAddress[1]);
            return new HttpHost(ip, port, HTTP_SCHEME);
        } else {
            return null;
        }
    }

    /**
     * 配置highLevelClient bean
     * @param restClientBuilder
     * @return
     */
    @Bean(name = "restHighLevelClient")
    public RestHighLevelClient restHighLevelClient(@Autowired RestClientBuilder restClientBuilder) {
        return new RestHighLevelClient(restClientBuilder);
    }
}
/**
 * @author Kou Shenhai
 */
@Configuration
@EnableSwagger2
public class SwaggerConfig {
    @Bean
    public Docket createRestApi() {
        return new Docket(DocumentationType.SWAGGER_2)
                .apiInfo(apiInfo())
                .select()
                .apis(RequestHandlerSelectors.withMethodAnnotation(ApiOperation.class))
                .paths(PathSelectors.any())
                .build();
    }
    private ApiInfo apiInfo() {
        return new ApiInfoBuilder()
                .title("API文档")
                .version("2.0.0")
                .description("API文档 - Elasticsearch服务")
                //作者信息
                .contact(new Contact("寇申海", "https://blog.csdn.net/qq_39893313", "2413176044@qq.com"))
                .build();
    }
}

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

/**
 * Elasticsearch工具类-用于操作ES
 * @author Kou Shenhai 2413176044@leimingtech.com
 * @version 1.0
 * @date 2021/1/24 0024 下午 5:42
 */
@Slf4j
@Component
public class ElasticsearchUtil {

    private static final String PRIMARY_KEY_NAME = "id";

    @Value("${elasticsearch.synonym.path}")
    private String synonymPath;

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    /**
     * 批量同步数据到ES
     * @param indexName 索引名称
     * @param indexAlias 别名
     * @param jsonDataList 数据列表
     * @param clazz 类型
     * @return
     * @throws IOException
     */
    public boolean saveDataBatch(String indexName,String indexAlias,String jsonDataList,Class clazz) throws IOException {
        //判空
        if (StringUtils.isBlank(jsonDataList)) {
            return false;
        }
        if (syncIndex(indexName, indexAlias, clazz)) {
            return false;
        }
        //3.批量操作Request
        BulkRequest bulkRequest = packBulkIndexRequest(indexName, jsonDataList);
        if (bulkRequest.requests().isEmpty()) {
            return false;
        }
        final BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
        if (bulk.hasFailures()) {
            for (BulkItemResponse item : bulk.getItems()) {
                log.error("索引[{}],主键[{}]更新操作失败，状态为:[{}],错误信息:{}",indexName,item.getId(),item.status(),item.getFailureMessage());
            }
            return false;
        }
        // 记录索引新增与修改数量
        Integer createdCount = 0;
        Integer updatedCount = 0;
        for (BulkItemResponse item : bulk.getItems()) {
            if (DocWriteResponse.Result.CREATED.equals(item.getResponse().getResult())) {
                createdCount++;
            } else if (DocWriteResponse.Result.UPDATED.equals(item.getResponse().getResult())){
                updatedCount++;
            }
        }
        log.info("索引[{}]批量同步更新成功，共新增[{}]个，修改[{}]个",indexName,createdCount,updatedCount);
        return true;
    }

    /**
     * ES修改数据
     * @param indexName 索引名称
     * @param id 主键
     * @param paramJson 参数JSON
     * @return
     */
    public boolean updateData(String indexName,String id,String paramJson) {
        UpdateRequest updateRequest = new UpdateRequest(indexName, id);
        //如果修改索引中不存在则进行新增
        updateRequest.docAsUpsert(true);
        //立即刷新数据
        updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        updateRequest.doc(paramJson,XContentType.JSON);
        try {
            UpdateResponse updateResponse = restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
            log.info("索引[{}],主键：【{}】操作结果:[{}]",indexName,id,updateResponse.getResult());
            if (DocWriteResponse.Result.CREATED.equals(updateResponse.getResult())) {
                //新增
                log.info("索引：【{}】,主键：【{}】新增成功",indexName,id);
                return true;
            } else if (DocWriteResponse.Result.UPDATED.equals(updateResponse.getResult())) {
                //修改
                log.info("索引：【{}】，主键：【{}】修改成功",indexName, id);
                return true;
            } else if (DocWriteResponse.Result.NOOP.equals(updateResponse.getResult())) {
                //无变化
                log.info("索引:[{}],主键:[{}]无变化",indexName, id);
                return true;
            }
        } catch (IOException e) {
            e.printStackTrace();
            log.error("索引：[{}],主键：【{}】，更新异常:[{}]",indexName, id,e);
            return false;
        }
        return false;
    }

    /**
     * 批量修改ES
     * @param indexName 索引名称
     * @param indexAlias 别名
     * @param jsonDataList 数据列表
     * @param clazz 类型
     * @return
     * @throws IOException
     */
    public boolean updateDataBatch(String indexName,String indexAlias, String jsonDataList,Class clazz) throws IOException {
        //1.创建索引
        boolean createIndexFlag = createIndex(indexName,indexAlias, clazz);
        if (!createIndexFlag) {
            return false;
        }
        return this.updateDataBatch(indexName,jsonDataList);
    }

    /**
     * 删除数据
     * @param indexName 索引名称
     * @param id 主键
     * @return
     */
    public boolean deleteData(String indexName,String id) {
        DeleteRequest deleteRequest = new DeleteRequest(indexName);
        deleteRequest.id(id);
        deleteRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        try {
            DeleteResponse deleteResponse = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
            if (DocWriteResponse.Result.NOT_FOUND.equals(deleteResponse.getResult())) {
                log.error("索引：【{}】，主键：【{}】删除失败",indexName, id);
                return false;
            } else {
                log.info("索引【{}】主键【{}】删除成功",indexName, id);
                return true;
            }
        } catch (IOException e) {
            e.printStackTrace();
            log.error("删除索引【{}】出现异常[{}]",indexName,e);
            return false;
        }
    }

    /**
     * 批量删除ES
     * @param indexName 索引名称
     * @param ids id列表
     * @return
     */
    public boolean deleteDataBatch(String indexName,List<String> ids) {
        if (CollectionUtils.isEmpty(ids)) {
            return false;
        }
        BulkRequest bulkRequest = packBulkDeleteRequest(indexName, ids);
        try {
            BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (bulk.hasFailures()) {
                for (BulkItemResponse item : bulk.getItems()) {
                    log.error("删除索引:[{}],主键：{}失败，信息：{}",indexName,item.getId(),item.getFailureMessage());
                }
                return false;
            }
            //记录索引新增与修改数量
            Integer deleteCount = 0;
            for (BulkItemResponse item : bulk.getItems()) {
                if (DocWriteResponse.Result.DELETED.equals(item.getResponse().getResult())) {
                    deleteCount++;
                }
            }
            log.info("批量删除索引[{}]成功，共删除[{}]个",indexName,deleteCount);
            return true;
        } catch (IOException e) {
            e.printStackTrace();
            log.error("删除索引：【{}】出现异常:{}",indexName,e);
            return false;
        }
    }

    /**
     * 组装删除操作
     * @param indexName 索引名称
     * @param ids id列表
     * @return
     */
    private BulkRequest packBulkDeleteRequest(String indexName, List<String> ids) {
        BulkRequest bulkRequest = new BulkRequest();
        bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        ids.forEach(id -> {
            DeleteRequest deleteRequest = new DeleteRequest(indexName);
            deleteRequest.id(id);
            bulkRequest.add(deleteRequest);
        });
        return bulkRequest;
    }

    /**
     * 批量修改ES
     * @param indexName 索引名称
     * @param jsonDataList json数据列表
     * @return
     */
    public boolean updateDataBatch(String indexName, String jsonDataList) {
        //判空
        if (StringUtils.isBlank(jsonDataList)) {
            return false;
        }
        BulkRequest bulkRequest = packBulkUpdateRequest(indexName, jsonDataList);
        if (bulkRequest.requests().isEmpty()) {
            return false;
        }
        try {
            //同步执行
            BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (bulk.hasFailures()) {
                for (BulkItemResponse item : bulk.getItems()) {
                    log.error("索引【{}】,主键[{}]修改操作失败，状态为【{}】,错误信息：{}",indexName,item.status(),item.getFailureMessage());
                }
                return false;
            }
            //记录索引新增与修改数量
            Integer createCount = 0;
            Integer updateCount = 0;
            for (BulkItemResponse item : bulk.getItems()) {
                if (DocWriteResponse.Result.CREATED.equals(item.getResponse().getResult())) {
                    createCount++;
                } else if (DocWriteResponse.Result.UPDATED.equals(item.getResponse().getResult())) {
                    updateCount++;
                }
            }
            log.info("索引【{}】批量修改更新成功，共新增[{}]个，修改[{}]个",indexName,createCount,updateCount);
        } catch (IOException e) {
            e.printStackTrace();
            log.error("索引【{}】批量修改更新出现异常",indexName);
            return false;
        }
        return true;
    }

    /**
     * 组装bulkUpdate
     * @param indexName 索引名称
     * @param jsonDataList 数据列表
     * @return
     */
    private BulkRequest packBulkUpdateRequest(String indexName,String jsonDataList) {
        BulkRequest bulkRequest = new BulkRequest();
        bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        JSONArray jsonArray = JSONArray.parseArray(jsonDataList);
        if (jsonArray.isEmpty()) {
            return bulkRequest;
        }
        jsonArray.forEach(o -> {
            Map<String, String> map = (Map<String, String>) o;
            UpdateRequest updateRequest = new UpdateRequest(indexName,map.get(PRIMARY_KEY_NAME));
            // 修改索引中不存在就新增
            updateRequest.docAsUpsert(true);
            updateRequest.doc(JSON.toJSONString(o),XContentType.JSON);
            bulkRequest.add(updateRequest);
        });
        return bulkRequest;
    }

    /**
     * 删除索引、新建索引
     * @param indexName 索引名称
     * @param indexAlias 别名
     * @param clazz 类型
     * @return
     * @throws IOException
     */
    private boolean syncIndex(String indexName, String indexAlias, Class clazz) throws IOException {
        //1.删除索引
        boolean deleteAllFlag = deleteIndex(indexName);
        if (!deleteAllFlag) {
            return true;
        }
        //2.创建索引
        boolean createIndexFlag = createIndex(indexName, indexAlias, clazz);
        if (!createIndexFlag) {
            return true;
        }
        return false;
    }

    /**
     * 根据主键查询ES
     * @param indexName 索引名称
     * @param id 主键
     * @return
     */
    public String getDataById(String indexName,String id) {
        //判断索引是否存在
        //1.判断索引是否存在
        boolean result = isIndexExists(indexName);
        if (!result) {
            return "";
        }
        GetRequest getRequest = new GetRequest(indexName, id);
        try {
            GetResponse getResponse = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);
            String resultJson = getResponse.getSourceAsString();
            log.info("索引【{}】主键【{}】，查询结果：【{}】",indexName,id,resultJson);
            return resultJson;
        } catch (IOException e) {
            e.printStackTrace();
            log.error("索引【{}】主键[{}]，查询异常：{}",indexName,id,e);
            return "";
        }

    }

    /**
     * 清空索引内容
     * @param indexName 索引名称
     * @return
     */
    public boolean deleteAll(String indexName) {
        //1.判断索引是否存在
        boolean result = isIndexExists(indexName);
        if (!result) {
            log.error("索引【{}】不存在，删除失败",indexName);
            return false;
        }
        DeleteRequest deleteRequest = new DeleteRequest(indexName);
        deleteRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        try {
            DeleteResponse deleteResponse = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
            if (DocWriteResponse.Result.NOT_FOUND.equals(deleteResponse.getResult())) {
                log.error("索引【{}】删除失败",indexName);
                return false;
            }
            log.info("索引【{}】删除成功",indexName);
            return true;
        } catch (IOException e) {
            e.printStackTrace();
            log.error("删除索引[{}]，出现异常[{}]",indexName,e);
            return false;
        }
    }

    /**
     * 批量数据保存到ES-异步
     * @param indexName 索引名称
     * @param indexAlias 索引别名
     * @param jsonDataList 数据列表
     * @param clazz 类型
     * @return
     * @throws IOException
     */
    public boolean saveDataBatchSync(String indexName,String indexAlias,String jsonDataList,Class clazz) throws IOException {
        //判空
        if (StringUtils.isBlank(jsonDataList)) {
            return false;
        }
        if (syncIndex(indexName, indexAlias, clazz)) {
            return false;
        }
        //3.批量操作Request
        BulkRequest bulkRequest = packBulkIndexRequest(indexName, jsonDataList);
        if (bulkRequest.requests().isEmpty()) {
            return false;
        }
        //4.异步执行
        ActionListener<BulkResponse> listener = new ActionListener<BulkResponse>() {
            @Override
            public void onResponse(BulkResponse bulkItemResponses) {
                if (bulkItemResponses.hasFailures()) {
                    for (BulkItemResponse item : bulkItemResponses.getItems()) {
                        log.error("索引【{}】,主键【{}】更新失败，状态【{}】，错误信息：{}",indexName,item.getId(),
                                item.status(),item.getFailureMessage());
                    }
                }
            }
            //失败操作
            @Override
            public void onFailure(Exception e) {
                log.error("索引【{}】批量异步更新出现异常:{}",indexName,e);
            }
        };
        restHighLevelClient.bulkAsync(bulkRequest,RequestOptions.DEFAULT,listener);
        log.info("索引批量更新索引【{}】中",indexName);
        return true;
    }

    /**
     * 删除索引
     * @param indexName 索引名称
     * @return
     * @throws IOException
     */
    public boolean deleteIndex(String indexName) throws IOException {
        //1.判断索引是否存在
        boolean result = isIndexExists(indexName);
        if (!result) {
            log.error("索引【{}】不存在，删除失败",indexName);
            return false;
        }
        //2.删除操作Request
        DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(indexName);
        deleteIndexRequest.indicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN);
        AcknowledgedResponse acknowledgedResponse = restHighLevelClient.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
        if (!acknowledgedResponse.isAcknowledged()) {
            log.error("索引【{}】删除失败",indexName);
            return false;
        }
        log.info("索引【{}】删除成功",indexName);
        return true;
    }

    /**
     * 批量操作的Request
     * @param indexName 索引名称
     * @param jsonDataList json数据列表
     * @return
     */
    private BulkRequest packBulkIndexRequest(String indexName,String jsonDataList) {
        BulkRequest bulkRequest = new BulkRequest();
        //IMMEDIATE > 请求向es提交数据，立即进行数据刷新<实时性高，资源消耗高>
        //WAIT_UNTIL >  请求向es提交数据，等待数据完成刷新<实时性高，资源消耗低>
        //NONE > 默认策略<实时性低>
        bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        JSONArray jsonArray = JSONArray.parseArray(jsonDataList);
        if (jsonArray.isEmpty()) {
            return bulkRequest;
        }
        //循环数据封装bulkRequest
        jsonArray.forEach(obj ->{
            final Map<String, String> map = (Map<String, String>) obj;
            IndexRequest indexRequest = new IndexRequest(indexName);
            indexRequest.source(JSON.toJSONString(obj),XContentType.JSON);
            indexRequest.id(map.get(PRIMARY_KEY_NAME));
            bulkRequest.add(indexRequest);
        });
        return bulkRequest;
    }

    /**
     * 创建索引
     * @param indexName 索引名称
     * @param indexAlias 别名
     * @param clazz 类型
     * @return
     * @throws IOException
     */
    public boolean createIndex(String indexName,String indexAlias,Class clazz) throws IOException {
        //判断索引是否存在
        boolean result = isIndexExists(indexName);
        if (!result) {
            boolean createResult = createIndexAndCreateMapping(indexName,indexAlias, FieldMappingUtil.getFieldInfo(clazz));
            if (!createResult) {
                log.info("索引【{}】创建失败",indexName);
                return false;
            }
        }
        log.info("索引：[{}]创建成功",indexName);
        return true;
    }

    /**
     * 数据同步到ES
     * @param id 主键
     * @param indexName 索引名称
     * @param jsonData json数据
     * @param clazz 类型
     * @return
     */
    public boolean saveData(String id,String indexName,String indexAlias,String jsonData,Class clazz) throws IOException {
        //1.创建索引
        boolean createIndexFlag = createIndex(indexName,indexAlias, clazz);
        if (!createIndexFlag) {
            return false;
        }
        //2.创建操作Request
        IndexRequest indexRequest = new IndexRequest(indexName);
        //3.配置相关信息
        indexRequest.source(jsonData, XContentType.JSON);
        //IMMEDIATE > 立即刷新
        indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        indexRequest.id(id);
        IndexResponse response = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
        //4.判断索引是新增还是修改
        if (IndexResponse.Result.CREATED.equals(response.getResult())) {
            log.info("索引【{}】保存成功",indexName);
            return true;
        } else if (IndexResponse.Result.UPDATED.equals(response.getResult())) {
            log.info("索引【{}】修改成功",indexName);
            return true;
        }
        return false;
    }

    /**
     * 判断索引是否存在
     * @param indexName 索引名称
     * @return
     */
    public boolean isIndexExists(String indexName) {
        try {
            GetIndexRequest getIndexRequest = new GetIndexRequest(indexName);
            return restHighLevelClient.indices().exists(getIndexRequest, RequestOptions.DEFAULT);
        }catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 创建索引设置相关配置信息
     * @param indexName 索引名称
     * @param indexAlias 索引别名
     * @param fieldMappingList 数据列表
     * @return
     * @throws IOException
     */
    private boolean createIndexAndCreateMapping(String indexName,String indexAlias, List<FieldMapping> fieldMappingList) throws IOException {
        //封装es索引的mapping
        XContentBuilder mapping = packEsMapping(fieldMappingList, null);
        mapping.endObject().endObject();
        mapping.close();
        //进行索引的创建
        CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);
        //配置分词器
        XContentBuilder settings = packSettingMapping();
        XContentBuilder aliases = packEsAliases(indexAlias);
        log.info("索引配置脚本：{}",settings);
        log.info("索引字段内容：{}",mapping);
        createIndexRequest.settings(settings);
        createIndexRequest.source(mapping);
        createIndexRequest.aliases(aliases);
        //同步方式创建索引
        CreateIndexResponse createIndexResponse = restHighLevelClient.indices().create(createIndexRequest,RequestOptions.DEFAULT);
        boolean acknowledged = createIndexResponse.isAcknowledged();
        if (acknowledged) {
            log.info("索引:{}创建成功", indexName);
            return true;
        } else {
            log.error("索引:{}创建失败", indexName);
            return false;
        }
    }

    /**
     * 配置ES别名
     * @author Kou Shenhai
     * @param alias 别名
     * @return
     * @throws IOException
     */
    private XContentBuilder packEsAliases(String alias) throws IOException{
        XContentBuilder aliases = XContentFactory.jsonBuilder().startObject()
                .startObject(alias).endObject();
        aliases.endObject();
        aliases.close();
        return aliases;
    }

    /**
     * 配置Mapping
     * @param fieldMappingList 组装的实体类信息
     * @param mapping
     * @return
     * @throws IOException
     */
    private XContentBuilder packEsMapping(List<FieldMapping> fieldMappingList,XContentBuilder mapping) throws IOException {
        if (mapping == null) {
            //如果对象是空，首次进入，设置开始结点
            mapping = XContentFactory.jsonBuilder().startObject()
                    .field("dynamic",true)
                    .startObject("properties");
        }
        //循环实体对象的类型集合封装ES的Mapping
        for (FieldMapping fieldMapping : fieldMappingList) {
            String field = fieldMapping.getField();
            String dataType = fieldMapping.getType();
            Integer participle = fieldMapping.getParticiple();
            //设置分词规则
            if (Constant.NOT_ANALYZED.equals(participle)) {
                if (FieldTypeEnum.DATE.getValue().equals(dataType)) {
                    mapping.startObject(field)
                            .field("type", dataType)
                            .field("format","yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis")
                            .endObject();
                } else {
                    mapping.startObject(field)
                            .field("type", dataType)
                            .endObject();
                }
            } else if (Constant.IK_INDEX.equals(participle)) {
                mapping.startObject(field)
                        .field("type",dataType)
                        .field("eager_global_ordinals",true)
                        //fielddata=true 用来解决text字段不能进行聚合操作
                        .field("fielddata",true)
                        .field("boost",100.0)
                        .field("analyzer","ik-index-synonym")
                        .field("search_analyzer","ik-search-synonym")
                        .startObject("fields").startObject("pinyin")
                        .field("term_vector", "with_positions_offsets")
                        .field("analyzer","ik-search-pinyin")
                        .field("type",dataType)
                        .field("boost",100.0)
                        .endObject().endObject()
                        .endObject();
            }
        }
        return mapping;
    }

    /**
     * 配置Settings
     * @return
     * @throws IOException
     */
    private XContentBuilder packSettingMapping() throws IOException {
        XContentBuilder setting = XContentFactory.jsonBuilder().startObject()
                .startObject("index")
                .field("number_of_shards",5)
                .field("number_of_replicas",1)
                .field("refresh_interval","120s")
                .endObject()
                .startObject("analysis");
        //ik分词 同义词 拼音
        setting.startObject("analyzer")
                .startObject("ik-search-pinyin")
                .field("type","custom")
                .field("tokenizer","ik_smart")
                .field("char_filter",new String[] {"html_strip"})
                .field("filter", new String[]{"laokou-pinyin","word_delimiter","lowercase", "asciifolding"})
                .endObject();
        setting.startObject("ik-index-synonym")
                .field("type","custom")
                .field("tokenizer","ik_max_word")
                .field("char_filter",new String[] {"html_strip"})
                .field("filter", new String[]{"laokou-remote-synonym"})
                .endObject();
        setting.startObject("ik-search-synonym")
                .field("type","custom")
                .field("tokenizer","ik_smart")
                .field("char_filter",new String[] {"html_strip"})
                .field("filter", new String[]{"laokou-remote-synonym"})
                .endObject();
        setting.endObject();
        //设置拼音分词器 同义词分词
        setting.startObject("filter")
                .startObject("laokou-pinyin")
                .field("type", "pinyin")
                .field("keep_first_letter", false)
                .field("keep_separate_first_letter", false)
                .field("keep_full_pinyin", true)
                .field("keep_original", false)
                .field("keep_joined_full_pinyin",true)
                .field("limit_first_letter_length", 16)
                .field("lowercase", true)
                .field("remove_duplicated_term", true)
                .endObject()
                .startObject("laokou-remote-synonym")
                .field("type","dynamic_synonym")
                .field("synonyms_path", synonymPath)
                .field("interval",120)
                .field("dynamic_reload",true)
                .endObject()
                .endObject();
        setting.endObject().endObject();
        setting.close();
        return setting;
    }
}

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

六、索引管理工具类

/**
 * 索引管理
 * @author Kou Shenhai 2413176044@leimingtech.com
 * @version 1.0
 * @date 2021/10/31 0031 上午 10:11
 */
public class FieldUtil {

    public static final String MESSAGE_INDEX = "message";

    private static final Map<String,Class<?>> classMap = new HashMap<>(16);

    static {
        classMap.put(FieldUtil.MESSAGE_INDEX, MessageIndex.class);
    }

    public static Class<?> getClazz(final String indexName) {
        return classMap.getOrDefault(indexName,Object.class);
    }
}

七、测试es

/**
 * Elasticsearch API 服务
 * @author Kou Shenhai 2413176044@leimingtech.com
 * @version 1.0
 * @date 2021/2/8 0008 下午 6:33
 */
@RestController
@RequestMapping("/api")
@Api(tags = "Elasticsearch API 服务")
public class ElasticsearchController {

    @Autowired
    private ElasticsearchUtil elasticsearchUtil;

    @PostMapping("/sync")
    @ApiOperation("同步数据到ES")
    @CrossOrigin
    public void syncIndex(@RequestBody final ElasticsearchModel model) throws IOException {
        String id = model.getId();
        String indexName = model.getIndexName();
        String indexAlias = model.getIndexAlias();
        String jsonData = model.getData();
        Class<?> clazz = FieldUtil.getClazz(indexAlias);
        elasticsearchUtil.saveData(id,indexName,indexAlias,jsonData,clazz);
    }

    @PostMapping("/batchSync")
    @ApiOperation("批量数据保存到ES-异步")
    @CrossOrigin
    public void batchSyncIndex(@RequestBody final ElasticsearchModel model) throws IOException {
        String indexName = model.getIndexName();
        String indexAlias = model.getIndexAlias();
        String jsonDataList = model.getData();
        Class<?> clazz = FieldUtil.getClazz(indexAlias);
        elasticsearchUtil.saveDataBatchSync(indexName,indexAlias,jsonDataList,clazz);
    }

    @PostMapping("/batch")
    @ApiOperation("批量同步数据到ES")
    @CrossOrigin
    public void saveBatchIndex(@RequestBody final ElasticsearchModel model) throws IOException {
        String indexName = model.getIndexName();
        String indexAlias = model.getIndexAlias();
        String jsonDataList = model.getData();
        Class<?> clazz = FieldUtil.getClazz(indexAlias);
        elasticsearchUtil.saveDataBatch(indexName,indexAlias,jsonDataList,clazz);
    }

    @GetMapping("/get")
    @ApiOperation("根据主键获取ES")
    @CrossOrigin
    @ApiImplicitParams({
            @ApiImplicitParam(name = "indexName",value = "索引名称",required = true,paramType = "query",dataType = "String"),
            @ApiImplicitParam(name = "id",value = "主键",required = true,paramType = "query",dataType = "String")
    })
    public HttpResultUtil<String> getDataById(@RequestParam("indexName")String indexName,@RequestParam("id")String id) {
        return new HttpResultUtil<String>().ok(elasticsearchUtil.getDataById(indexName,id));
    }

    @PutMapping("/batch")
    @ApiOperation("批量修改ES")
    @CrossOrigin
    public void updateDataBatch(@RequestBody final ElasticsearchModel model) throws IOException {
        String indexName = model.getIndexName();
        String indexAlias = model.getIndexAlias();
        String jsonDataList = model.getData();
        Class<?> clazz = FieldUtil.getClazz(indexAlias);
        elasticsearchUtil.updateDataBatch(indexName,indexAlias,jsonDataList,clazz);
    }

    @PutMapping("/sync")
    @ApiOperation("同步修改ES")
    @CrossOrigin
    public void updateData(@RequestBody final ElasticsearchModel model) {
        String id = model.getId();
        String indexName = model.getIndexName();
        String paramJson = model.getData();
        elasticsearchUtil.updateData(indexName,id,paramJson);
    }

    @DeleteMapping("/batch")
    @ApiOperation("批量删除ES")
    @CrossOrigin
    public void deleteDataBatch(@RequestParam("indexName")String indexName,@RequestParam("ids")List<String> ids) {
        elasticsearchUtil.deleteDataBatch(indexName,ids);
    }

    @DeleteMapping("/sync")
    @ApiOperation("同步删除ES")
    @CrossOrigin
    public void deleteData(@RequestParam("indexName")String indexName,@RequestParam("id")String id) {
        elasticsearchUtil.deleteData(indexName,id);
    }

    @DeleteMapping("/all")
    @ApiOperation("清空ES")
    @CrossOrigin
    public void deleteAll(@RequestParam("indexName")String indexName) {
        elasticsearchUtil.deleteAll(indexName);
    }
}

大功告成

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

elasticsearchspringbootindexName34String

有关springboot 2.0集成elasticsearch 7.6.2(集群)的更多相关文章

  1. ruby-on-rails - 如何使辅助方法在 Rails 集成测试中可用？ - 2

    我在app/helpers/sessions_helper.rb中有一个帮助程序文件，其中包含一个方法my_preference，它返回当前登录用户的首选项。我想在集成测试中访问该方法。例如，这样我就可以在测试中使用getuser_path(my_preference)。在其他帖子中，我读到这可以通过在测试文件中包含requiresessions_helper来实现，但我仍然收到错误NameError:undefinedlocalvariableormethod'my_preference'.我做错了什么？require'test_helper'require'sessions_hel

  2. ruby-on-rails - 我如何将 Hoptoad 与 DelayedJob 和 DaemonSpawn 集成？ - 2

    我一直很高兴地使用DelayedJob习惯用法:foo.send_later(:bar)这会调用DelayedJob进程中对象foo的方法bar。我一直在使用DaemonSpawn在我的服务器上启动DelayedJob进程。但是...如果foo抛出异常，Hoptoad不会捕获它。这是任何这些包中的错误...还是我需要更改某些配置...或者我是否需要在DS或DJ中插入一些异常处理来调用Hoptoad通知程序？回应下面的第一条评论。classDelayedJobWorker 最佳答案 尝试monkeypatchingDelayed::W

  3. jenkins部署1--jenkins+gitee持续集成 - 2

    前置步骤我们都操作完了，这篇开始介绍jenkins的集成。话不多说，看操作1、登录进入jenkins后会让你选择安装插件，选择第一个默认的就行。安装完成后设置账号密码，重新登录。2、配置JDK和Git都需要执行路径，所以需要先把执行路径找到，先进入服务器的docker容器，2.1JDK的路径root@69eef9ee86cf:/usr/bin#echo$JAVA_HOME/usr/local/openjdk-82.2Git的路径root@69eef9ee86cf:/#whichgit/usr/bin/git3、先配置JDK和Git。点击：ManageJenkins>>GlobalToolCon

  4. ruby - Rails Elasticsearch 聚合 - 2

    不知何故，我似乎无法获得包含我的聚合的响应...使用curl它按预期工作:HBZUMB01$curl-XPOST"http://localhost:9200/contents/_search"-d'{"size":0,"aggs":{"sport_count":{"value_count":{"field":"dwid"}}}}'我收到回复:{"took":4,"timed_out":false,"_shards":{"total":5,"successful":5,"failed":0},"hits":{"total":90,"max_score":0.0,"hits":[]},"a

  5. 三分钟集成 TapTap 防沉迷 SDK（Unity 版） - 2

    三分钟集成Tap防沉迷SDK（Unity版）一、SDK介绍基于国家对上线所有游戏必须增加防沉迷功能的政策下，TapTap推出防沉迷SDK，供游戏开发者进行接入；允许未成年用户在周五、六、日以及法定节假日晚上8:00-9:00进行游戏，防沉谜时间段进入游戏会弹窗进行提示！开发环境要求：Unity2019.4或更高版本iOS10或更高版本Android5.0（APIlevel21）或更高版本🔗Unity集成Demo参考链接🔗UnityTapSDK功能体验APK下载链接二、集成前准备1.创建应用进入开发者后台,按照提示开始创建应用；2.开通服务在使用TDS实名认证和防沉迷服务之前，需要在上面创建的应

  6. elasticsearch源码关于TransportSearchAction【阶段三】 - 2

    1.回顾.TransportServicepublicclassTransportServiceextendsAbstractLifecycleComponentTransportService：方法：1publicfinalTextendsTransportResponse>voidsendRequest(finalTransport.Connectionconnection,finalStringaction,finalTransportRequestrequest,finalTransportRequestOptionsoptions,TransportResponseHandlerT>

  7. ruby-on-rails - RailsTutorial - 第 8.4.3 章 - 在集成测试中添加用户后未清除测试数据库 - 2

    我被这个难住了。到目前为止教程中的一切都进行得很顺利，但是当我将这段代码添加到我的/spec/requests/users_spec.rb文件中时，事情开始变得糟糕:describe"success"doit"shouldmakeanewuser"dolambdadovisitsignup_pathfill_in"Name",:with=>"ExampleUser"fill_in"Email",:with=>"ryan@example.com"fill_in"Password",:with=>"foobar"fill_in"Confirmation",:with=>"foobar"cl

  8. ruby-on-rails - 使用 Rails (Tire) 和 ElasticSearch 进行模糊字符串匹配 - 2

    我有一个Rails应用程序，现在设置了ElasticSearch和Tiregem以在模型上进行搜索，我想知道我应该如何设置我的应用程序以对模型中的某些索引进行模糊字符串匹配。我将我的模型设置为索引标题、描述等内容，但我想对其中一些进行模糊字符串匹配，但我不确定在何处进行此操作。如果您想发表评论，我将在下面包含我的代码!谢谢!在Controller中:defsearch@resource=Resource.search(params[:q],:page=>(params[:page]||1),:per_page=>15,load:true)end在模型中:classResource'Us

  9. ruby-on-rails - 将 Angular JS 与 Rails 集成 - 2

    我需要一些指导来了解如何将Angular整合到rails中。选择Rails的原因:我喜欢他们偏执的做事方式。还有迁移，gem真的很酷。使用angular的原因:我正在研究和寻找最适合SPA的框架。Backbone似乎太抽象了。我不得不在Angular和Ember之间做出选择。我首先开始阅读Angular，它对我来说很有意义。所以我从来没有去读过关于ember的文章。使用Angular和Rails的原因:我研究并尝试使用小型框架，例如grape、slim(是的，我也使用php)。但我觉得需要坚持项目的长期范围。我个人喜欢用Rails的方式做事。这就是我需要帮助的地方，我在Rails4中有

  10. ruby - 在 Maven 集成中运行 Ruby 单元测试 - 2

    有没有人有在Maven中运行用Ruby编写的单元测试的经验。任何输入，如要使用的库/maven插件，将不胜感激!我们已经在使用Maven+hudson+Junit。但是我们正在引入Ruby单元测试，找不到任何同样好的组合。 最佳答案 我建议让Maven使用ExecMavenPlugin启动rake测试(exec:exec目标)并使用ci_reportergem生成单元测试结果的XML文件，Hudson、Bamboo等可以读取该文件，以与JUnit测试相同的格式显示测试结果。如果您不需要使用mvntest运行Ruby测试，您也可以只使

随机推荐