Public/cf_imes_server
6
0
Fork 0
mirror of http://192.168.1.205:9980/cf_devdept2/cf_imes_server.git synced 2026-08-12 21:02:08 +08:00
Code Issues Packages Projects Releases Wiki Activity

elasticsearch移除

Browse Source
This commit is contained in:
gaoqr
2026-06-17 16:10:58 +08:00
parent 3b41c42951
commit 7c224fe197
47 changed files with 1087 additions and 1910 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cf-framework/cf-spring-boot-starter-elasticsearch/src/main/java/com/cf/imes/framework/es/config/ChenfengElasticsearchAutoConfiguration.java
+212 -212
View File
@@ -1,212 +1,212 @@
package com.cf.imes.framework.es.config;
import cn.hutool.core.io.file.PathUtil;
import cn.hutool.core.text.CharSequenceUtil;
import cn.hutool.core.util.ObjectUtil;
import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import com.cf.imes.framework.common.util.encrypt.AesUtils;
import com.cf.imes.framework.es.core.service.ESDocumentService;
import com.cf.imes.framework.es.core.service.ESDocumentServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.conn.ssl.NoopHostnameVerifier;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.message.BasicHeader;
import org.apache.http.ssl.SSLContextBuilder;
import org.apache.http.ssl.SSLContexts;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.util.StringUtils;
import javax.net.ssl.SSLContext;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.security.KeyManagementException;
import java.security.KeyStore;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.security.cert.CertificateException;
import java.time.Duration;
import java.util.List;
/**
* @author there
*/
@AutoConfiguration
@ConditionalOnClass(ElasticsearchClient.class)
@EnableConfigurationProperties(EsProperties.class)
@Slf4j
public class ChenfengElasticsearchAutoConfiguration {
@Value("${chenfeng.encrypt.publicKey:}")
private String publicKey;
/**
* 同步方式
*
*/
@Bean
public ElasticsearchClient elasticsearchClient(RestClientTransport transport) {
return new ElasticsearchClient(transport);
}
/**
* 异步方式
*
*/
@Bean
public ElasticsearchAsyncClient elasticsearchAsyncClient(RestClientTransport transport) {
return new ElasticsearchAsyncClient(transport);
}
@Bean
public ESDocumentService esDocumentService(ElasticsearchClient elasticsearchClient, ElasticsearchAsyncClient elasticsearchAsyncClient) {
return new ESDocumentServiceImpl(elasticsearchClient, elasticsearchAsyncClient);
}
/**
* 获取客户端 RestClientTransport
*/
@Bean
public RestClientTransport getTransport(RestClient client){
return new RestClientTransport(client, new JacksonJsonpMapper());
}
/**
* 获取客户端RestClient
* chenfeng.encrypt.enable：true
*
* @param properties es配置
*/
@Bean
@ConditionalOnProperty(name = "chenfeng.encrypt.enable", havingValue = "true")
public RestClient getAuthRestClient(EsProperties properties) {
return getRestClient(properties, true);
}
/**
* 获取客户端RestClient（默认）
* chenfeng.encrypt.enable：false或者缺省
*
* @param properties es配置
*/
@Bean
@ConditionalOnProperty(name = "chenfeng.encrypt.enable", havingValue = "false", matchIfMissing = true)
public RestClient getRestClient(EsProperties properties) {
return getRestClient(properties, false);
}
/**
* 动态获取es client连接
*
* @param properties 配置
* @param needDecrypt 敏感信息是否需要解密
* @return
*/
private RestClient getRestClient(EsProperties properties, boolean needDecrypt) {
// 配置账号密码
CredentialsProvider credentialsProvider;
String username = properties.getUsername();
String password = properties.getPassword();
if (CharSequenceUtil.isAllNotEmpty(username, password)) {
credentialsProvider = new BasicCredentialsProvider();
credentialsProvider.setCredentials(AuthScope.ANY,
new UsernamePasswordCredentials(needDecrypt ? AesUtils.decrypt(username, publicKey) : username, needDecrypt ? AesUtils.decrypt(password, publicKey) : password));
} else {
credentialsProvider = null;
}
RestClientBuilder.HttpClientConfigCallback httpClientConfigCallback = httpClientBuilder -> {
// 设置账号密码、连接信息
HttpAsyncClientBuilder httpAsyncClientBuilder = httpClientBuilder
.setDefaultRequestConfig(RequestConfig.custom()
.setConnectTimeout((int) properties.getConnectionTimeout().toMillis())
.setSocketTimeout((int) properties.getSocketTimeout().toMillis())
.setConnectionRequestTimeout((int) properties.getConnectionRequestTimeout().toMillis())
// 启用数据压缩
.setContentCompressionEnabled(true)
.build());
if (ObjectUtil.isNotNull(credentialsProvider)) {
httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
}
if (properties.isSecurityHttpSslEnable()) {
// 开启ssl配置连接证书
httpAsyncClientBuilder.setSSLContext(buildSSLContext(properties)).setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE);
}
// 手动开启keepalive
httpAsyncClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setSoKeepAlive(true).build());
httpAsyncClientBuilder.setDefaultHeaders(List.of(new BasicHeader("Accept-Encoding", "gzip")));
// 手动设置保活时长
httpAsyncClientBuilder.setKeepAliveStrategy(((response, context) -> Duration.ofMinutes(5).toMillis()));
return httpAsyncClientBuilder;
};
return RestClient.builder(toHttpHost(properties.getUris(), properties.isSecurityHttpSslEnable())).setHttpClientConfigCallback(httpClientConfigCallback).build();
}
/**
* 构建ssl请求信息
*
* @param properties es配置
* @return
*/
private SSLContext buildSSLContext(EsProperties properties) {
SSLContext sslContext = null;
try {
Path trustStorePath = Paths.get(properties.getCertificatePath());
KeyStore trustStore = KeyStore.getInstance("pkcs12");
try (InputStream is = PathUtil.getInputStream(trustStorePath)) {
trustStore.load(is, properties.getPassword().toCharArray());
}
SSLContextBuilder sslContextBuilder = SSLContexts.custom()
.loadTrustMaterial(trustStore, null);
sslContext = sslContextBuilder.build();
} catch (CertificateException | IOException | KeyStoreException | NoSuchAlgorithmException |
KeyManagementException e) {
log.error("ES连接认证失败", e);
}
return sslContext;
}
/**
* 解析域名
*
* @param hosts 域名字符串
* @param isSslEnable 是否开启ssl
* @return
*/
private HttpHost[] toHttpHost(String hosts, boolean isSslEnable) {
if (!StringUtils.hasLength(hosts)) {
throw new IllegalArgumentException("invalid elasticsearch configuration. elasticsearch.hosts不能为空！");
}
// 多个IP逗号隔开
String[] hostArray = hosts.split(",");
HttpHost[] httpHosts = new HttpHost[hostArray.length];
HttpHost httpHost;
for (int i = 0; i < hostArray.length; i++) {
String[] strings = hostArray[i].split(":");
httpHost = new HttpHost(strings[0], Integer.parseInt(strings[1]), isSslEnable ? "https" : "http");
httpHosts[i] = httpHost;
}
return httpHosts;
}
}
//package com.cf.imes.framework.es.config;
//
//import cn.hutool.core.io.file.PathUtil;
//import cn.hutool.core.text.CharSequenceUtil;
//import cn.hutool.core.util.ObjectUtil;
//import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient;
//import co.elastic.clients.elasticsearch.ElasticsearchClient;
//import co.elastic.clients.json.jackson.JacksonJsonpMapper;
//import co.elastic.clients.transport.rest_client.RestClientTransport;
//import com.cf.imes.framework.common.util.encrypt.AesUtils;
//import com.cf.imes.framework.es.core.service.ESDocumentService;
//import com.cf.imes.framework.es.core.service.ESDocumentServiceImpl;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.http.HttpHost;
//import org.apache.http.auth.AuthScope;
//import org.apache.http.auth.UsernamePasswordCredentials;
//import org.apache.http.client.CredentialsProvider;
//import org.apache.http.client.config.RequestConfig;
//import org.apache.http.conn.ssl.NoopHostnameVerifier;
//import org.apache.http.impl.client.BasicCredentialsProvider;
//import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
//import org.apache.http.impl.nio.reactor.IOReactorConfig;
//import org.apache.http.message.BasicHeader;
//import org.apache.http.ssl.SSLContextBuilder;
//import org.apache.http.ssl.SSLContexts;
//import org.elasticsearch.client.RestClient;
//import org.elasticsearch.client.RestClientBuilder;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.boot.autoconfigure.AutoConfiguration;
//import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
//import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
//import org.springframework.boot.context.properties.EnableConfigurationProperties;
//import org.springframework.context.annotation.Bean;
//import org.springframework.util.StringUtils;
//
//import javax.net.ssl.SSLContext;
//import java.io.IOException;
//import java.io.InputStream;
//import java.nio.file.Path;
//import java.nio.file.Paths;
//import java.security.KeyManagementException;
//import java.security.KeyStore;
//import java.security.KeyStoreException;
//import java.security.NoSuchAlgorithmException;
//import java.security.cert.CertificateException;
//import java.time.Duration;
//import java.util.List;
//
//
///**
// * @author there
// */
//@AutoConfiguration
//@ConditionalOnClass(ElasticsearchClient.class)
//@EnableConfigurationProperties(EsProperties.class)
//@Slf4j
//public class ChenfengElasticsearchAutoConfiguration {
//
// @Value("${chenfeng.encrypt.publicKey:}")
// private String publicKey;
//
// /**
// * 同步方式
// *
// */
// @Bean
// public ElasticsearchClient elasticsearchClient(RestClientTransport transport) {
// return new ElasticsearchClient(transport);
// }
//
// /**
// * 异步方式
// *
// */
// @Bean
// public ElasticsearchAsyncClient elasticsearchAsyncClient(RestClientTransport transport) {
// return new ElasticsearchAsyncClient(transport);
// }
//
//
// @Bean
// public ESDocumentService esDocumentService(ElasticsearchClient elasticsearchClient, ElasticsearchAsyncClient elasticsearchAsyncClient) {
// return new ESDocumentServiceImpl(elasticsearchClient, elasticsearchAsyncClient);
// }
//
// /**
// * 获取客户端 RestClientTransport
// */
// @Bean
// public RestClientTransport getTransport(RestClient client){
// return new RestClientTransport(client, new JacksonJsonpMapper());
// }
//
//
// /**
// * 获取客户端RestClient
// * chenfeng.encrypt.enable：true
// *
// * @param properties es配置
// */
// @Bean
// @ConditionalOnProperty(name = "chenfeng.encrypt.enable", havingValue = "true")
// public RestClient getAuthRestClient(EsProperties properties) {
// return getRestClient(properties, true);
// }
//
// /**
// * 获取客户端RestClient（默认）
// * chenfeng.encrypt.enable：false或者缺省
// *
// * @param properties es配置
// */
// @Bean
// @ConditionalOnProperty(name = "chenfeng.encrypt.enable", havingValue = "false", matchIfMissing = true)
// public RestClient getRestClient(EsProperties properties) {
// return getRestClient(properties, false);
// }
//
// /**
// * 动态获取es client连接
// *
// * @param properties 配置
// * @param needDecrypt 敏感信息是否需要解密
// * @return
// */
// private RestClient getRestClient(EsProperties properties, boolean needDecrypt) {
// // 配置账号密码
// CredentialsProvider credentialsProvider;
// String username = properties.getUsername();
// String password = properties.getPassword();
// if (CharSequenceUtil.isAllNotEmpty(username, password)) {
// credentialsProvider = new BasicCredentialsProvider();
// credentialsProvider.setCredentials(AuthScope.ANY,
// new UsernamePasswordCredentials(needDecrypt ? AesUtils.decrypt(username, publicKey) : username, needDecrypt ? AesUtils.decrypt(password, publicKey) : password));
// } else {
// credentialsProvider = null;
// }
// RestClientBuilder.HttpClientConfigCallback httpClientConfigCallback = httpClientBuilder -> {
// // 设置账号密码、连接信息
// HttpAsyncClientBuilder httpAsyncClientBuilder = httpClientBuilder
// .setDefaultRequestConfig(RequestConfig.custom()
// .setConnectTimeout((int) properties.getConnectionTimeout().toMillis())
// .setSocketTimeout((int) properties.getSocketTimeout().toMillis())
// .setConnectionRequestTimeout((int) properties.getConnectionRequestTimeout().toMillis())
// // 启用数据压缩
// .setContentCompressionEnabled(true)
// .build());
// if (ObjectUtil.isNotNull(credentialsProvider)) {
// httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
// }
// if (properties.isSecurityHttpSslEnable()) {
// // 开启ssl配置连接证书
// httpAsyncClientBuilder.setSSLContext(buildSSLContext(properties)).setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE);
// }
// // 手动开启keepalive
// httpAsyncClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setSoKeepAlive(true).build());
// httpAsyncClientBuilder.setDefaultHeaders(List.of(new BasicHeader("Accept-Encoding", "gzip")));
// // 手动设置保活时长
// httpAsyncClientBuilder.setKeepAliveStrategy(((response, context) -> Duration.ofMinutes(5).toMillis()));
// return httpAsyncClientBuilder;
// };
// return RestClient.builder(toHttpHost(properties.getUris(), properties.isSecurityHttpSslEnable())).setHttpClientConfigCallback(httpClientConfigCallback).build();
// }
//
// /**
// * 构建ssl请求信息
// *
// * @param properties es配置
// * @return
// */
// private SSLContext buildSSLContext(EsProperties properties) {
// SSLContext sslContext = null;
// try {
// Path trustStorePath = Paths.get(properties.getCertificatePath());
// KeyStore trustStore = KeyStore.getInstance("pkcs12");
// try (InputStream is = PathUtil.getInputStream(trustStorePath)) {
// trustStore.load(is, properties.getPassword().toCharArray());
// }
// SSLContextBuilder sslContextBuilder = SSLContexts.custom()
// .loadTrustMaterial(trustStore, null);
// sslContext = sslContextBuilder.build();
// } catch (CertificateException | IOException | KeyStoreException | NoSuchAlgorithmException |
// KeyManagementException e) {
// log.error("ES连接认证失败", e);
// }
// return sslContext;
// }
//
//
// /**
// * 解析域名
// *
// * @param hosts 域名字符串
// * @param isSslEnable 是否开启ssl
// * @return
// */
// private HttpHost[] toHttpHost(String hosts, boolean isSslEnable) {
// if (!StringUtils.hasLength(hosts)) {
// throw new IllegalArgumentException("invalid elasticsearch configuration. elasticsearch.hosts不能为空！");
// }
// // 多个IP逗号隔开
// String[] hostArray = hosts.split(",");
// HttpHost[] httpHosts = new HttpHost[hostArray.length];
// HttpHost httpHost;
// for (int i = 0; i < hostArray.length; i++) {
// String[] strings = hostArray[i].split(":");
// httpHost = new HttpHost(strings[0], Integer.parseInt(strings[1]), isSslEnable ? "https" : "http");
// httpHosts[i] = httpHost;
// }
// return httpHosts;
// }
//}
cf-framework/cf-spring-boot-starter-elasticsearch/src/main/java/com/cf/imes/framework/es/core/dal/ESDocument.java
+1 -2
View File
@@ -1,7 +1,6 @@
package com.cf.imes.framework.es.core.dal;
import com.cf.imes.framework.es.core.valid.UpdateGroup;
import com.fasterxml.jackson.annotation.JsonFormat;
import lombok.Data;
import org.springframework.format.annotation.DateTimeFormat;
@@ -14,7 +13,7 @@ import static com.cf.imes.framework.common.util.date.DateUtils.TIME_ZONE_DEFAULT
*/
@Data
public class ESDocument {
@NotBlank(groups = UpdateGroup.class, message = "文档ID不能空")
@NotBlank(message = "文档ID不能空")
public String id;
@DateTimeFormat(pattern = FORMAT_YEAR_MONTH_DAY_HOUR_MINUTE_SECOND)
@JsonFormat(pattern = FORMAT_YEAR_MONTH_DAY_HOUR_MINUTE_SECOND, timezone = TIME_ZONE_DEFAULT)
cf-framework/cf-spring-boot-starter-elasticsearch/src/main/java/com/cf/imes/framework/es/core/service/ESDocumentService.java
+91 -103
View File
@@ -1,112 +1,100 @@
package com.cf.imes.framework.es.core.service;
import cn.hutool.json.JSONObject;
import co.elastic.clients.elasticsearch._types.ElasticsearchException;
import co.elastic.clients.elasticsearch._types.Result;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.IndexResponse;
import com.cf.imes.framework.es.core.dal.ESDocument;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.function.BiConsumer;
/**
* 公用es服务接口
*/
public interface ESDocumentService {
/**
* 新增一个文档，此种方式若发现没有索引，会自动创建一个索引
* @param idxName 索引名
* @param idxId 索引id
* @param document 文档对象
*/
<T> IndexResponse createByFluentDSL(String idxName, String idxId, ESDocument document) throws Exception;
/**
* 新增一个文档，此种方式若发现没有索引，会自动创建一个索引
* @param idxName 索引名
* @param idxId 索引id
* @param document 文档对象
*/
<T> IndexResponse createByBuilderPattern(String idxName, String idxId, ESDocument document) throws Exception;
/**
* 用JSON字符串创建文档，此种方式若发现没有索引，会自动创建一个索引
* @param idxName 索引名
* @param idxId 索引id
* @param jsonContent json字符串
*/
IndexResponse createByJson(String idxName, String idxId, String jsonContent) throws Exception;
/**
* 异步新增文档，此种方式若发现没有索引，会自动创建一个索引
* @param idxName 索引名
* @param idxId 索引id
* @param document 文档
* @param action 操作
*/
<T> void createAsync(String idxName, String idxId, T document, BiConsumer<IndexResponse, Throwable> action);
/**
* 批量增加文档
* @param idxName 索引名
* @param documents 要增加的对象集合
* @return 批量操作的结果
*/
<T> BulkResponse bulkCreate(String idxName, List<?extends ESDocument> documents) throws IOException, ElasticsearchException;
/**
* 批量更新文档
* @param idxName 索引名
* @param documents 要更新的对象集合
* @return 批量更新的结果
*/
<T> BulkResponse bulkUpdate(String idxName, List<?extends ESDocument> documents) throws Exception;
/**
* 根据文档id查找文档
* @param idxName 索引名
* @param docId 文档id
* @return Object类型的查找结果
*/
<T> T getById(String idxName, String docId ,Class<T> tClass) throws IOException;
/**
*
* @param idxName 索引名称
* @param docId 文档id
* @param tClass 返回的类型
* @param map 修改内容的map
*/
<T> Result updateById(String idxName, String docId, Class<T> tClass, Map<String,Object> map) throws IOException, ElasticsearchException;
/**
* 根据文档id查找文档，返回类型是ObjectNode
* @param idxName 索引名
* @param docId 文档id
* @return ObjectNode类型的查找结果
*/
JSONObject getObjectNodeById(String idxName, String docId) throws IOException;
/**
* 根据文档id删除文档
* @param idxName 索引名
* @param docId 文档id
* @return Object类型的查找结果
*/
Boolean deleteById(String idxName, String docId) throws IOException;
/**
* 批量删除文档
* @param idxName 索引名
* @param docIds 要删除的文档id集合
*/
BulkResponse bulkDeleteByIds(String idxName, List<String> docIds) throws Exception;
// /**
// * 新增一个文档，此种方式若发现没有索引，会自动创建一个索引
// * @param idxName 索引名
// * @param idxId 索引id
// * @param document 文档对象
// */
// <T> IndexResponse createByFluentDSL(String idxName, String idxId, ESDocument document) throws Exception;
//
// /**
// * 新增一个文档，此种方式若发现没有索引，会自动创建一个索引
// * @param idxName 索引名
// * @param idxId 索引id
// * @param document 文档对象
// */
// <T> IndexResponse createByBuilderPattern(String idxName, String idxId, ESDocument document) throws Exception;
//
// /**
// * 用JSON字符串创建文档，此种方式若发现没有索引，会自动创建一个索引
// * @param idxName 索引名
// * @param idxId 索引id
// * @param jsonContent json字符串
// */
// IndexResponse createByJson(String idxName, String idxId, String jsonContent) throws Exception;
//
//
// /**
// * 异步新增文档，此种方式若发现没有索引，会自动创建一个索引
// * @param idxName 索引名
// * @param idxId 索引id
// * @param document 文档
// * @param action 操作
// */
// <T> void createAsync(String idxName, String idxId, T document, BiConsumer<IndexResponse, Throwable> action);
//
// /**
// * 批量增加文档
// * @param idxName 索引名
// * @param documents 要增加的对象集合
// * @return 批量操作的结果
// */
// <T> BulkResponse bulkCreate(String idxName, List<?extends ESDocument> documents) throws IOException, ElasticsearchException;
//
//
// /**
// * 批量更新文档
// * @param idxName 索引名
// * @param documents 要更新的对象集合
// * @return 批量更新的结果
// */
// <T> BulkResponse bulkUpdate(String idxName, List<?extends ESDocument> documents) throws Exception;
//
//
// /**
// * 根据文档id查找文档
// * @param idxName 索引名
// * @param docId 文档id
// * @return Object类型的查找结果
// */
// <T> T getById(String idxName, String docId ,Class<T> tClass) throws IOException;
//
// /**
// *
// * @param idxName 索引名称
// * @param docId 文档id
// * @param tClass 返回的类型
// * @param map 修改内容的map
// */
// <T> Result updateById(String idxName, String docId, Class<T> tClass, Map<String,Object> map) throws IOException, ElasticsearchException;
//
// /**
// * 根据文档id查找文档，返回类型是ObjectNode
// * @param idxName 索引名
// * @param docId 文档id
// * @return ObjectNode类型的查找结果
// */
// JSONObject getObjectNodeById(String idxName, String docId) throws IOException;
//
// /**
// * 根据文档id删除文档
// * @param idxName 索引名
// * @param docId 文档id
// * @return Object类型的查找结果
// */
// Boolean deleteById(String idxName, String docId) throws IOException;
//
// /**
// * 批量删除文档
// * @param idxName 索引名
// * @param docIds 要删除的文档id集合
// */
// BulkResponse bulkDeleteByIds(String idxName, List<String> docIds) throws Exception;
}
cf-framework/cf-spring-boot-starter-elasticsearch/src/main/java/com/cf/imes/framework/es/core/service/ESDocumentServiceImpl.java
+237 -259
View File
@@ -1,267 +1,245 @@
package com.cf.imes.framework.es.core.service;
import cn.hutool.core.lang.Snowflake;
import cn.hutool.core.text.CharSequenceUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.json.JSONObject;
import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.ElasticsearchException;
import co.elastic.clients.elasticsearch._types.Result;
import co.elastic.clients.elasticsearch.core.*;
import com.cf.imes.framework.es.core.dal.ESDocument;
import com.cf.imes.framework.security.core.LoginUser;
import com.cf.imes.framework.security.core.util.SecurityFrameworkUtils;
import java.io.IOException;
import java.io.StringReader;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.function.BiConsumer;
/**
* @author there
*/
public class ESDocumentServiceImpl implements ESDocumentService {
//同步客户端
private final ElasticsearchClient elasticsearchClient;
// 异步客户端
private final ElasticsearchAsyncClient elasticsearchAsyncClient;
private Snowflake snowflake = IdUtil.getSnowflake();
SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
public ESDocumentServiceImpl(ElasticsearchClient elasticsearchClient, ElasticsearchAsyncClient elasticsearchAsyncClient) {
this.elasticsearchClient = elasticsearchClient;
this.elasticsearchAsyncClient = elasticsearchAsyncClient;
}
@Override
public <T> IndexResponse createByFluentDSL(String idxName, String idxId, ESDocument document) throws Exception {
Date date = new Date();
LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();
document.setCreator(loginUser.getNickname());
document.setUpdater(loginUser.getNickname());
document.setOrganId(loginUser.getOrganId());
document.setCreateTime(simpleDateFormat.format(date));
document.setUpdateTime(simpleDateFormat.format(date));
if (CharSequenceUtil.isBlank(document.getId())) {
document.setId(snowflake.nextIdStr());
}
return elasticsearchClient.index(idx -> idx
.index(idxName)
.id(document.getId())
.document(document));
}
/**
* BuilderPattern 方式创建文档
*
* @param idxName 索引名
* @param idxId 索引id
* @param document 文档对象
*/
@Override
public <T> IndexResponse createByBuilderPattern(String idxName, String idxId, ESDocument document) throws Exception {
LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();
Date date = new Date();
document.setCreator(loginUser.getNickname());
document.setUpdater(loginUser.getNickname());
document.setOrganId(loginUser.getOrganId());
document.setCreateTime(simpleDateFormat.format(date));
document.setUpdateTime(simpleDateFormat.format(date));
IndexRequest.Builder<Object> indexReqBuilder = new IndexRequest.Builder<>();
indexReqBuilder.index(idxName);
if (CharSequenceUtil.isBlank(idxId)) {
idxId = snowflake.nextIdStr();
}
indexReqBuilder.id(idxId);
indexReqBuilder.document(document);
return elasticsearchClient.index(indexReqBuilder.build());
}
/**
* json方式创建文档
*
* @param idxName 索引名
* @param idxId 索引id
* @param jsonContent json字符串
*/
@Override
public IndexResponse createByJson(String idxName, String idxId, String jsonContent) throws Exception {
if (CharSequenceUtil.isBlank(idxId)) {
idxId = snowflake.nextIdStr();
}
String finalIdxId = idxId;
return elasticsearchClient.index(i -> i
.index(idxName)
.id(finalIdxId)
.withJson(new StringReader(jsonContent))
);
}
/**
* 异步方式创建文档
*
* @param idxName 索引名
* @param idxId 索引id
* @param document 文档
* @param action 操作
*/
@Override
public <T> void createAsync(String idxName, String idxId, T document, BiConsumer<IndexResponse, Throwable> action) {
elasticsearchAsyncClient.index(idx -> idx
.index(idxName)
.id(idxId)
.document(document)
).whenComplete(action);
}
/**
* 批量方式创建文档
*
* @param idxName 索引名
* @param documents 要增加的对象集合
*/
@Override
public <T> BulkResponse bulkCreate(String idxName, List<? extends ESDocument> documents) throws IOException, ElasticsearchException {
BulkRequest.Builder br = new BulkRequest.Builder();
LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();
Date date = new Date();
LocalDateTime now = LocalDateTime.now();
documents.forEach(esDocument -> {
if (CharSequenceUtil.isBlank(esDocument.getId())) {
esDocument.setId(snowflake.nextIdStr());
}
esDocument.setCreator(loginUser.getNickname());
esDocument.setCreateTime(simpleDateFormat.format(date));
esDocument.setUpdater(loginUser.getNickname());
esDocument.setOrganId(loginUser.getOrganId());
esDocument.setUpdateTime(simpleDateFormat.format(date));
br.operations(op -> op.index(idx -> idx
.index(idxName)
.id(esDocument.getId())
.document(esDocument)));
});
return elasticsearchClient.bulk(br.build());
}
/**
* 批量方式更新文档
*
* @param idxName 索引名
* @param documents 要更新的对象集合
*/
@Override
public <T> BulkResponse bulkUpdate(String idxName, List<? extends ESDocument> documents) throws Exception {
BulkRequest.Builder br = new BulkRequest.Builder();
LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();
Date date = new Date();
documents.forEach(esDocument -> {
esDocument.setCreator(loginUser.getNickname());
esDocument.setUpdater(loginUser.getNickname());
esDocument.setOrganId(loginUser.getOrganId());
esDocument.setUpdateTime(simpleDateFormat.format(date));
br.operations(op -> op.index(idx -> idx
.index(idxName)
.id(esDocument.getId())
.document(esDocument)));
});
return elasticsearchClient.bulk(br.build());
}
/**
* @param idxName 索引名称
* @param docId 文档id
* @param tClass 返回的类型
* @param map 修改内容的map
* Map<String, Object> map = new HashMap<>();
* map.put("age", 35);
* 把年龄改成35
*/
@Override
public <T> Result updateById(String idxName, String docId, Class<T> tClass, Map<String, Object> map) throws IOException, ElasticsearchException {
UpdateResponse<T> response = elasticsearchClient.update(e -> e.index(idxName).id(docId).doc(map), tClass);
return response.result();
}
/**
* 文档id查询信息
*
* @param idxName 索引名
* @param docId 文档id
*/
@Override
public <T> T getById(String idxName, String docId, Class<T> tClass) throws IOException {
GetResponse<T> response = elasticsearchClient.get(g -> g
.index(idxName)
.id(docId),
tClass);
return response.found() ? response.source() : null;
}
/**
* 根据索引名称和文档id查询ObjectNode
*
* @param idxName 索引名
* @param docId 文档id
*/
@Override
public JSONObject getObjectNodeById(String idxName, String docId) throws IOException {
GetResponse<JSONObject> response = elasticsearchClient.get(g -> g
.index(idxName)
.id(docId),
JSONObject.class);
return response.found() ? response.source() : null;
}
/**
* 单条输出
*
* @param idxName 索引名
* @param docId 文档id
*/
@Override
public Boolean deleteById(String idxName, String docId) throws IOException {
DeleteResponse delete = elasticsearchClient.delete(d -> d
.index(idxName)
.id(docId));
return delete.forcedRefresh();
}
/**
* 批量删除
*
* @param idxName 索引名
* @param docIds 要删除的文档id集合
*/
@Override
public BulkResponse bulkDeleteByIds(String idxName, List<String> docIds) throws Exception {
BulkRequest.Builder br = new BulkRequest.Builder();
// 将每一个对象都放入builder中
docIds.forEach(id -> br
.operations(op -> op
.delete(d -> d
.index(idxName)
.id(id))));
return elasticsearchClient.bulk(br.build());
}
// //同步客户端
// private final ElasticsearchClient elasticsearchClient;
//
// // 异步客户端
// private final ElasticsearchAsyncClient elasticsearchAsyncClient;
//
// private Snowflake snowflake = IdUtil.getSnowflake();
//
// SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//
// public ESDocumentServiceImpl(ElasticsearchClient elasticsearchClient, ElasticsearchAsyncClient elasticsearchAsyncClient) {
// this.elasticsearchClient = elasticsearchClient;
// this.elasticsearchAsyncClient = elasticsearchAsyncClient;
// }
//
// @Override
// public <T> IndexResponse createByFluentDSL(String idxName, String idxId, ESDocument document) throws Exception {
// Date date = new Date();
// LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();
// document.setCreator(loginUser.getNickname());
// document.setUpdater(loginUser.getNickname());
// document.setOrganId(loginUser.getOrganId());
// document.setCreateTime(simpleDateFormat.format(date));
// document.setUpdateTime(simpleDateFormat.format(date));
//
// if (CharSequenceUtil.isBlank(document.getId())) {
// document.setId(snowflake.nextIdStr());
// }
// return elasticsearchClient.index(idx -> idx
// .index(idxName)
// .id(document.getId())
// .document(document));
// }
//
// /**
// * BuilderPattern 方式创建文档
// *
// * @param idxName 索引名
// * @param idxId 索引id
// * @param document 文档对象
// */
// @Override
// public <T> IndexResponse createByBuilderPattern(String idxName, String idxId, ESDocument document) throws Exception {
// LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();
// Date date = new Date();
// document.setCreator(loginUser.getNickname());
// document.setUpdater(loginUser.getNickname());
// document.setOrganId(loginUser.getOrganId());
// document.setCreateTime(simpleDateFormat.format(date));
// document.setUpdateTime(simpleDateFormat.format(date));
// IndexRequest.Builder<Object> indexReqBuilder = new IndexRequest.Builder<>();
// indexReqBuilder.index(idxName);
// if (CharSequenceUtil.isBlank(idxId)) {
// idxId = snowflake.nextIdStr();
// }
// indexReqBuilder.id(idxId);
// indexReqBuilder.document(document);
// return elasticsearchClient.index(indexReqBuilder.build());
// }
//
// /**
// * json方式创建文档
// *
// * @param idxName 索引名
// * @param idxId 索引id
// * @param jsonContent json字符串
// */
// @Override
// public IndexResponse createByJson(String idxName, String idxId, String jsonContent) throws Exception {
// if (CharSequenceUtil.isBlank(idxId)) {
// idxId = snowflake.nextIdStr();
// }
// String finalIdxId = idxId;
// return elasticsearchClient.index(i -> i
// .index(idxName)
// .id(finalIdxId)
// .withJson(new StringReader(jsonContent))
// );
// }
//
// /**
// * 异步方式创建文档
// *
// * @param idxName 索引名
// * @param idxId 索引id
// * @param document 文档
// * @param action 操作
// */
// @Override
// public <T> void createAsync(String idxName, String idxId, T document, BiConsumer<IndexResponse, Throwable> action) {
// elasticsearchAsyncClient.index(idx -> idx
// .index(idxName)
// .id(idxId)
// .document(document)
// ).whenComplete(action);
// }
//
// /**
// * 批量方式创建文档
// *
// * @param idxName 索引名
// * @param documents 要增加的对象集合
// */
//
// @Override
// public <T> BulkResponse bulkCreate(String idxName, List<? extends ESDocument> documents) throws IOException, ElasticsearchException {
// BulkRequest.Builder br = new BulkRequest.Builder();
// LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();
// Date date = new Date();
// LocalDateTime now = LocalDateTime.now();
// documents.forEach(esDocument -> {
// if (CharSequenceUtil.isBlank(esDocument.getId())) {
// esDocument.setId(snowflake.nextIdStr());
// }
// esDocument.setCreator(loginUser.getNickname());
// esDocument.setCreateTime(simpleDateFormat.format(date));
// esDocument.setUpdater(loginUser.getNickname());
// esDocument.setOrganId(loginUser.getOrganId());
//
// esDocument.setUpdateTime(simpleDateFormat.format(date));
// br.operations(op -> op.index(idx -> idx
// .index(idxName)
// .id(esDocument.getId())
// .document(esDocument)));
// });
// return elasticsearchClient.bulk(br.build());
// }
//
//
//
//
// /**
// * 批量方式更新文档
// *
// * @param idxName 索引名
// * @param documents 要更新的对象集合
// */
// @Override
// public <T> BulkResponse bulkUpdate(String idxName, List<? extends ESDocument> documents) throws Exception {
// BulkRequest.Builder br = new BulkRequest.Builder();
// LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();
// Date date = new Date();
// documents.forEach(esDocument -> {
//
// esDocument.setCreator(loginUser.getNickname());
// esDocument.setUpdater(loginUser.getNickname());
// esDocument.setOrganId(loginUser.getOrganId());
// esDocument.setUpdateTime(simpleDateFormat.format(date));
//
// br.operations(op -> op.index(idx -> idx
// .index(idxName)
// .id(esDocument.getId())
// .document(esDocument)));
// });
// return elasticsearchClient.bulk(br.build());
// }
//
//
//
//
// /**
// * @param idxName 索引名称
// * @param docId 文档id
// * @param tClass 返回的类型
// * @param map 修改内容的map
// * Map<String, Object> map = new HashMap<>();
// * map.put("age", 35);
// * 把年龄改成35
// */
// @Override
// public <T> Result updateById(String idxName, String docId, Class<T> tClass, Map<String, Object> map) throws IOException, ElasticsearchException {
// UpdateResponse<T> response = elasticsearchClient.update(e -> e.index(idxName).id(docId).doc(map), tClass);
// return response.result();
// }
//
// /**
// * 文档id查询信息
// *
// * @param idxName 索引名
// * @param docId 文档id
// */
// @Override
// public <T> T getById(String idxName, String docId, Class<T> tClass) throws IOException {
// GetResponse<T> response = elasticsearchClient.get(g -> g
// .index(idxName)
// .id(docId),
// tClass);
// return response.found() ? response.source() : null;
// }
//
// /**
// * 根据索引名称和文档id查询ObjectNode
// *
// * @param idxName 索引名
// * @param docId 文档id
// */
// @Override
// public JSONObject getObjectNodeById(String idxName, String docId) throws IOException {
// GetResponse<JSONObject> response = elasticsearchClient.get(g -> g
// .index(idxName)
// .id(docId),
// JSONObject.class);
//
// return response.found() ? response.source() : null;
// }
//
// /**
// * 单条输出
// *
// * @param idxName 索引名
// * @param docId 文档id
// */
// @Override
// public Boolean deleteById(String idxName, String docId) throws IOException {
// DeleteResponse delete = elasticsearchClient.delete(d -> d
// .index(idxName)
// .id(docId));
// return delete.forcedRefresh();
// }
//
// /**
// * 批量删除
// *
// * @param idxName 索引名
// * @param docIds 要删除的文档id集合
// */
// @Override
// public BulkResponse bulkDeleteByIds(String idxName, List<String> docIds) throws Exception {
// BulkRequest.Builder br = new BulkRequest.Builder();
// // 将每一个对象都放入builder中
// docIds.forEach(id -> br
// .operations(op -> op
// .delete(d -> d
// .index(idxName)
// .id(id))));
// return elasticsearchClient.bulk(br.build());
// }
}
cf-framework/cf-spring-boot-starter-elasticsearch/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
-1
View File
@@ -1 +0,0 @@
com.cf.imes.framework.es.config.ChenfengElasticsearchAutoConfiguration