feat: ES6 java 客户端示例

2023-06-28 07:47:06 +08:00 · 2023-06-28 07:47:06 +08:00 · 8d1ef2086d
parent 2f93f74c77
commit 8d1ef2086d
9 changed files with 581 additions and 0 deletions
--- a/codes/javadb/elasticsearch/elasticsearch6/pom.xml
+++ b/codes/javadb/elasticsearch/elasticsearch6/pom.xml
@ -0,0 +1,48 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <groupId>io.github.dunwu</groupId>
+  <artifactId>javadb-elasticsearch6</artifactId>
+  <version>1.0.0</version>
+  <packaging>jar</packaging>
+
+  <properties>
+    <java.version>1.8</java.version>
+    <maven.compiler.source>${java.version}</maven.compiler.source>
+    <maven.compiler.target>${java.version}</maven.compiler.target>
+    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.elasticsearch.client</groupId>
+      <artifactId>elasticsearch-rest-high-level-client</artifactId>
+      <version>6.4.3</version>
+    </dependency>
+    <dependency>
+      <groupId>org.projectlombok</groupId>
+      <artifactId>lombok</artifactId>
+      <version>1.18.22</version>
+    </dependency>
+    <dependency>
+      <groupId>cn.hutool</groupId>
+      <artifactId>hutool-all</artifactId>
+      <version>5.7.20</version>
+    </dependency>
+
+    <dependency>
+      <groupId>ch.qos.logback</groupId>
+      <artifactId>logback-classic</artifactId>
+      <version>1.2.10</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.logging.log4j</groupId>
+      <artifactId>log4j-to-slf4j</artifactId>
+      <version>2.17.1</version>
+    </dependency>
+  </dependencies>
+
+</project>
--- a/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/Demo.java
+++ b/codes/javadb/elasticsearch/elasticsearch6/src/main/java/io/github/dunwu/javadb/elasticsearch/Demo.java
@ -0,0 +1,29 @@
+package io.github.dunwu.javadb.elasticsearch;
+
+import io.github.dunwu.javadb.elasticsearch.entity.User;
+import io.github.dunwu.javadb.elasticsearch.mapper.UserEsMapper;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+
+
+public class Demo {
+    public static void main(String[] args) throws IOException, InterruptedException {
+        UserEsMapper mapper = new UserEsMapper();
+
+        System.out.println("索引是否存在：" + mapper.isIndexExists());
+
+        User jack = User.builder().id(1L).username("jack").age(18).build();
+        User tom = User.builder().id(2L).username("tom").age(20).build();
+        List<User> users = Arrays.asList(jack, tom);
+
+        System.out.println("批量插入：" + mapper.batchInsert(users));
+        System.out.println("根据ID查询：" + mapper.getById("1").toString());
+        System.out.println("根据ID查询：" + mapper.pojoById("2").toString());
+        System.out.println("根据ID批量查询：" + mapper.pojoListByIds(Arrays.asList("1", "2")).toString());
+
+        Thread.sleep(1000);
+        System.out.println("根据ID批量删除：" + mapper.deleteByIds(Arrays.asList("1", "2")));
+    }
+}
--- 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
@ -0,0 +1,11 @@
+package io.github.dunwu.javadb.elasticsearch.entity;
+
+/**
+ * ES 实体接口
+ *
+ * @author <a href="mailto:forbreak@163.com">Zhang Peng</a>
+ * @since 2023-06-28
+ */
+public interface EsEntity {
+    Long getId();
+}
--- 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
@ -0,0 +1,21 @@
+package io.github.dunwu.javadb.elasticsearch.entity;
+
+import lombok.Builder;
+import lombok.Data;
+
+/**
+ * 用户实体
+ *
+ * @author <a href="mailto:forbreak@163.com">Zhang Peng</a>
+ * @since 2023-06-28
+ */
+@Data
+@Builder
+public class User implements EsEntity {
+    private Long id;
+    private String username;
+    private String password;
+    private Integer age;
+    private String email;
+
+}
--- 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
@ -0,0 +1,250 @@
+package io.github.dunwu.javadb.elasticsearch.mapper;
+
+import cn.hutool.core.bean.BeanUtil;
+import cn.hutool.core.bean.copier.CopyOptions;
+import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.core.io.IoUtil;
+import cn.hutool.core.lang.Assert;
+import io.github.dunwu.javadb.elasticsearch.entity.EsEntity;
+import io.github.dunwu.javadb.elasticsearch.util.ElasticsearchUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
+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.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.client.IndicesClient;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.Requests;
+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.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+
+/**
+ * ES Mapper 基础类
+ *
+ * @author <a href="mailto:forbreak@163.com">Zhang Peng</a>
+ * @date 2023-06-27
+ */
+@Slf4j
+public abstract class BaseEsMapper<T extends EsEntity> implements EsMapper<T>, Closeable {
+
+    public static final String HOSTS = "127.0.0.1:9200";
+
+    private BulkProcessor bulkProcessor;
+
+    private final RestHighLevelClient restHighLevelClient = ElasticsearchUtil.newRestHighLevelClient(HOSTS);
+
+    @Override
+    public RestHighLevelClient getClient() throws IOException {
+        Assert.notNull(restHighLevelClient, () -> new IOException("【ES】not connected."));
+        return restHighLevelClient;
+    }
+
+
+    @Override
+    public synchronized BulkProcessor getBulkProcessor() {
+        if (bulkProcessor == null) {
+            bulkProcessor = newAsyncBulkProcessor();
+        }
+        return bulkProcessor;
+    }
+
+    @Override
+    public boolean isIndexExists() throws IOException {
+        IndicesClient indicesClient = getClient().indices();
+        GetIndexRequest request = new GetIndexRequest();
+        request.indices(getIndexAlias());
+        return indicesClient.exists(request, RequestOptions.DEFAULT);
+    }
+
+    @Override
+    public SearchResponse getById(String id) throws IOException {
+        SearchRequest searchRequest = Requests.searchRequest(getIndexAlias());
+        QueryBuilder queryBuilder = QueryBuilders.idsQuery().addIds(id);
+        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
+        sourceBuilder.query(queryBuilder);
+        searchRequest.source(sourceBuilder);
+        return getClient().search(searchRequest, RequestOptions.DEFAULT);
+    }
+
+    @Override
+    public T pojoById(String id) throws IOException {
+        SearchResponse response = getById(id);
+        if (response == null) {
+            return null;
+        }
+        List<T> list = ElasticsearchUtil.toPojoList(response, getEntityClass());
+        if (CollectionUtil.isEmpty(list)) {
+            return null;
+        }
+        return list.get(0);
+    }
+
+    @Override
+    public List<T> pojoListByIds(Collection<String> ids) throws IOException {
+
+        if (CollectionUtil.isEmpty(ids)) {
+            return null;
+        }
+
+        MultiGetRequest request = new MultiGetRequest();
+        for (String id : ids) {
+            request.add(new MultiGetRequest.Item(getIndexAlias(), getIndexType(), id));
+        }
+
+        MultiGetResponse multiGetResponse = getClient().mget(request, RequestOptions.DEFAULT);
+        if (null == multiGetResponse || multiGetResponse.getResponses() == null || multiGetResponse.getResponses().length <= 0) {
+            return new ArrayList<>();
+        }
+
+        List<T> list = new ArrayList<>();
+        for (MultiGetItemResponse itemResponse : multiGetResponse.getResponses()) {
+            if (itemResponse.isFailed()) {
+                log.error("通过id获取文档失败", itemResponse.getFailure().getFailure());
+            } else {
+                T entity = ElasticsearchUtil.toPojo(itemResponse.getResponse(), getEntityClass());
+                if (entity != null) {
+                    list.add(entity);
+                }
+            }
+        }
+        return list;
+    }
+
+    @Override
+    public String insert(T entity) throws IOException {
+        Map<String, Object> map = new HashMap<>();
+        BeanUtil.beanToMap(entity, map, CopyOptions.create().ignoreError());
+        XContentBuilder builder = XContentFactory.jsonBuilder();
+        builder.startObject();
+        for (Map.Entry<String, Object> entry : map.entrySet()) {
+            String key = entry.getKey();
+            Object value = entry.getValue();
+            builder.field(key, value);
+        }
+        builder.endObject();
+
+        IndexRequest request = Requests.indexRequest(getIndexAlias()).type(getIndexType()).source(builder);
+        if (entity.getId() != null) {
+            request.id(entity.getId().toString());
+        }
+
+        IndexResponse response = getClient().index(request, RequestOptions.DEFAULT);
+        if (response == null) {
+            return null;
+        }
+        return response.getId();
+    }
+
+    @Override
+    public boolean batchInsert(Collection<T> list) throws IOException {
+
+        if (CollectionUtil.isEmpty(list)) {
+            return true;
+        }
+
+        BulkRequest bulkRequest = new BulkRequest();
+        for (T entity : list) {
+            Map<String, Object> map = ElasticsearchUtil.toMap(entity);
+            IndexRequest request = Requests.indexRequest(getIndexAlias()).type(getIndexType()).source(map);
+            if (entity.getId() != null) {
+                request.id(entity.getId().toString());
+            }
+            bulkRequest.add(request);
+        }
+
+        BulkResponse response = getClient().bulk(bulkRequest, RequestOptions.DEFAULT);
+        return !(response == null || response.hasFailures());
+    }
+
+    @Override
+    public boolean deleteById(String id) throws IOException {
+        return deleteByIds(Collections.singleton(id));
+    }
+
+    @Override
+    public boolean deleteByIds(Collection<String> ids) throws IOException {
+
+        if (CollectionUtil.isEmpty(ids)) {
+            return true;
+        }
+
+        BulkRequest bulkRequest = new BulkRequest();
+        ids.forEach(id -> {
+            DeleteRequest deleteRequest = Requests.deleteRequest(getIndexAlias()).type(getIndexType()).id(id);
+            bulkRequest.add(deleteRequest);
+        });
+
+        BulkResponse response = getClient().bulk(bulkRequest, RequestOptions.DEFAULT);
+        return response != null && !response.hasFailures();
+    }
+
+    private BulkProcessor newAsyncBulkProcessor() {
+        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<BulkRequest, ActionListener<BulkResponse>> bulkConsumer = (request, bulkListener) -> restHighLevelClient.bulkAsync(request, RequestOptions.DEFAULT, bulkListener);
+        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;
+    }
+
+    @Override
+    public void close() {
+        IoUtil.close(restHighLevelClient);
+    }
+
+}
--- 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
@ -0,0 +1,59 @@
+package io.github.dunwu.javadb.elasticsearch.mapper;
+
+import io.github.dunwu.javadb.elasticsearch.entity.EsEntity;
+import org.elasticsearch.action.bulk.BulkProcessor;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.RestHighLevelClient;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * ES Mapper
+ *
+ * @author <a href="mailto:forbreak@163.com">Zhang Peng</a>
+ * @date 2023-06-27
+ */
+public interface EsMapper<T extends EsEntity> {
+
+    /**
+     * 获取索引别名
+     */
+    String getIndexAlias();
+
+    /**
+     * 获取索引名
+     */
+    String getIndexName();
+
+    /**
+     * 获取索引类型
+     */
+    String getIndexType();
+
+    /**
+     * 获取实体类型
+     */
+    Class<T> getEntityClass();
+
+    RestHighLevelClient getClient() throws IOException;
+
+    BulkProcessor getBulkProcessor();
+
+    boolean isIndexExists() throws IOException;
+
+    SearchResponse getById(String id) throws IOException;
+
+    T pojoById(String id) throws IOException;
+
+    List<T> pojoListByIds(Collection<String> ids) throws IOException;
+
+    String insert(T entity) throws IOException;
+
+    boolean batchInsert(Collection<T> list) throws IOException;
+
+    boolean deleteById(String id) throws IOException;
+
+    boolean deleteByIds(Collection<String> ids) throws IOException;
+}
--- 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
@ -0,0 +1,33 @@
+package io.github.dunwu.javadb.elasticsearch.mapper;
+
+import io.github.dunwu.javadb.elasticsearch.entity.User;
+
+/**
+ * User ES Mapper
+ *
+ * @author <a href="mailto:forbreak@163.com">Zhang Peng</a>
+ * @date 2023-06-27
+ */
+public class UserEsMapper extends BaseEsMapper<User> {
+
+    @Override
+    public String getIndexAlias() {
+        return "user";
+    }
+
+    @Override
+    public String getIndexName() {
+        return "user";
+    }
+
+    @Override
+    public String getIndexType() {
+        return "_doc";
+    }
+
+    @Override
+    public Class<User> getEntityClass() {
+        return User.class;
+    }
+
+}
--- 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
@ -0,0 +1,129 @@
+package io.github.dunwu.javadb.elasticsearch.util;
+
+import cn.hutool.core.bean.BeanUtil;
+import cn.hutool.core.bean.copier.CopyOptions;
+import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.json.JSONUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.http.HttpHost;
+import org.elasticsearch.action.get.GetResponse;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.search.SearchHit;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * ES 工具类
+ *
+ * @author <a href="mailto:forbreak@163.com">Zhang Peng</a>
+ * @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 HttpHost[] toHttpHostList(String hosts) {
+        if (StrUtil.isBlank(hosts)) {
+            return null;
+        }
+        List<String> strList = StrUtil.split(hosts, ",");
+        List<HttpHost> list = strList.stream().map(str -> {
+            List<String> 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 <T> T toPojo(GetResponse response, Class<T> clazz) {
+        if (null == response) {
+            return null;
+        } else if (StrUtil.isBlank(response.getSourceAsString())) {
+            return null;
+        } else {
+            return JSONUtil.toBean(response.getSourceAsString(), clazz);
+        }
+    }
+
+    public static <T> List<T> toPojoList(SearchResponse response, Class<T> clazz) {
+        List<T> list = null;
+        try {
+            if (response != null) {
+                SearchHit[] searchHits = response.getHits().getHits();
+                T entity;
+                list = new ArrayList<T>(searchHits.length);
+                for (SearchHit hit : searchHits) {
+                    if (null == hit) {
+                        continue;
+                    }
+                    entity = JSONUtil.toBean(hit.getSourceAsString(), clazz);
+                    list.add(entity);
+                }
+            }
+        } catch (Exception e) {
+            log.error("解析ES返回结果异常, response:{}", response);
+            throw e;
+        }
+        return list;
+    }
+
+    public static <T> Map<String, Object> toMap(T entity) {
+        Map<String, Object> map = new HashMap<>();
+        BeanUtil.beanToMap(entity, map, CopyOptions.create().ignoreError());
+        return map;
+    }
+
+}
--- a/codes/javadb/elasticsearch/pom.xml
+++ b/codes/javadb/elasticsearch/pom.xml
@ -9,6 +9,7 @@
  <packaging>pom</packaging>

  <modules>
+    <module>elasticsearch6</module>
    <module>elasticsearch7</module>
  </modules>
 </project>