list = new ArrayList<>();
+ for (MultiGetItemResponse itemResponse : multiGetResponse.getResponses()) {
+ if (itemResponse.isFailed()) {
+ log.error("通过id获取文档失败", itemResponse.getFailure().getFailure());
+ } else {
+ T entity = toPojo(itemResponse.getResponse(), clazz);
+ if (entity != null) {
+ list.add(entity);
+ }
+ }
+ }
+ return list;
+ }
+
+ public long count(String index, String type, SearchSourceBuilder builder) throws IOException {
+ SearchResponse response = query(index, type, builder);
+ if (response == null || response.status() != RestStatus.OK) {
+ return 0L;
+ }
+ SearchHits searchHits = response.getHits();
+ return searchHits.getTotalHits();
+ }
+
+ public long count(String index, String type, QueryBuilder queryBuilder) throws IOException {
+ SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
+ searchSourceBuilder.query(queryBuilder);
+ return count(index, type, searchSourceBuilder);
+ }
+
+ public SearchResponse query(String index, String type, SearchSourceBuilder builder) throws IOException {
+ SearchRequest request = new SearchRequest(index).types(type);
+ request.source(builder);
+ return client.search(request, RequestOptions.DEFAULT);
+ }
+
+ public SearchResponse query(SearchRequest request) throws IOException {
+ return client.search(request, RequestOptions.DEFAULT);
+ }
+
+ /**
+ * from+size 分页
+ *
+ * 注：在深分页的场景下，效率很低（一般超过 1万条数据就不适用了）
+ */
+ public PageData pojoPage(String index, String type, SearchSourceBuilder builder, Class clazz)
+ throws IOException {
+ SearchResponse response = query(index, type, builder);
+ if (response == null || response.status() != RestStatus.OK) {
+ return null;
+ }
+
+ List content = toPojoList(response, clazz);
+ SearchHits searchHits = response.getHits();
+ int from = builder.from();
+ int size = builder.size();
+ int page = from / size + (from % size == 0 ? 0 : 1) + 1;
+ return new PageData<>(page, size, searchHits.getTotalHits(), content);
+ }
+
+ /**
+ * from+size 分页
+ *
+ * 注：在深分页的场景下，效率很低（一般超过 1万条数据就不适用了）
+ */
+ public PageData pojoPage(String index, String type, int from, int size, QueryBuilder queryBuilder,
+ Class clazz) throws IOException {
+ SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
+ searchSourceBuilder.query(queryBuilder);
+ searchSourceBuilder.from(from);
+ searchSourceBuilder.size(size);
+ return pojoPage(index, type, searchSourceBuilder, clazz);
+ }
+
+ /**
+ * search after 分页
+ */
+ public ScrollData pojoPageByScrollId(String index, String type, String scrollId,
+ int size,
+ QueryBuilder queryBuilder, Class clazz) throws IOException {
+
+ SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
+ searchSourceBuilder.size(size);
+ searchSourceBuilder.sort(BaseEsEntity.DOC_ID, SortOrder.ASC);
+ if (StrUtil.isNotBlank(scrollId)) {
+ BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
+ boolQueryBuilder.must(queryBuilder).must(QueryBuilders.rangeQuery(BaseEsEntity.DOC_ID).gt(scrollId));
+ searchSourceBuilder.query(boolQueryBuilder);
+ } else {
+ searchSourceBuilder.query(queryBuilder);
+ }
+
+ SearchResponse response = query(index, type, searchSourceBuilder);
+ if (response == null || response.status() != RestStatus.OK) {
+ return null;
+ }
+ List content = toPojoList(response, clazz);
+ ScrollData scrollData = new ScrollData<>();
+ scrollData.setSize(size);
+ scrollData.setTotal(response.getHits().getTotalHits());
+ scrollData.setContent(content);
+ if (CollectionUtil.isNotEmpty(content)) {
+ T lastEntity = content.get(content.size() - 1);
+ scrollData.setScrollId(lastEntity.getDocId());
+ }
+ return scrollData;
+ }
+
+ /**
+ * 首次滚动查询批量查询，但是不适用与搜索，仅用于批查询
+ **/
+ public ScrollData pojoScrollBegin(String index, String type, SearchSourceBuilder searchBuilder,
+ Class clazz) throws IOException {
+
+ int scrollTime = 10;
+ final Scroll scroll = new Scroll(TimeValue.timeValueSeconds(scrollTime));
+ SearchRequest request = new SearchRequest(index);
+ request.types(type);
+ request.source(searchBuilder);
+ request.scroll(scroll);
+ SearchResponse response = client.search(request, RequestOptions.DEFAULT);
+ if (response == null || response.status() != RestStatus.OK) {
+ return null;
+ }
+ List content = toPojoList(response, clazz);
+ ScrollData scrollData = new ScrollData<>();
+ scrollData.setSize(searchBuilder.size());
+ scrollData.setTotal(response.getHits().getTotalHits());
+ scrollData.setScrollId(response.getScrollId());
+ scrollData.setContent(content);
+ return scrollData;
+ }
+
+ /**
+ * 知道ScrollId之后，后续根据scrollId批量查询
+ **/
+ public ScrollData pojoScroll(String scrollId, SearchSourceBuilder searchBuilder, Class clazz)
+ throws IOException {
+
+ int scrollTime = 10;
+ final Scroll scroll = new Scroll(TimeValue.timeValueSeconds(scrollTime));
+ SearchScrollRequest request = new SearchScrollRequest(scrollId);
+ request.scroll(scroll);
+ SearchResponse response = client.scroll(request, RequestOptions.DEFAULT);
+ if (response == null || response.status() != RestStatus.OK) {
+ return null;
+ }
+ List content = toPojoList(response, clazz);
+ ScrollData scrollData = new ScrollData<>();
+ scrollData.setSize(searchBuilder.size());
+ scrollData.setTotal(response.getHits().getTotalHits());
+ scrollData.setScrollId(response.getScrollId());
+ scrollData.setContent(content);
+ return scrollData;
+ }
+
+ public boolean pojoScrollEnd(String scrollId) throws IOException {
+ ClearScrollRequest request = new ClearScrollRequest();
+ request.addScrollId(scrollId);
+ ClearScrollResponse response = client.clearScroll(request, RequestOptions.DEFAULT);
+ if (response != null) {
+ return response.isSucceeded();
+ }
+ return false;
+ }
+
+ public T toPojo(GetResponse response, Class clazz) {
+ if (null == response || StrUtil.isBlank(response.getSourceAsString())) {
+ return null;
+ } else {
+ return JsonUtil.toBean(response.getSourceAsString(), clazz);
+ }
+ }
+
+ public List toPojoList(SearchResponse response, Class clazz) {
+ if (response == null || response.status() != RestStatus.OK) {
+ return new ArrayList<>(0);
+ }
+ if (ArrayUtil.isEmpty(response.getHits().getHits())) {
+ return new ArrayList<>(0);
+ }
+ return Stream.of(response.getHits().getHits())
+ .map(hit -> JsonUtil.toBean(hit.getSourceAsString(), clazz))
+ .collect(Collectors.toList());
+ }
+
+ public Map toMap(T entity) {
+ return JsonUtil.toMap(JsonUtil.toString(entity));
+ }
+
+ @Override
+ public synchronized void close() {
+ if (null == client) {
+ return;
+ }
+ IoUtil.close(client);
+ }
+
+}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/config/ElasticsearchConfig.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/config/ElasticsearchConfig.java
new file mode 100644
index 00000000..791bbf1c
--- /dev/null
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/config/ElasticsearchConfig.java
@@ -0,0 +1,44 @@
+package io.github.dunwu.javadb.elasticsearch.config;
+
+import cn.hutool.core.util.StrUtil;
+import io.github.dunwu.javadb.elasticsearch.ElasticsearchFactory;
+import io.github.dunwu.javadb.elasticsearch.ElasticsearchTemplate;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.List;
+
+/**
+ * ES 配置
+ *
+ * @author Zhang Peng
+ * @date 2024-02-07
+ */
+@Configuration
+@ComponentScan(value = "io.github.dunwu.javadb.elasticsearch.mapper")
+public class ElasticsearchConfig {
+
+ @Value("${es.hosts:#{null}}")
+ private String hostsConfig;
+
+ @Bean("restHighLevelClient")
+ @ConditionalOnMissingBean
+ public RestHighLevelClient restHighLevelClient() {
+ if (hostsConfig == null) {
+ return ElasticsearchFactory.newRestHighLevelClient();
+ } else {
+ List hosts = StrUtil.split(hostsConfig, ",");
+ return ElasticsearchFactory.newRestHighLevelClient(hosts);
+ }
+ }
+
+ @Bean("elasticsearchTemplate")
+ public ElasticsearchTemplate elasticsearchTemplate(RestHighLevelClient restHighLevelClient) {
+ return ElasticsearchFactory.newElasticsearchTemplate(restHighLevelClient);
+ }
+
+}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/config/EnableElasticsearch.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/config/EnableElasticsearch.java
new file mode 100644
index 00000000..c2c24479
--- /dev/null
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/config/EnableElasticsearch.java
@@ -0,0 +1,26 @@
+package io.github.dunwu.javadb.elasticsearch.config;
+
+import org.springframework.context.annotation.EnableAspectJAutoProxy;
+import org.springframework.context.annotation.Import;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * 启动 Elasticsearch 配置注解
+ *
+ * @author Zhang Peng
+ * @date 2023-06-30
+ */
+@Target({ ElementType.TYPE })
+@Retention(RetentionPolicy.RUNTIME)
+@EnableAspectJAutoProxy(
+ proxyTargetClass = false
+)
+@Import({ ElasticsearchConfig.class })
+@Documented
+public @interface EnableElasticsearch {
+}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/constant/CodeMsg.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/constant/CodeMsg.java
new file mode 100644
index 00000000..96e46f6c
--- /dev/null
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/constant/CodeMsg.java
@@ -0,0 +1,15 @@
+package io.github.dunwu.javadb.elasticsearch.constant;
+
+/**
+ * 请求 / 应答状态接口
+ *
+ * @author Zhang Peng
+ * @since 2019-06-06
+ */
+public interface CodeMsg {
+
+ int getCode();
+
+ String getMsg();
+
+}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/constant/ResultCode.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/constant/ResultCode.java
new file mode 100644
index 00000000..d4822fb8
--- /dev/null
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/constant/ResultCode.java
@@ -0,0 +1,97 @@
+package io.github.dunwu.javadb.elasticsearch.constant;
+
+import cn.hutool.core.util.StrUtil;
+
+import java.util.stream.Stream;
+
+/**
+ * 系统级错误码
+ *
+ * @author Zhang Peng
+ * @see HTTP 状态码
+ * @see 腾讯开放平台错误码
+ * @see 新浪开放平台错误码
+ * @see 支付宝开放平台API
+ * @see 微信开放平台错误码
+ * @since 2019-04-11
+ */
+public enum ResultCode implements CodeMsg {
+
+ OK(0, "成功"),
+
+ PART_OK(1, "部分成功"),
+
+ FAIL(-1, "失败"),
+
+ // -----------------------------------------------------
+ // 系统级错误码
+ // -----------------------------------------------------
+
+ ERROR(1000, "服务器错误"),
+
+ PARAM_ERROR(1001, "参数错误"),
+
+ TASK_ERROR(1001, "调度任务错误"),
+
+ CONFIG_ERROR(1003, "配置错误"),
+
+ REQUEST_ERROR(1004, "请求错误"),
+
+ IO_ERROR(1005, "IO 错误"),
+
+ // -----------------------------------------------------
+ // 2000 ~ 2999 数据库错误
+ // -----------------------------------------------------
+
+ DATA_ERROR(2000, "数据库错误"),
+
+ // -----------------------------------------------------
+ // 3000 ~ 3999 三方错误
+ // -----------------------------------------------------
+
+ THIRD_PART_ERROR(3000, "三方错误"),
+
+ // -----------------------------------------------------
+ // 3000 ~ 3999 认证错误
+ // -----------------------------------------------------
+
+ AUTH_ERROR(4000, "认证错误");
+
+ private final int code;
+
+ private final String msg;
+
+ ResultCode(int code, String msg) {
+ this.code = code;
+ this.msg = msg;
+ }
+
+ @Override
+ public int getCode() {
+ return code;
+ }
+
+ @Override
+ public String getMsg() {
+ return msg;
+ }
+
+ public static String getNameByCode(int code) {
+ return Stream.of(ResultCode.values()).filter(item -> item.getCode() == code).findFirst()
+ .map(ResultCode::getMsg).orElse(null);
+ }
+
+ public static ResultCode getEnumByCode(int code) {
+ return Stream.of(ResultCode.values()).filter(item -> item.getCode() == code).findFirst().orElse(null);
+ }
+
+ public static String getTypeInfo() {
+ StringBuilder sb = new StringBuilder();
+ ResultCode[] types = ResultCode.values();
+ for (ResultCode type : types) {
+ sb.append(StrUtil.format("{}:{}, ", type.getCode(), type.getMsg()));
+ }
+ return sb.toString();
+ }
+}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/BaseEsEntity.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/BaseEsEntity.java
new file mode 100644
index 00000000..32206066
--- /dev/null
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/BaseEsEntity.java
@@ -0,0 +1,37 @@
+package io.github.dunwu.javadb.elasticsearch.entity;
+
+import lombok.Data;
+import lombok.ToString;
+
+import java.io.Serializable;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * ES 实体接口
+ *
+ * @author Zhang Peng
+ * @since 2023-06-28
+ */
+@Data
+@ToString
+public abstract class BaseEsEntity implements Serializable {
+
+ public static final String DOC_ID = "docId";
+
+ /**
+ * 获取版本
+ */
+ protected Long version;
+
+ protected Float hitScore;
+
+ public abstract String getDocId();
+
+ public static Map getPropertiesMap() {
+ Map map = new LinkedHashMap<>(1);
+ map.put(BaseEsEntity.DOC_ID, "keyword");
+ return map;
+ }
+
+}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/EsEntity.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/EsEntity.java
deleted file mode 100644
index df6d973f..00000000
--- a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/EsEntity.java
+++ /dev/null
@@ -1,16 +0,0 @@
-package io.github.dunwu.javadb.elasticsearch.entity;
-
-/**
- * ES 实体接口
- *
- * @author Zhang Peng
- * @since 2023-06-28
- */
-public interface EsEntity {
-
- /**
- * 获取 ES 主键
- */
- String getDocId();
-
-}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/Page.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/Page.java
deleted file mode 100644
index 76f178fa..00000000
--- a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/Page.java
+++ /dev/null
@@ -1,36 +0,0 @@
-package io.github.dunwu.javadb.elasticsearch.entity;
-
-import lombok.Data;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-
-/**
- * 分页实体
- *
- * @author Zhang Peng
- * @date 2023-06-28
- */
-@Data
-public class Page {
-
- private long total;
- private int page;
- private int size;
- private List content = new ArrayList<>();
-
- public Page(long total, int page, int size) {
- this.total = total;
- this.page = page;
- this.size = size;
- }
-
- public Page(long total, int page, int size, Collection list) {
- this.total = total;
- this.page = page;
- this.size = size;
- this.content.addAll(list);
- }
-
-}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/User.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/User.java
index 58428a14..b21b229c 100644
--- a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/User.java
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/User.java
@@ -1,27 +1,44 @@
package io.github.dunwu.javadb.elasticsearch.entity;
+import lombok.AllArgsConstructor;
import lombok.Builder;
-import lombok.Data;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import java.io.Serializable;
+import java.util.LinkedHashMap;
+import java.util.Map;
/**
- * 用户实体
+ * 短剧、长视频消费数据 ES 实体
*
* @author Zhang Peng
- * @since 2023-06-28
+ * @date 2024-04-02
*/
-@Data
@Builder
-public class User implements EsEntity {
+@Getter
+@Setter
+@NoArgsConstructor
+@AllArgsConstructor
+public class User extends BaseEsEntity implements Serializable {
- private Long id;
- private String username;
- private String password;
+ private String id;
+ private String name;
private Integer age;
- private String email;
@Override
public String getDocId() {
- return null;
+ return id;
+ }
+
+ public static Map getPropertiesMap() {
+ Map map = new LinkedHashMap<>();
+ map.put(BaseEsEntity.DOC_ID, "keyword");
+ map.put("id", "long");
+ map.put("name", "keyword");
+ map.put("age", "integer");
+ return map;
}
}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/common/PageData.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/common/PageData.java
new file mode 100644
index 00000000..e436f8b6
--- /dev/null
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/common/PageData.java
@@ -0,0 +1,37 @@
+package io.github.dunwu.javadb.elasticsearch.entity.common;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 分页实体
+ *
+ * @author Zhang Peng
+ * @date 2023-06-28
+ */
+@Getter
+@Setter
+@NoArgsConstructor
+@AllArgsConstructor
+public class PageData implements Serializable {
+
+ private int page;
+ private int size;
+ private long total;
+ private List content = new ArrayList<>();
+
+ public PageData(int page, int size, long total) {
+ this.total = total;
+ this.page = page;
+ this.size = size;
+ }
+
+ private static final long serialVersionUID = 1L;
+
+}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/common/ScrollData.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/common/ScrollData.java
new file mode 100644
index 00000000..4f90cb85
--- /dev/null
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/entity/common/ScrollData.java
@@ -0,0 +1,30 @@
+package io.github.dunwu.javadb.elasticsearch.entity.common;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import java.io.Serializable;
+import java.util.Collection;
+
+/**
+ * Hbase 滚动数据实体
+ *
+ * @author Zhang Peng
+ * @date 2023-11-16
+ */
+@Getter
+@Setter
+@NoArgsConstructor
+@AllArgsConstructor
+public class ScrollData implements Serializable {
+
+ private String scrollId;
+ private int size = 10;
+ private long total = 0L;
+ private Collection content;
+
+ private static final long serialVersionUID = 1L;
+
+}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/exception/CodeMsgException.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/exception/CodeMsgException.java
new file mode 100644
index 00000000..98ab1995
--- /dev/null
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/exception/CodeMsgException.java
@@ -0,0 +1,128 @@
+package io.github.dunwu.javadb.elasticsearch.exception;
+
+import cn.hutool.core.util.StrUtil;
+import io.github.dunwu.javadb.elasticsearch.constant.CodeMsg;
+import io.github.dunwu.javadb.elasticsearch.constant.ResultCode;
+
+/**
+ * 基础异常
+ *
+ * @author Zhang Peng
+ * @since 2021-09-25
+ */
+public class CodeMsgException extends RuntimeException implements CodeMsg {
+
+ private static final long serialVersionUID = 6146660782281445735L;
+
+ /**
+ * 状态码
+ */
+ protected int code;
+
+ /**
+ * 响应信息
+ */
+ protected String msg;
+
+ /**
+ * 提示信息
+ */
+ protected String toast;
+
+ public CodeMsgException() {
+ this(ResultCode.FAIL);
+ }
+
+ public CodeMsgException(CodeMsg codeMsg) {
+ this(codeMsg.getCode(), codeMsg.getMsg());
+ }
+
+ public CodeMsgException(CodeMsg codeMsg, String msg) {
+ this(codeMsg.getCode(), msg, null);
+ }
+
+ public CodeMsgException(CodeMsg codeMsg, String msg, String toast) {
+ this(codeMsg.getCode(), msg, toast);
+ }
+
+ public CodeMsgException(String msg) {
+ this(ResultCode.FAIL, msg);
+ }
+
+ public CodeMsgException(int code, String msg) {
+ this(code, msg, msg);
+ }
+
+ public CodeMsgException(int code, String msg, String toast) {
+ super(msg);
+ setCode(code);
+ setMsg(msg);
+ setToast(toast);
+ }
+
+ public CodeMsgException(Throwable cause) {
+ this(cause, ResultCode.FAIL);
+ }
+
+ public CodeMsgException(Throwable cause, String msg) {
+ this(cause, ResultCode.FAIL, msg);
+ }
+
+ public CodeMsgException(Throwable cause, CodeMsg codeMsg) {
+ this(cause, codeMsg.getCode(), codeMsg.getMsg());
+ }
+
+ public CodeMsgException(Throwable cause, CodeMsg codeMsg, String msg) {
+ this(cause, codeMsg.getCode(), msg, null);
+ }
+
+ public CodeMsgException(Throwable cause, CodeMsg codeMsg, String msg, String toast) {
+ this(cause, codeMsg.getCode(), msg, toast);
+ }
+
+ public CodeMsgException(Throwable cause, int code, String msg) {
+ this(cause, code, msg, null);
+ }
+
+ public CodeMsgException(Throwable cause, int code, String msg, String toast) {
+ super(msg, cause);
+ setCode(code);
+ setMsg(msg);
+ setToast(toast);
+ }
+
+ @Override
+ public String getMessage() {
+ if (StrUtil.isNotBlank(msg)) {
+ return StrUtil.format("[{}]{}", code, msg);
+ }
+ return super.getMessage();
+ }
+
+ @Override
+ public int getCode() {
+ return code;
+ }
+
+ public void setCode(int code) {
+ this.code = code;
+ }
+
+ @Override
+ public String getMsg() {
+ return msg;
+ }
+
+ public void setMsg(String msg) {
+ this.msg = msg;
+ }
+
+ public String getToast() {
+ return toast;
+ }
+
+ public void setToast(String toast) {
+ this.toast = toast;
+ }
+
+}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/exception/DefaultException.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/exception/DefaultException.java
new file mode 100644
index 00000000..14908e39
--- /dev/null
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/exception/DefaultException.java
@@ -0,0 +1,72 @@
+package io.github.dunwu.javadb.elasticsearch.exception;
+
+import io.github.dunwu.javadb.elasticsearch.constant.CodeMsg;
+import io.github.dunwu.javadb.elasticsearch.constant.ResultCode;
+
+/**
+ * 默认异常
+ *
+ * @author Zhang Peng
+ * @since 2021-12-30
+ */
+public class DefaultException extends CodeMsgException {
+
+ private static final long serialVersionUID = -7027578114976830416L;
+
+ public DefaultException() {
+ this(ResultCode.FAIL);
+ }
+
+ public DefaultException(CodeMsg codeMsg) {
+ this(codeMsg.getCode(), codeMsg.getMsg());
+ }
+
+ public DefaultException(CodeMsg codeMsg, String msg) {
+ this(codeMsg.getCode(), msg, null);
+ }
+
+ public DefaultException(CodeMsg codeMsg, String msg, String toast) {
+ this(codeMsg.getCode(), msg, toast);
+ }
+
+ public DefaultException(String msg) {
+ this(ResultCode.FAIL, msg);
+ }
+
+ public DefaultException(int code, String msg) {
+ this(code, msg, msg);
+ }
+
+ public DefaultException(int code, String msg, String toast) {
+ super(code, msg, toast);
+ }
+
+ public DefaultException(Throwable cause) {
+ this(cause, ResultCode.FAIL);
+ }
+
+ public DefaultException(Throwable cause, String msg) {
+ this(cause, ResultCode.FAIL, msg);
+ }
+
+ public DefaultException(Throwable cause, CodeMsg codeMsg) {
+ this(cause, codeMsg.getCode(), codeMsg.getMsg());
+ }
+
+ public DefaultException(Throwable cause, CodeMsg codeMsg, String msg) {
+ this(cause, codeMsg.getCode(), msg, null);
+ }
+
+ public DefaultException(Throwable cause, CodeMsg codeMsg, String msg, String toast) {
+ this(cause, codeMsg.getCode(), msg, toast);
+ }
+
+ public DefaultException(Throwable cause, int code, String msg) {
+ this(cause, code, msg, null);
+ }
+
+ public DefaultException(Throwable cause, int code, String msg, String toast) {
+ super(cause, code, msg, toast);
+ }
+
+}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/mapper/BaseDynamicEsMapper.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/mapper/BaseDynamicEsMapper.java
new file mode 100644
index 00000000..c75c6cc6
--- /dev/null
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/mapper/BaseDynamicEsMapper.java
@@ -0,0 +1,331 @@
+package io.github.dunwu.javadb.elasticsearch.mapper;
+
+import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.core.date.DatePattern;
+import cn.hutool.core.date.DateTime;
+import cn.hutool.core.date.DateUtil;
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.json.JSONUtil;
+import io.github.dunwu.javadb.elasticsearch.ElasticsearchTemplate;
+import io.github.dunwu.javadb.elasticsearch.constant.ResultCode;
+import io.github.dunwu.javadb.elasticsearch.entity.BaseEsEntity;
+import io.github.dunwu.javadb.elasticsearch.entity.common.PageData;
+import io.github.dunwu.javadb.elasticsearch.entity.common.ScrollData;
+import io.github.dunwu.javadb.elasticsearch.exception.DefaultException;
+import lombok.extern.slf4j.Slf4j;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.get.GetResponse;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * 动态 ES Mapper 基础类（以时间为维度动态创建、删除 index），用于数据量特别大，需要按照日期分片的索引。
+ *
+ * 注：使用此 Mapper 的索引、别名必须遵循命名格式：索引名 = 别名_yyyyMMdd
+ *
+ * @author Zhang Peng
+ * @date 2024-04-07
+ */
+@Slf4j
+public abstract class BaseDynamicEsMapper extends BaseEsMapper {
+
+ public BaseDynamicEsMapper(ElasticsearchTemplate elasticsearchTemplate) {
+ super(elasticsearchTemplate);
+ }
+
+ @Override
+ public boolean enableAutoCreateIndex() {
+ return true;
+ }
+
+ // ====================================================================
+ // 索引管理操作
+ // ====================================================================
+
+ public String getIndex(String day) {
+
+ String alias = getAlias();
+ if (StrUtil.isBlank(day)) {
+ String msg = StrUtil.format("【ES】获取 {} 索引失败！day 不能为空！", alias);
+ throw new DefaultException(ResultCode.PARAM_ERROR, msg);
+ }
+
+ DateTime date;
+ try {
+ date = DateUtil.parse(day, DatePattern.NORM_DATE_PATTERN);
+ } catch (Exception e) {
+ String msg = StrUtil.format("【ES】获取 {} 索引失败！day: {} 不符合日期格式 {}！",
+ alias, day, DatePattern.NORM_DATE_PATTERN);
+ throw new DefaultException(e, ResultCode.PARAM_ERROR, msg);
+ }
+
+ String formatDate = DateUtil.format(date, DatePattern.PURE_DATE_FORMAT);
+ return alias + "_" + formatDate;
+ }
+
+ public boolean isIndexExistsInDay(String day) {
+ if (StrUtil.isBlank(day)) {
+ return false;
+ }
+ String index = getIndex(day);
+ try {
+ return elasticsearchTemplate.isIndexExists(getIndex(day));
+ } catch (Exception e) {
+ log.error("【ES】判断索引是否存在异常！index: {}", index, e);
+ return false;
+ }
+ }
+
+ public String createIndexIfNotExistsInDay(String day) {
+ String index = getIndex(day);
+ String type = getType();
+ String alias = getAlias();
+ int shard = getShard();
+ int replica = getReplica();
+ return createIndex(index, type, alias, shard, replica);
+ }
+
+ public void deleteIndexInDay(String day) {
+ String index = getIndex(day);
+ try {
+ log.info("【ES】删除索引成功！index: {}", index);
+ elasticsearchTemplate.deleteIndex(index);
+ } catch (Exception e) {
+ log.error("【ES】删除索引异常！index: {}", index, e);
+ }
+ }
+
+ public void updateAliasInDay(String day) {
+ String index = getIndex(day);
+ String alias = getAlias();
+ try {
+ log.info("【ES】更新别名成功！alias: {} -> index: {}", alias, index);
+ elasticsearchTemplate.updateAlias(index, alias);
+ } catch (IOException e) {
+ log.error("【ES】更新别名异常！alias: {} -> index: {}", alias, index, e);
+ }
+ }
+
+ // ====================================================================
+ // CRUD 操作
+ // ====================================================================
+
+ public GetResponse getByIdInDay(String day, String id) {
+ String index = getIndex(day);
+ String type = getType();
+ try {
+ return elasticsearchTemplate.getById(index, type, id, null);
+ } catch (IOException e) {
+ log.error("【ES】根据ID查询异常！index: {}, type: {}, id: {}", index, type, id, e);
+ return null;
+ }
+ }
+
+ public T pojoByIdInDay(String day, String id) {
+ String index = getIndex(day);
+ String type = getType();
+ try {
+ return elasticsearchTemplate.pojoById(index, type, id, null, getEntityClass());
+ } catch (IOException e) {
+ log.error("【ES】根据ID查询POJO异常！index: {}, type: {}, id: {}", index, type, id, e);
+ return null;
+ }
+ }
+
+ public List pojoListByIdsInDay(String day, Collection ids) {
+ String index = getIndex(day);
+ String type = getType();
+ try {
+ return elasticsearchTemplate.pojoListByIds(index, type, ids, getEntityClass());
+ } catch (IOException e) {
+ log.error("【ES】根据ID查询POJO列表异常！index: {}, type: {}, ids: {}", index, type, ids, e);
+ return new ArrayList<>(0);
+ }
+ }
+
+ public long countInDay(String day, SearchSourceBuilder builder) {
+ String index = getIndex(day);
+ String type = getType();
+ try {
+ return elasticsearchTemplate.count(index, type, builder);
+ } catch (IOException e) {
+ log.error("【ES】获取匹配记录数异常！index: {}, type: {}", index, type, e);
+ return 0L;
+ }
+ }
+
+ public SearchResponse queryInDay(String day, SearchSourceBuilder builder) {
+ String index = getIndex(day);
+ String type = getType();
+ try {
+ return elasticsearchTemplate.query(index, type, builder);
+ } catch (IOException e) {
+ log.error("【ES】条件查询异常！index: {}, type: {}", index, type, e);
+ return null;
+ }
+ }
+
+ public PageData pojoPageInDay(String day, SearchSourceBuilder builder) {
+ String index = getIndex(day);
+ String type = getType();
+ try {
+ return elasticsearchTemplate.pojoPage(index, type, builder, getEntityClass());
+ } catch (IOException e) {
+ log.error("【ES】from + size 分页条件查询异常！index: {}, type: {}", index, type, e);
+ return null;
+ }
+ }
+
+ public ScrollData pojoPageByLastIdInDay(String day, String scrollId, int size, QueryBuilder queryBuilder) {
+ String index = getIndex(day);
+ String type = getType();
+ try {
+ return elasticsearchTemplate.pojoPageByScrollId(index, type, scrollId, size, queryBuilder, getEntityClass());
+ } catch (IOException e) {
+ log.error("【ES】search after 分页条件查询异常！index: {}, type: {}", index, type, e);
+ return null;
+ }
+ }
+
+ public ScrollData pojoScrollBeginInDay(String day, SearchSourceBuilder builder) {
+ String index = getIndex(day);
+ String type = getType();
+ try {
+ return elasticsearchTemplate.pojoScrollBegin(index, type, builder, getEntityClass());
+ } catch (IOException e) {
+ log.error("【ES】开启滚动分页条件查询异常！index: {}, type: {}", index, type, e);
+ return null;
+ }
+ }
+
+ /**
+ * 根据日期动态选择索引并更新
+ *
+ * @param day 日期，格式为：yyyy-MM-dd
+ * @param entity 待更新的数据
+ * @return /
+ */
+ public T saveInDay(String day, T entity) {
+ if (StrUtil.isBlank(day) || entity == null) {
+ return null;
+ }
+ String index = getIndex(day);
+ String type = getType();
+ try {
+ checkIndex(day);
+ checkData(entity);
+ return elasticsearchTemplate.save(index, getType(), entity);
+ } catch (IOException e) {
+ log.error("【ES】添加数据异常！index: {}, type: {}, entity: {}", index, type, JSONUtil.toJsonStr(entity), e);
+ return null;
+ }
+ }
+
+ /**
+ * 根据日期动态选择索引并批量更新
+ *
+ * @param day 日期，格式为：yyyy-MM-dd
+ * @param list 待更新的数据
+ * @return /
+ */
+ public boolean saveBatchInDay(String day, Collection list) {
+ if (StrUtil.isBlank(day) || CollectionUtil.isEmpty(list)) {
+ return false;
+ }
+ String index = getIndex(day);
+ String type = getType();
+ try {
+ checkIndex(day);
+ checkData(list);
+ return elasticsearchTemplate.saveBatch(index, type, list);
+ } catch (IOException e) {
+ log.error("【ES】批量添加数据异常！index: {}, type: {}, size: {}", index, type, list.size(), e);
+ return false;
+ }
+ }
+
+ public void asyncSaveBatchInDay(String day, Collection list) {
+ asyncSaveBatchInDay(day, list, DEFAULT_BULK_LISTENER);
+ }
+
+ public void asyncSaveBatchInDay(String day, Collection list, ActionListener listener) {
+ if (StrUtil.isBlank(day) || CollectionUtil.isEmpty(list)) {
+ return;
+ }
+ String index = getIndex(day);
+ String type = getType();
+ try {
+ checkIndex(day);
+ checkData(list);
+ elasticsearchTemplate.asyncSaveBatch(index, type, list, listener);
+ } catch (Exception e) {
+ log.error("【ES】异步批量添加数据异常！index: {}, type: {}, size: {}", index, type, list.size(), e);
+ }
+ }
+
+ public void asyncUpdateBatchIdsInDay(String day, Collection list) {
+ asyncUpdateBatchIdsInDay(day, list, DEFAULT_BULK_LISTENER);
+ }
+
+ public void asyncUpdateBatchIdsInDay(String day, Collection list, ActionListener listener) {
+ if (StrUtil.isBlank(day) || CollectionUtil.isEmpty(list)) {
+ return;
+ }
+ String index = getIndex(day);
+ String type = getType();
+ try {
+ checkData(list);
+ elasticsearchTemplate.asyncUpdateBatchIds(index, type, list, listener);
+ } catch (Exception e) {
+ log.error("【ES】异步批量更新数据异常！index: {}, type: {}, size: {}", index, type, list.size(), e);
+ }
+ }
+
+ public boolean deleteByIdInDay(String day, String id) {
+ if (StrUtil.isBlank(day) || StrUtil.isBlank(id)) {
+ return false;
+ }
+ String index = getIndex(day);
+ String type = getType();
+ try {
+ return elasticsearchTemplate.deleteById(index, type, id);
+ } catch (IOException e) {
+ log.error("【ES】根据ID删除数据异常！index: {}, type: {}, id: {}", index, type, id, e);
+ return false;
+ }
+ }
+
+ public boolean deleteBatchIdsInDay(String day, Collection ids) {
+ if (StrUtil.isBlank(day) || CollectionUtil.isEmpty(ids)) {
+ return false;
+ }
+ String index = getIndex(day);
+ String type = getType();
+ try {
+ return elasticsearchTemplate.deleteBatchIds(index, type, ids);
+ } catch (IOException e) {
+ log.error("【ES】根据ID批量删除数据异常！index: {}, type: {}, ids: {}", index, type, ids, e);
+ return false;
+ }
+ }
+
+ protected String checkIndex(String day) {
+ if (!enableAutoCreateIndex()) {
+ return getIndex(day);
+ }
+ String index = createIndexIfNotExistsInDay(day);
+ if (StrUtil.isBlank(index)) {
+ String msg = StrUtil.format("【ES】索引 {}_{} 找不到且创建失败！", getAlias(), day);
+ throw new DefaultException(ResultCode.ERROR, msg);
+ }
+ return index;
+ }
+
+}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/mapper/BaseEsMapper.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/mapper/BaseEsMapper.java
index 4033b4e8..b125bea9 100644
--- a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/mapper/BaseEsMapper.java
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/mapper/BaseEsMapper.java
@@ -1,23 +1,34 @@
package io.github.dunwu.javadb.elasticsearch.mapper;
-import cn.hutool.core.lang.Assert;
-import io.github.dunwu.javadb.elasticsearch.entity.EsEntity;
-import io.github.dunwu.javadb.elasticsearch.entity.Page;
-import io.github.dunwu.javadb.elasticsearch.util.ElasticsearchUtil;
+import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.core.map.MapUtil;
+import cn.hutool.core.util.ReflectUtil;
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.json.JSONUtil;
+import io.github.dunwu.javadb.elasticsearch.ElasticsearchTemplate;
+import io.github.dunwu.javadb.elasticsearch.constant.ResultCode;
+import io.github.dunwu.javadb.elasticsearch.entity.BaseEsEntity;
+import io.github.dunwu.javadb.elasticsearch.entity.common.PageData;
+import io.github.dunwu.javadb.elasticsearch.entity.common.ScrollData;
+import io.github.dunwu.javadb.elasticsearch.exception.DefaultException;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.client.IndicesClient;
-import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
-import java.io.IOException;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
+import java.util.Map;
+import java.util.Set;
/**
* ES Mapper 基础类
@@ -26,161 +37,466 @@
* @date 2023-06-27
*/
@Slf4j
-public abstract class BaseEsMapper implements EsMapper {
+public abstract class BaseEsMapper implements EsMapper {
- private BulkProcessor bulkProcessor;
+ protected BulkProcessor bulkProcessor;
- protected final RestHighLevelClient restHighLevelClient;
+ protected final ElasticsearchTemplate elasticsearchTemplate;
- public BaseEsMapper(RestHighLevelClient restHighLevelClient) {
- this.restHighLevelClient = restHighLevelClient;
+ public BaseEsMapper(ElasticsearchTemplate elasticsearchTemplate) {
+ this.elasticsearchTemplate = elasticsearchTemplate;
+ }
+
+ public int getShard() {
+ return 5;
+ }
+
+ public int getReplica() {
+ return 1;
}
@Override
- public RestHighLevelClient getClient() throws IOException {
- Assert.notNull(restHighLevelClient, () -> new IOException("【ES】not connected."));
- return restHighLevelClient;
+ public RestHighLevelClient getClient() {
+ if (elasticsearchTemplate == null) {
+ return null;
+ }
+ return elasticsearchTemplate.getClient();
}
@Override
- public synchronized BulkProcessor getBulkProcessor() throws IOException {
+ public synchronized BulkProcessor getBulkProcessor() {
if (bulkProcessor == null) {
- bulkProcessor = ElasticsearchUtil.newAsyncBulkProcessor(getClient());
+ bulkProcessor = elasticsearchTemplate.newAsyncBulkProcessor();
}
return bulkProcessor;
}
+ @SuppressWarnings("unchecked")
+ public Map getPropertiesMap() {
+ Class clazz = getEntityClass();
+ Method method;
+ try {
+ method = clazz.getMethod("getPropertiesMap");
+ } catch (NoSuchMethodException e) {
+ log.error("【ES】{} 中不存在 getPropertiesMap 方法！", clazz.getCanonicalName());
+ return new HashMap<>(0);
+ }
+
+ Object result = ReflectUtil.invokeStatic(method);
+ if (result == null) {
+ return new HashMap<>(0);
+ }
+ return (Map) result;
+ }
+
+ // ====================================================================
+ // 索引管理操作
+ // ====================================================================
+
@Override
- public boolean isIndexExists() throws IOException {
- IndicesClient indicesClient = getClient().indices();
- GetIndexRequest request = new GetIndexRequest();
- request.indices(getIndex());
- return indicesClient.exists(request, RequestOptions.DEFAULT);
+ public boolean isIndexExists() {
+ String index = getIndex();
+ try {
+ return elasticsearchTemplate.isIndexExists(index);
+ } catch (Exception e) {
+ log.error("【ES】判断索引是否存在异常！index: {}", index, e);
+ return false;
+ }
}
@Override
- public SearchResponse getById(String id) throws IOException {
- return ElasticsearchUtil.getById(getClient(), getIndex(), getType(), id);
+ public String createIndexIfNotExists() {
+ String index = getIndex();
+ String type = getType();
+ String alias = getAlias();
+ int shard = getShard();
+ int replica = getReplica();
+ return createIndex(index, type, alias, shard, replica);
+ }
+
+ protected String createIndex(String index, String type, String alias, int shard, int replica) {
+ try {
+ if (elasticsearchTemplate.isIndexExists(index)) {
+ return index;
+ }
+ elasticsearchTemplate.createIndex(index, type, alias, shard, replica);
+ log.info("【ES】创建索引成功！index: {}, type: {}, alias: {}, shard: {}, replica: {}",
+ index, type, alias, shard, replica);
+ Map propertiesMap = getPropertiesMap();
+ if (MapUtil.isNotEmpty(propertiesMap)) {
+ elasticsearchTemplate.setMapping(index, type, propertiesMap);
+ log.error("【ES】设置索引 mapping 成功！index: {}, type: {}, propertiesMap: {}",
+ index, type, JSONUtil.toJsonStr(propertiesMap));
+ }
+ return index;
+ } catch (Exception e) {
+ log.error("【ES】创建索引异常！index: {}, type: {}, alias: {}, shard: {}, replica: {}",
+ index, type, alias, shard, replica, e);
+ return null;
+ }
}
@Override
- public T pojoById(String id) throws IOException {
- return ElasticsearchUtil.pojoById(getClient(), getIndex(), getType(), id, getEntityClass());
+ public void deleteIndex() {
+ String index = getIndex();
+ try {
+ log.info("【ES】删除索引成功！index: {}", index);
+ elasticsearchTemplate.deleteIndex(index);
+ } catch (Exception e) {
+ log.error("【ES】删除索引异常！index: {}", index, e);
+ }
}
@Override
- public List pojoListByIds(Collection ids) throws IOException {
- return ElasticsearchUtil.pojoListByIds(getClient(), getIndex(), getType(), ids, getEntityClass());
+ public void updateAlias() {
+ String index = getIndex();
+ String alias = getAlias();
+ try {
+ log.info("【ES】更新别名成功！alias: {} -> index: {}", alias, index);
+ elasticsearchTemplate.updateAlias(index, alias);
+ } catch (Exception e) {
+ log.error("【ES】更新别名异常！alias: {} -> index: {}", alias, index, e);
+ }
}
@Override
- public Page pojoPage(SearchSourceBuilder builder) throws IOException {
- return ElasticsearchUtil.pojoPage(getClient(), getIndex(), getType(), builder, getEntityClass());
+ public Set getIndexSet() {
+ String alias = getAlias();
+ try {
+ return elasticsearchTemplate.getIndexSet(alias);
+ } catch (Exception e) {
+ log.error("【ES】获取别名的所有索引异常！alias: {}", alias, e);
+ return new HashSet<>(0);
+ }
}
+ // ====================================================================
+ // CRUD 操作
+ // ====================================================================
+
@Override
- public String insert(T entity) throws IOException {
- return ElasticsearchUtil.insert(getClient(), getIndex(), getType(), entity);
+ public GetResponse getById(String id) {
+ return getById(id, null);
}
@Override
- public boolean batchInsert(Collection list) throws IOException {
- return ElasticsearchUtil.batchInsert(getClient(), getIndex(), getType(), list);
+ public GetResponse getById(String id, Long version) {
+ String index = getIndex();
+ String type = getType();
+ try {
+ return elasticsearchTemplate.getById(index, type, id, version);
+ } catch (Exception e) {
+ log.error("【ES】根据ID查询异常！index: {}, type: {}, id: {}, version: {}", index, type, id, version, e);
+ return null;
+ }
}
@Override
- public void asyncBatchInsert(Collection list) throws IOException {
- ActionListener listener = new ActionListener() {
- @Override
- public void onResponse(BulkResponse response) {
- if (response != null && !response.hasFailures()) {
- log.info("【ES】异步批量插入成功！");
- } else {
- log.warn("【ES】异步批量插入失败！");
- }
- }
+ public T pojoById(String id) {
+ return pojoById(id, null);
+ }
- @Override
- public void onFailure(Exception e) {
- log.error("【ES】异步批量插入异常！", e);
- }
- };
- asyncBatchInsert(list, listener);
+ @Override
+ public T pojoById(String id, Long version) {
+ String index = getIndex();
+ String type = getType();
+ try {
+ return elasticsearchTemplate.pojoById(index, type, id, version, getEntityClass());
+ } catch (Exception e) {
+ log.error("【ES】根据ID查询POJO异常！index: {}, type: {}, id: {}, version: {}", index, type, id, version, e);
+ return null;
+ }
}
@Override
- public void asyncBatchInsert(Collection list, ActionListener listener) throws IOException {
- ElasticsearchUtil.asyncBatchInsert(getClient(), getIndex(), getType(), list, listener);
+ public List pojoListByIds(Collection ids) {
+ String index = getIndex();
+ String type = getType();
+ try {
+ return elasticsearchTemplate.pojoListByIds(index, type, ids, getEntityClass());
+ } catch (Exception e) {
+ log.error("【ES】根据ID查询POJO列表异常！index: {}, type: {}, ids: {}", index, type, ids, e);
+ return new ArrayList<>(0);
+ }
}
@Override
- public boolean updateById(T entity) throws IOException {
- return ElasticsearchUtil.updateById(getClient(), getIndex(), getType(), entity);
+ public long count(SearchSourceBuilder builder) {
+ String index = getIndex();
+ String type = getType();
+ try {
+ return elasticsearchTemplate.count(index, type, builder);
+ } catch (Exception e) {
+ log.error("【ES】获取匹配记录数异常！index: {}, type: {}", index, type, e);
+ return 0L;
+ }
}
@Override
- public boolean batchUpdateById(Collection list) throws IOException {
- return ElasticsearchUtil.batchUpdateById(getClient(), getIndex(), getType(), list);
+ public SearchResponse query(SearchSourceBuilder builder) {
+ String index = getIndex();
+ String type = getType();
+ try {
+ return elasticsearchTemplate.query(index, type, builder);
+ } catch (Exception e) {
+ log.error("【ES】条件查询异常！index: {}, type: {}", index, type, e);
+ return null;
+ }
}
@Override
- public void asyncBatchUpdateById(Collection list) throws IOException {
- ActionListener listener = new ActionListener() {
- @Override
- public void onResponse(BulkResponse response) {
- if (response != null && !response.hasFailures()) {
- log.info("【ES】异步批量更新成功！");
- } else {
- log.warn("【ES】异步批量更新失败！");
- }
- }
+ public PageData pojoPage(SearchSourceBuilder builder) {
+ String index = getIndex();
+ String type = getType();
+ try {
+ return elasticsearchTemplate.pojoPage(index, type, builder, getEntityClass());
+ } catch (Exception e) {
+ log.error("【ES】from + size 分页条件查询异常！index: {}, type: {}", index, type, e);
+ return null;
+ }
+ }
- @Override
- public void onFailure(Exception e) {
- log.error("【ES】异步批量更新异常！", e);
- }
- };
- asyncBatchUpdateById(list, listener);
+ @Override
+ public ScrollData pojoPageByLastId(String scrollId, int size, QueryBuilder queryBuilder) {
+ String index = getIndex();
+ String type = getType();
+ try {
+ return elasticsearchTemplate.pojoPageByScrollId(index, type, scrollId, size, queryBuilder,
+ getEntityClass());
+ } catch (Exception e) {
+ log.error("【ES】search after 分页条件查询异常！index: {}, type: {}", index, type, e);
+ return null;
+ }
+ }
+
+ @Override
+ public ScrollData pojoScrollBegin(SearchSourceBuilder builder) {
+ String index = getIndex();
+ String type = getType();
+ try {
+ return elasticsearchTemplate.pojoScrollBegin(index, type, builder, getEntityClass());
+ } catch (Exception e) {
+ log.error("【ES】开启滚动分页条件查询异常！index: {}, type: {}", index, type, e);
+ return null;
+ }
}
@Override
- public void asyncBatchUpdateById(Collection list, ActionListener listener) throws IOException {
- ElasticsearchUtil.asyncBatchUpdateById(getClient(), getIndex(), getType(), list, listener);
+ public ScrollData pojoScroll(String scrollId, SearchSourceBuilder builder) {
+ try {
+ return elasticsearchTemplate.pojoScroll(scrollId, builder, getEntityClass());
+ } catch (Exception e) {
+ log.error("【ES】滚动分页条件查询异常！scrollId: {}", scrollId, e);
+ return null;
+ }
}
@Override
- public boolean deleteById(String id) throws IOException {
- return ElasticsearchUtil.deleteById(getClient(), getIndex(), getType(), id);
+ public boolean pojoScrollEnd(String scrollId) {
+ try {
+ return elasticsearchTemplate.pojoScrollEnd(scrollId);
+ } catch (Exception e) {
+ log.error("【ES】关闭滚动分页条件查询异常！scrollId: {}", scrollId, e);
+ return false;
+ }
}
@Override
- public boolean batchDeleteById(Collection ids) throws IOException {
- return ElasticsearchUtil.batchDeleteById(getClient(), getIndex(), getType(), ids);
+ public T save(T entity) {
+ if (entity == null) {
+ return null;
+ }
+ String index = getIndex();
+ String type = getType();
+ try {
+ checkIndex();
+ checkData(entity);
+ return elasticsearchTemplate.save(index, type, entity);
+ } catch (Exception e) {
+ log.error("【ES】添加数据异常！index: {}, type: {}, entity: {}", index, type, JSONUtil.toJsonStr(entity), e);
+ return null;
+ }
}
@Override
- public void asyncBatchDeleteById(Collection ids) throws IOException {
- ActionListener listener = new ActionListener() {
- @Override
- public void onResponse(BulkResponse response) {
- if (response != null && !response.hasFailures()) {
- log.info("【ES】异步批量删除成功！");
- } else {
- log.warn("【ES】异步批量删除失败！ids: {}", ids);
- }
- }
+ public boolean saveBatch(Collection list) {
+ if (CollectionUtil.isEmpty(list)) {
+ return false;
+ }
+ String index = getIndex();
+ String type = getType();
+ try {
+ checkIndex();
+ checkData(list);
+ return elasticsearchTemplate.saveBatch(index, type, list);
+ } catch (Exception e) {
+ log.error("【ES】批量添加数据异常！index: {}, type: {}, size: {}", index, type, list.size(), e);
+ return false;
+ }
+ }
- @Override
- public void onFailure(Exception e) {
- log.error("【ES】异步批量删除异常！ids: {}", ids, e);
- }
- };
- asyncBatchDeleteById(ids, listener);
+ @Override
+ public void asyncSaveBatch(Collection list) {
+ asyncSaveBatch(list, DEFAULT_BULK_LISTENER);
}
@Override
- public void asyncBatchDeleteById(Collection ids, ActionListener listener) throws IOException {
- ElasticsearchUtil.asyncBatchDeleteById(getClient(), getIndex(), getType(), ids, listener);
+ public void asyncSaveBatch(Collection list, ActionListener listener) {
+ if (CollectionUtil.isEmpty(list)) {
+ return;
+ }
+ String index = getIndex();
+ String type = getType();
+ try {
+ checkIndex();
+ checkData(list);
+ elasticsearchTemplate.asyncSaveBatch(index, getType(), list, listener);
+ } catch (Exception e) {
+ log.error("【ES】异步批量添加数据异常！index: {}, type: {}, size: {}", index, type, list.size(), e);
+ }
}
+ @Override
+ public T updateById(T entity) {
+ if (entity == null) {
+ return null;
+ }
+ String index = getIndex();
+ String type = getType();
+ try {
+ checkData(entity);
+ return elasticsearchTemplate.updateById(index, type, entity);
+ } catch (Exception e) {
+ log.error("【ES】更新数据异常！index: {}, type: {}", index, type, e);
+ return null;
+ }
+ }
+
+ @Override
+ public boolean updateBatchIds(Collection list) {
+ if (CollectionUtil.isEmpty(list)) {
+ return false;
+ }
+ String index = getIndex();
+ String type = getType();
+ try {
+ checkData(list);
+ return elasticsearchTemplate.updateBatchIds(index, type, list);
+ } catch (Exception e) {
+ log.error("【ES】批量更新数据异常！index: {}, type: {}, size: {}", index, type, list.size(), e);
+ return false;
+ }
+ }
+
+ @Override
+ public void asyncUpdateBatchIds(Collection list) {
+ asyncUpdateBatchIds(list, DEFAULT_BULK_LISTENER);
+ }
+
+ @Override
+ public void asyncUpdateBatchIds(Collection list, ActionListener listener) {
+ if (CollectionUtil.isEmpty(list)) {
+ return;
+ }
+ String index = getIndex();
+ String type = getType();
+ try {
+ checkData(list);
+ elasticsearchTemplate.asyncUpdateBatchIds(index, type, list, listener);
+ } catch (Exception e) {
+ log.error("【ES】异步批量更新数据异常！index: {}, type: {}, size: {}", index, type, list.size(), e);
+ }
+ }
+
+ @Override
+ public boolean deleteById(String id) {
+ if (StrUtil.isBlank(id)) {
+ return false;
+ }
+ String index = getIndex();
+ String type = getType();
+ try {
+ return elasticsearchTemplate.deleteById(index, type, id);
+ } catch (Exception e) {
+ log.error("【ES】根据ID删除数据异常！index: {}, type: {}, id: {}", index, type, id, e);
+ return false;
+ }
+ }
+
+ @Override
+ public boolean deleteBatchIds(Collection ids) {
+ if (CollectionUtil.isEmpty(ids)) {
+ return false;
+ }
+ String index = getIndex();
+ String type = getType();
+ try {
+ return elasticsearchTemplate.deleteBatchIds(index, type, ids);
+ } catch (Exception e) {
+ log.error("【ES】根据ID批量删除数据异常！index: {}, type: {}, ids: {}", index, type, ids, e);
+ return false;
+ }
+ }
+
+ @Override
+ public void asyncDeleteBatchIds(Collection ids) {
+ asyncDeleteBatchIds(ids, DEFAULT_BULK_LISTENER);
+ }
+
+ @Override
+ public void asyncDeleteBatchIds(Collection ids, ActionListener listener) {
+ if (CollectionUtil.isEmpty(ids)) {
+ return;
+ }
+ String index = getIndex();
+ String type = getType();
+ try {
+ elasticsearchTemplate.asyncDeleteBatchIds(index, type, ids, listener);
+ } catch (Exception e) {
+ log.error("【ES】异步根据ID批量删除数据异常！index: {}, type: {}, ids: {}", index, type, ids, e);
+ }
+ }
+
+ protected String checkIndex() {
+ if (!enableAutoCreateIndex()) {
+ return getIndex();
+ }
+ String index = createIndexIfNotExists();
+ if (StrUtil.isBlank(index)) {
+ String msg = StrUtil.format("【ES】索引 {} 找不到且创建失败！", index);
+ throw new DefaultException(ResultCode.ERROR, msg);
+ }
+ return index;
+ }
+
+ protected void checkData(Collection list) {
+ if (CollectionUtil.isEmpty(list)) {
+ String msg = StrUtil.format("【ES】写入 {} 失败！list 不能为空！", getIndex());
+ throw new DefaultException(ResultCode.PARAM_ERROR, msg);
+ }
+ }
+
+ protected void checkData(T entity) {
+ if (entity == null) {
+ String msg = StrUtil.format("【ES】写入 {} 失败！entity 不能为空！", getIndex());
+ throw new DefaultException(ResultCode.PARAM_ERROR, msg);
+ }
+ }
+
+ protected final ActionListener DEFAULT_BULK_LISTENER = new ActionListener() {
+ @Override
+ public void onResponse(BulkResponse response) {
+ if (response != null && !response.hasFailures()) {
+ log.info("【ES】异步批量写数据成功！index: {}, type: {}", getIndex(), getType());
+ } else {
+ log.warn("【ES】异步批量写数据失败！index: {}, type: {}", getIndex(), getType());
+ }
+ }
+
+ @Override
+ public void onFailure(Exception e) {
+ log.error("【ES】异步批量写数据异常！index: {}, type: {}", getIndex(), getType());
+ }
+ };
+
}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/mapper/EsMapper.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/mapper/EsMapper.java
index fa01d7c4..5630d664 100644
--- a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/mapper/EsMapper.java
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/mapper/EsMapper.java
@@ -1,17 +1,23 @@
package io.github.dunwu.javadb.elasticsearch.mapper;
-import io.github.dunwu.javadb.elasticsearch.entity.EsEntity;
-import io.github.dunwu.javadb.elasticsearch.entity.Page;
+import cn.hutool.core.collection.CollectionUtil;
+import io.github.dunwu.javadb.elasticsearch.entity.BaseEsEntity;
+import io.github.dunwu.javadb.elasticsearch.entity.common.PageData;
+import io.github.dunwu.javadb.elasticsearch.entity.common.ScrollData;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
-import java.io.IOException;
import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Set;
/**
* ES Mapper
@@ -19,7 +25,12 @@
* @author Zhang Peng
* @date 2023-06-27
*/
-public interface EsMapper {
+public interface EsMapper {
+
+ /**
+ * 获取别名
+ */
+ String getAlias();
/**
* 获取索引名
@@ -31,47 +42,101 @@ public interface EsMapper {
*/
String getType();
+ /**
+ * 获取分片数
+ */
+ int getShard();
+
+ /**
+ * 获取副本数
+ */
+ int getReplica();
+
/**
* 获取实体类型
*/
Class getEntityClass();
- RestHighLevelClient getClient() throws IOException;
+ /**
+ * 如果开启，添加 ES 数据时，如果索引不存在，会自动创建索引
+ */
+ default boolean enableAutoCreateIndex() {
+ return false;
+ }
+
+ RestHighLevelClient getClient();
+
+ BulkProcessor getBulkProcessor();
+
+ boolean isIndexExists();
+
+ String createIndexIfNotExists();
+
+ void deleteIndex();
+
+ void updateAlias();
+
+ Set getIndexSet();
+
+ GetResponse getById(String id);
+
+ GetResponse getById(String id, Long version);
+
+ T pojoById(String id);
+
+ T pojoById(String id, Long version);
+
+ List pojoListByIds(Collection ids);
+
+ default Map pojoMapByIds(Collection ids) {
+ List list = pojoListByIds(ids);
+ if (CollectionUtil.isEmpty(list)) {
+ return new HashMap<>(0);
+ }
+
+ Map map = new HashMap<>(list.size());
+ for (T entity : list) {
+ map.put(entity.getDocId(), entity);
+ }
+ return map;
+ }
+
+ long count(SearchSourceBuilder builder);
- BulkProcessor getBulkProcessor() throws IOException;
+ SearchResponse query(SearchSourceBuilder builder);
- boolean isIndexExists() throws IOException;
+ PageData pojoPage(SearchSourceBuilder builder);
- SearchResponse getById(String id) throws IOException;
+ ScrollData pojoPageByLastId(String scrollId, int size, QueryBuilder queryBuilder);
- T pojoById(String id) throws IOException;
+ ScrollData pojoScrollBegin(SearchSourceBuilder builder);
- List pojoListByIds(Collection ids) throws IOException;
+ ScrollData pojoScroll(String scrollId, SearchSourceBuilder builder);
- Page pojoPage(SearchSourceBuilder builder) throws IOException;
+ boolean pojoScrollEnd(String scrollId);
- String insert(T entity) throws IOException;
+ T save(T entity);
- boolean batchInsert(Collection list) throws IOException;
+ boolean saveBatch(Collection list);
- void asyncBatchInsert(Collection list) throws IOException;
+ void asyncSaveBatch(Collection list);
- void asyncBatchInsert(Collection list, ActionListener listener) throws IOException;
+ void asyncSaveBatch(Collection list, ActionListener listener);
- boolean updateById(T entity) throws IOException;
+ T updateById(T entity);
- boolean batchUpdateById(Collection list) throws IOException;
+ boolean updateBatchIds(Collection list);
- void asyncBatchUpdateById(Collection list) throws IOException;
+ void asyncUpdateBatchIds(Collection list);
- void asyncBatchUpdateById(Collection list, ActionListener listener) throws IOException;
+ void asyncUpdateBatchIds(Collection list, ActionListener listener);
- boolean deleteById(String id) throws IOException;
+ boolean deleteById(String id);
- boolean batchDeleteById(Collection ids) throws IOException;
+ boolean deleteBatchIds(Collection ids);
- void asyncBatchDeleteById(Collection ids) throws IOException;
+ void asyncDeleteBatchIds(Collection ids);
- void asyncBatchDeleteById(Collection ids, ActionListener listener) throws IOException;
+ void asyncDeleteBatchIds(Collection ids, ActionListener listener);
}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/mapper/UserEsMapper.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/mapper/UserEsMapper.java
index c60af8aa..de6d1b7c 100644
--- a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/mapper/UserEsMapper.java
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/mapper/UserEsMapper.java
@@ -1,25 +1,39 @@
package io.github.dunwu.javadb.elasticsearch.mapper;
+import cn.hutool.core.date.DatePattern;
+import cn.hutool.core.date.DateUtil;
+import io.github.dunwu.javadb.elasticsearch.ElasticsearchTemplate;
import io.github.dunwu.javadb.elasticsearch.entity.User;
-import org.elasticsearch.client.RestHighLevelClient;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.Date;
/**
- * User ES Mapper
+ * open_applet_consume_yyyyMMdd ES Mapper
*
* @author Zhang Peng
* @date 2023-06-27
*/
-public class UserEsMapper extends BaseEsMapper {
+@Slf4j
+@Component
+public class UserEsMapper extends BaseDynamicEsMapper {
- public UserEsMapper(RestHighLevelClient restHighLevelClient) {
- super(restHighLevelClient);
+ public UserEsMapper(ElasticsearchTemplate elasticsearchTemplate) {
+ super(elasticsearchTemplate);
}
@Override
- public String getIndex() {
+ public String getAlias() {
return "user";
}
+ @Override
+ public String getIndex() {
+ String date = DateUtil.format(new Date(), DatePattern.PURE_DATE_FORMAT);
+ return getAlias() + "_" + date;
+ }
+
@Override
public String getType() {
return "_doc";
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/util/ElasticsearchUtil.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/util/ElasticsearchUtil.java
deleted file mode 100644
index e3352082..00000000
--- a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/util/ElasticsearchUtil.java
+++ /dev/null
@@ -1,419 +0,0 @@
-package io.github.dunwu.javadb.elasticsearch.util;
-
-import cn.hutool.core.collection.CollectionUtil;
-import cn.hutool.core.util.ArrayUtil;
-import cn.hutool.core.util.StrUtil;
-import cn.hutool.json.JSONUtil;
-import io.github.dunwu.javadb.elasticsearch.entity.EsEntity;
-import io.github.dunwu.javadb.elasticsearch.entity.Page;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.http.HttpHost;
-import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.DocWriteResponse;
-import org.elasticsearch.action.bulk.BackoffPolicy;
-import org.elasticsearch.action.bulk.BulkProcessor;
-import org.elasticsearch.action.bulk.BulkRequest;
-import org.elasticsearch.action.bulk.BulkResponse;
-import org.elasticsearch.action.delete.DeleteRequest;
-import org.elasticsearch.action.get.GetResponse;
-import org.elasticsearch.action.get.MultiGetItemResponse;
-import org.elasticsearch.action.get.MultiGetRequest;
-import org.elasticsearch.action.get.MultiGetResponse;
-import org.elasticsearch.action.index.IndexRequest;
-import org.elasticsearch.action.index.IndexResponse;
-import org.elasticsearch.action.search.SearchRequest;
-import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.action.update.UpdateRequest;
-import org.elasticsearch.action.update.UpdateResponse;
-import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.Requests;
-import org.elasticsearch.client.RestClient;
-import org.elasticsearch.client.RestClientBuilder;
-import org.elasticsearch.client.RestHighLevelClient;
-import org.elasticsearch.common.unit.ByteSizeUnit;
-import org.elasticsearch.common.unit.ByteSizeValue;
-import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.index.query.QueryBuilder;
-import org.elasticsearch.index.query.QueryBuilders;
-import org.elasticsearch.rest.RestStatus;
-import org.elasticsearch.search.SearchHits;
-import org.elasticsearch.search.builder.SearchSourceBuilder;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-import java.util.function.BiConsumer;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-/**
- * ES 工具类
- *
- * @author Zhang Peng
- * @date 2023-06-27
- */
-@Slf4j
-public class ElasticsearchUtil {
-
- public static int CONNECT_TIMEOUT_MILLIS = 1000;
- public static int SOCKET_TIMEOUT_MILLIS = 30000;
- public static int CONNECTION_REQUEST_TIMEOUT_MILLIS = 500;
- public static int MAX_CONN_TOTAL = 30;
- public static int MAX_CONN_PER_ROUTE = 10;
-
- public static RestClient newRestClient(String hosts) {
- HttpHost[] httpHosts = toHttpHostList(hosts);
- RestClientBuilder builder = builder(httpHosts);
- try {
- return builder.build();
- } catch (Exception e) {
- log.error("【ES】connect failed.", e);
- return null;
- }
- }
-
- public static RestHighLevelClient newRestHighLevelClient(String hosts) {
- HttpHost[] httpHosts = toHttpHostList(hosts);
- RestClientBuilder builder = builder(httpHosts);
- try {
- return new RestHighLevelClient(builder);
- } catch (Exception e) {
- log.error("【ES】connect failed.", e);
- return null;
- }
- }
-
- public static BulkProcessor newAsyncBulkProcessor(RestHighLevelClient client) {
- BulkProcessor.Listener listener = new BulkProcessor.Listener() {
- @Override
- public void beforeBulk(long executionId, BulkRequest request) {
- }
-
- @Override
- public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
- if (response.hasFailures()) {
- log.error("Bulk [{}] executed with failures,response = {}", executionId,
- response.buildFailureMessage());
- }
- }
-
- @Override
- public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
- }
- };
- BiConsumer> bulkConsumer =
- (request, bulkListener) -> client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener);
- BulkProcessor bulkProcessor = BulkProcessor.builder(bulkConsumer, listener)
- // 1000条数据请求执行一次bulk
- .setBulkActions(1000)
- // 5mb的数据刷新一次bulk
- .setBulkSize(new ByteSizeValue(5L, ByteSizeUnit.MB))
- // 并发请求数量, 0不并发, 1并发允许执行
- .setConcurrentRequests(2)
- // 固定1s必须刷新一次
- .setFlushInterval(TimeValue.timeValueMillis(1000L))
- // 重试3次，间隔100ms
- .setBackoffPolicy(
- BackoffPolicy.constantBackoff(TimeValue.timeValueMillis(200L),
- 3)).build();
- Runtime.getRuntime().addShutdownHook(new Thread(() -> {
- try {
- bulkProcessor.flush();
- bulkProcessor.awaitClose(30, TimeUnit.SECONDS);
- } catch (Exception e) {
- log.error("Failed to close bulkProcessor", e);
- }
- log.info("bulkProcessor closed!");
- }));
- return bulkProcessor;
- }
-
- public static HttpHost[] toHttpHostList(String hosts) {
- if (StrUtil.isBlank(hosts)) {
- return null;
- }
- List strList = StrUtil.split(hosts, ",");
- List list = strList.stream().map(str -> {
- List params = StrUtil.split(str, ":");
- return new HttpHost(params.get(0), Integer.parseInt(params.get(1)), "http");
- }).collect(Collectors.toList());
- if (CollectionUtil.isEmpty(list)) {
- return new HttpHost[0];
- }
- return list.toArray(new HttpHost[0]);
- }
-
- public static RestClientBuilder builder(HttpHost[] httpHosts) {
- RestClientBuilder restClientBuilder = RestClient.builder(httpHosts);
- restClientBuilder.setRequestConfigCallback(builder -> {
- builder.setConnectTimeout(CONNECT_TIMEOUT_MILLIS);
- builder.setSocketTimeout(SOCKET_TIMEOUT_MILLIS);
- builder.setConnectionRequestTimeout(CONNECTION_REQUEST_TIMEOUT_MILLIS);
- return builder;
- });
- restClientBuilder.setHttpClientConfigCallback(builder -> {
- builder.setMaxConnTotal(MAX_CONN_TOTAL);
- builder.setMaxConnPerRoute(MAX_CONN_PER_ROUTE);
- return builder;
- });
- return restClientBuilder;
- }
-
- public static String insert(RestHighLevelClient client, String index, String type, T entity)
- throws IOException {
- Map map = toMap(entity);
- IndexRequest request = new IndexRequest(index, type).source(map);
- if (entity.getDocId() != null) {
- request.id(entity.getDocId());
- }
-
- IndexResponse response = client.index(request, RequestOptions.DEFAULT);
- if (response != null && response.getResult() == DocWriteResponse.Result.CREATED) {
- return response.getId();
- } else {
- return null;
- }
- }
-
- public static boolean batchInsert(RestHighLevelClient client, String index, String type,
- Collection list) throws IOException {
-
- if (CollectionUtil.isEmpty(list)) {
- return true;
- }
-
- BulkRequest bulkRequest = new BulkRequest();
- for (T entity : list) {
- Map map = ElasticsearchUtil.toMap(entity);
- IndexRequest request = new IndexRequest(index, type).source(map);
- if (entity.getDocId() != null) {
- request.id(entity.getDocId());
- }
- bulkRequest.add(request);
- }
-
- BulkResponse response = client.bulk(bulkRequest, RequestOptions.DEFAULT);
- return response != null && !response.hasFailures();
- }
-
- public static void asyncBatchInsert(RestHighLevelClient client, String index, String type,
- Collection list, ActionListener listener) {
-
- if (CollectionUtil.isEmpty(list)) {
- return;
- }
-
- BulkRequest bulkRequest = new BulkRequest();
- for (T entity : list) {
- Map map = ElasticsearchUtil.toMap(entity);
- IndexRequest request = new IndexRequest(index, type).source(map);
- if (entity.getDocId() != null) {
- request.id(entity.getDocId());
- }
- bulkRequest.add(request);
- }
-
- client.bulkAsync(bulkRequest, RequestOptions.DEFAULT, listener);
- }
-
- public static boolean updateById(RestHighLevelClient client, String index, String type,
- T entity) throws IOException {
-
- if (entity == null || entity.getDocId() == null) {
- return false;
- }
-
- Map map = toMap(entity);
- UpdateRequest request = new UpdateRequest(index, type, entity.getDocId()).doc(map);
- UpdateResponse response = client.update(request, RequestOptions.DEFAULT);
- return response != null && response.getResult() == DocWriteResponse.Result.UPDATED;
- }
-
- public static boolean batchUpdateById(RestHighLevelClient client, String index, String type,
- Collection list) throws IOException {
-
- if (CollectionUtil.isEmpty(list)) {
- return true;
- }
-
- BulkRequest bulkRequest = new BulkRequest();
- for (T entity : list) {
- if (entity == null || entity.getDocId() == null) {
- continue;
- }
- Map map = ElasticsearchUtil.toMap(entity);
- UpdateRequest request = new UpdateRequest(index, type, entity.getDocId()).doc(map);
- bulkRequest.add(request);
- }
-
- BulkResponse response = client.bulk(bulkRequest, RequestOptions.DEFAULT);
- return response != null && !response.hasFailures();
- }
-
- public static void asyncBatchUpdateById(RestHighLevelClient client, String index,
- String type, Collection list, ActionListener listener) {
-
- if (CollectionUtil.isEmpty(list)) {
- return;
- }
-
- BulkRequest bulkRequest = new BulkRequest();
- for (T entity : list) {
- if (entity == null || entity.getDocId() == null) {
- continue;
- }
- Map map = ElasticsearchUtil.toMap(entity);
- UpdateRequest request = new UpdateRequest(index, type, entity.getDocId()).doc(map);
- bulkRequest.add(request);
- }
-
- client.bulkAsync(bulkRequest, RequestOptions.DEFAULT, listener);
- }
-
- public static boolean deleteById(RestHighLevelClient client, String index, String type, String id)
- throws IOException {
- return batchDeleteById(client, index, type, Collections.singleton(id));
- }
-
- public static boolean batchDeleteById(RestHighLevelClient client, String index, String type, Collection ids)
- throws IOException {
-
- if (CollectionUtil.isEmpty(ids)) {
- return true;
- }
-
- BulkRequest bulkRequest = new BulkRequest();
- ids.forEach(id -> {
- DeleteRequest deleteRequest = new DeleteRequest(index, type, id);
- bulkRequest.add(deleteRequest);
- });
-
- BulkResponse response = client.bulk(bulkRequest, RequestOptions.DEFAULT);
- return response != null && !response.hasFailures();
- }
-
- public static void asyncBatchDeleteById(RestHighLevelClient client, String index, String type,
- Collection ids, ActionListener listener) {
-
- if (CollectionUtil.isEmpty(ids)) {
- return;
- }
-
- BulkRequest bulkRequest = new BulkRequest();
- ids.forEach(id -> {
- DeleteRequest deleteRequest = new DeleteRequest(index, type, id);
- bulkRequest.add(deleteRequest);
- });
-
- client.bulkAsync(bulkRequest, RequestOptions.DEFAULT, listener);
- }
-
- public static SearchResponse getById(RestHighLevelClient client, String index, String type, String id)
- throws IOException {
- SearchRequest searchRequest = Requests.searchRequest(index).types(type);
- QueryBuilder queryBuilder = QueryBuilders.idsQuery().addIds(id);
- SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
- sourceBuilder.query(queryBuilder);
- searchRequest.source(sourceBuilder);
- return client.search(searchRequest, RequestOptions.DEFAULT);
- }
-
- public static T pojoById(RestHighLevelClient client, String index, String type, String id, Class clazz)
- throws IOException {
- SearchResponse response = getById(client, index, type, id);
- if (response == null) {
- return null;
- }
- List list = ElasticsearchUtil.toPojoList(response, clazz);
- if (CollectionUtil.isEmpty(list)) {
- return null;
- }
- return list.get(0);
- }
-
- public static List pojoListByIds(RestHighLevelClient client, String index, String type,
- Collection ids, Class clazz) throws IOException {
-
- if (CollectionUtil.isEmpty(ids)) {
- return null;
- }
-
- MultiGetRequest request = new MultiGetRequest();
- for (String id : ids) {
- request.add(new MultiGetRequest.Item(index, type, id));
- }
-
- MultiGetResponse multiGetResponse = client.mget(request, RequestOptions.DEFAULT);
- if (null == multiGetResponse
- || multiGetResponse.getResponses() == null
- || multiGetResponse.getResponses().length <= 0) {
- return new ArrayList<>();
- }
-
- List list = new ArrayList<>();
- for (MultiGetItemResponse itemResponse : multiGetResponse.getResponses()) {
- if (itemResponse.isFailed()) {
- log.error("通过id获取文档失败", itemResponse.getFailure().getFailure());
- } else {
- T entity = ElasticsearchUtil.toPojo(itemResponse.getResponse(), clazz);
- if (entity != null) {
- list.add(entity);
- }
- }
- }
- return list;
- }
-
- public static Page pojoPage(RestHighLevelClient client, String index, String type,
- SearchSourceBuilder builder, Class clazz) throws IOException {
- SearchResponse response = query(client, index, type, builder);
- if (response == null || response.status() != RestStatus.OK) {
- return null;
- }
-
- List content = toPojoList(response, clazz);
- SearchHits searchHits = response.getHits();
- return new Page<>(searchHits.getTotalHits(), builder.from(), builder.size(), content);
- }
-
- public static SearchResponse query(RestHighLevelClient client, String index, String type,
- SearchSourceBuilder builder) throws IOException {
- SearchRequest request = new SearchRequest(index).types(type);
- request.source(builder);
- return client.search(request, RequestOptions.DEFAULT);
- }
-
- public static T toPojo(GetResponse response, Class clazz) {
- if (null == response) {
- return null;
- } else if (StrUtil.isBlank(response.getSourceAsString())) {
- return null;
- } else {
- return JSONUtil.toBean(response.getSourceAsString(), clazz);
- }
- }
-
- public static List toPojoList(SearchResponse response, Class clazz) {
-
- if (response == null || response.status() != RestStatus.OK) {
- return new ArrayList<>();
- }
-
- if (ArrayUtil.isEmpty(response.getHits().getHits())) {
- return new ArrayList<>();
- }
-
- return Stream.of(response.getHits().getHits())
- .map(hit -> JSONUtil.toBean(hit.getSourceAsString(), clazz))
- .collect(Collectors.toList());
- }
-
- public static Map toMap(T entity) {
- return JsonUtil.toMap(JsonUtil.toJson(entity));
- }
-
-}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/util/JsonUtil.java b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/util/JsonUtil.java
index 0ed7ddb5..dabe0df5 100644
--- a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/util/JsonUtil.java
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/util/JsonUtil.java
@@ -81,7 +81,7 @@ public static T toBean(String json, TypeReference typeReference) {
return null;
}
- public static String toJson(T obj) {
+ public static String toString(T obj) {
if (obj == null) {
return null;
}
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/test/java/io/github/dunwu/javadb/elasticsearch/BaseApplicationTests.java b/codes/javadb/elasticsearch/elasticsearch6/src/test/java/io/github/dunwu/javadb/elasticsearch/BaseApplicationTests.java
new file mode 100644
index 00000000..1fadeff5
--- /dev/null
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/test/java/io/github/dunwu/javadb/elasticsearch/BaseApplicationTests.java
@@ -0,0 +1,33 @@
+package io.github.dunwu.javadb.elasticsearch;
+
+import io.github.dunwu.javadb.elasticsearch.config.EnableElasticsearch;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.setup.MockMvcBuilders;
+import org.springframework.web.context.WebApplicationContext;
+
+@EnableElasticsearch
+@ExtendWith(SpringExtension.class)
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+public abstract class BaseApplicationTests {
+
+ protected MockMvc mockMvc;
+
+ @Autowired
+ private WebApplicationContext context;
+
+ @BeforeEach
+ public void setUp() {
+ mockMvc = MockMvcBuilders.webAppContextSetup(context).build();
+ }
+
+ @BeforeAll
+ public static void setEnvironmentInDev() {
+ }
+
+}
\ No newline at end of file
diff --git a/codes/javadb/elasticsearch/elasticsearch6/src/test/java/io/github/dunwu/javadb/elasticsearch/BaseElasticsearchTemplateTest.java b/codes/javadb/elasticsearch/elasticsearch6/src/test/java/io/github/dunwu/javadb/elasticsearch/BaseElasticsearchTemplateTest.java
new file mode 100644
index 00000000..4ec530a1
--- /dev/null
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/test/java/io/github/dunwu/javadb/elasticsearch/BaseElasticsearchTemplateTest.java
@@ -0,0 +1,295 @@
+package io.github.dunwu.javadb.elasticsearch;
+
+import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.core.util.StrUtil;
+import io.github.dunwu.javadb.elasticsearch.entity.BaseEsEntity;
+import io.github.dunwu.javadb.elasticsearch.entity.common.PageData;
+import io.github.dunwu.javadb.elasticsearch.entity.common.ScrollData;
+import io.github.dunwu.javadb.elasticsearch.util.JsonUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.assertj.core.api.Assertions;
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.get.GetResponse;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.index.query.BoolQueryBuilder;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * ElasticsearchTemplate 测试
+ *
+ * @author Zhang Peng
+ * @date 2023-11-13
+ */
+@Slf4j
+public abstract class BaseElasticsearchTemplateTest {
+
+ static final int FROM = 0;
+ static final int SIZE = 10;
+ static final String TEST_ID_01 = "1";
+ static final String TEST_ID_02 = "2";
+
+ protected ElasticsearchTemplate TEMPLATE = ElasticsearchFactory.newElasticsearchTemplate();
+
+ protected abstract String getAlias();
+
+ protected abstract String getIndex();
+
+ protected abstract String getType();
+
+ protected abstract int getShard();
+
+ protected abstract int getReplica();
+
+ protected abstract Class getEntityClass();
+
+ protected abstract Map getPropertiesMap();
+
+ protected abstract T getOneMockData(String id);
+
+ protected abstract List getMockList(int num);
+
+ protected void deleteIndex() throws IOException {
+ try {
+ Set set = TEMPLATE.getIndexSet(getAlias());
+ if (CollectionUtil.isNotEmpty(set)) {
+ for (String index : set) {
+ log.info("删除 alias: {}, index: {}", getAlias(), index);
+ TEMPLATE.deleteIndex(index);
+ }
+ }
+ } catch (IOException | ElasticsearchException e) {
+ log.error("删除索引失败！", e);
+ }
+ boolean exists = TEMPLATE.isIndexExists(getIndex());
+ Assertions.assertThat(exists).isFalse();
+ }
+
+ protected void createIndex() throws IOException {
+ boolean exists = TEMPLATE.isIndexExists(getIndex());
+ if (exists) {
+ return;
+ }
+ TEMPLATE.createIndex(getIndex(), getType(), getAlias(), getShard(), getReplica());
+ TEMPLATE.setMapping(getIndex(), getType(), getPropertiesMap());
+ exists = TEMPLATE.isIndexExists(getIndex());
+ Assertions.assertThat(exists).isTrue();
+ }
+
+ public void getIndexList() throws IOException {
+ Set