springboot 2.0集成elasticsearch 7.6.2(集群)(下)

本文涉及的产品
检索分析服务 Elasticsearch 版,2核4GB开发者规格 1个月
简介: springboot 2.0集成elasticsearch 7.6.2(集群)(下)

正文


五、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.mapping("_doc", 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);
    }
}


大功告成


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


111.png






相关实践学习
使用阿里云Elasticsearch体验信息检索加速
通过创建登录阿里云Elasticsearch集群,使用DataWorks将MySQL数据同步至Elasticsearch,体验多条件检索效果,简单展示数据同步和信息检索加速的过程和操作。
ElasticSearch 入门精讲
ElasticSearch是一个开源的、基于Lucene的、分布式、高扩展、高实时的搜索与数据分析引擎。根据DB-Engines的排名显示,Elasticsearch是最受欢迎的企业搜索引擎,其次是Apache Solr(也是基于Lucene)。 ElasticSearch的实现原理主要分为以下几个步骤: 用户将数据提交到Elastic Search 数据库中 通过分词控制器去将对应的语句分词,将其权重和分词结果一并存入数据 当用户搜索数据时候,再根据权重将结果排名、打分 将返回结果呈现给用户 Elasticsearch可以用于搜索各种文档。它提供可扩展的搜索,具有接近实时的搜索,并支持多租户。
目录
相关文章
|
15天前
|
前端开发 Java 应用服务中间件
从零手写实现 tomcat-08-tomcat 如何与 springboot 集成?
该文是一系列关于从零开始手写实现 Apache Tomcat 的教程概述。作者希望通过亲自动手实践理解 Tomcat 的核心机制。文章讨论了 Spring Boot 如何实现直接通过 `main` 方法启动,Spring 与 Tomcat 容器的集成方式,以及两者生命周期的同步原理。文中还提出了实现 Tomcat 的启发,强调在设计启动流程时确保资源的正确加载和初始化。最后提到了一个名为 mini-cat(嗅虎)的简易 Tomcat 实现项目,开源于 [GitHub](https://github.com/houbb/minicat)。
|
15天前
|
前端开发 Java 应用服务中间件
从零手写实现 tomcat-08-tomcat 如何与 springboot 集成?
本文探讨了Spring Boot如何实现像普通Java程序一样通过main方法启动,关键在于Spring Boot的自动配置、内嵌Servlet容器(如Tomcat)以及`SpringApplication`类。Spring与Tomcat集成有两种方式:独立模式和嵌入式模式,两者通过Servlet规范、Spring MVC协同工作。Spring和Tomcat的生命周期同步涉及启动、运行和关闭阶段,通过事件和监听器实现。文章鼓励读者从实现Tomcat中学习资源管理和生命周期管理。此外,推荐了Netty权威指南系列文章,并提到了一个名为mini-cat的简易Tomcat实现项目。
|
1天前
|
搜索推荐 Java 数据库
springboot集成ElasticSearch的具体操作(系统全文检索)
springboot集成ElasticSearch的具体操作(系统全文检索)
|
1天前
|
消息中间件 Java Spring
Springboot 集成Rabbitmq之延时队列
Springboot 集成Rabbitmq之延时队列
4 0
|
1天前
|
网络协议 Java Spring
Springboot 集成websocket
Springboot 集成websocket
5 0
|
2天前
|
XML Java 数据库
springboot集成flowable
springboot集成flowable
|
3天前
|
NoSQL Java Redis
springboot之RedisTemplate的访问单机,哨兵,集群模式
以上是配置RedisTemplate以连接到单机、哨兵和集群模式的示例。在实际应用中,还可以根据需求配置连接池、序列化方式、超时等其他参数。
22 0
|
11天前
|
缓存 NoSQL Java
springboot业务开发--springboot集成redis解决缓存雪崩穿透问题
该文介绍了缓存使用中可能出现的三个问题及解决方案:缓存穿透、缓存击穿和缓存雪崩。为防止缓存穿透,可校验请求数据并缓存空值;缓存击穿可采用限流、热点数据预加载或加锁策略;缓存雪崩则需避免同一时间大量缓存失效,可设置随机过期时间。文章还提及了Spring Boot中Redis缓存的配置,包括缓存null值、使用前缀和自定义过期时间,并提供了改造代码以实现缓存到期时间的个性化设置。
|
12天前
|
SQL JSON DataWorks
DataWorks产品使用合集之DataWorks 数据集成任务中,将数据同步到 Elasticsearch(ES)中,并指定 NESTED 字段中的 properties 类型如何解决
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
21 0
|
16天前
|
安全 Java API
SpringBoot 实现 elasticsearch 索引操作(RestHighLevelClient 的应用)
SpringBoot 实现 elasticsearch 索引操作(RestHighLevelClient 的应用)
19 1