devbox
你好赵伟
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Java引入Elasticsearch实现增删改查_resthighlevelclient设置index.mapping.total_fields.li

作者：你好赵伟 | 2024-02-08 10:59:04

赞

踩

resthighlevelclient设置index.mapping.total_fields.limit

引入Elasticsearch的jar

  1. <dependency>
  2.   <groupId>org.elasticsearch.client</groupId>
  3.   <artifactId>elasticsearch-rest-high-level-client</artifactId>
  4.   <version>7.6.2</version>
  5.   <exclusions>
  6.     <exclusion>
  7.       <groupId>org.elasticsearch</groupId>
  8.       <artifactId>elasticsearch</artifactId>
  9.     </exclusion>
  10.   </exclusions>
  11. </dependency>
  12. <dependency>
  13.    <groupId>org.elasticsearch</groupId>
  14.    <artifactId>elasticsearch</artifactId>
  15.    <version>7.6.2</version>
  16. </dependency>
  17. <dependency>
  18.   <groupId>org.elasticsearch.plugin</groupId>
  19.   <artifactId>delete-by-query</artifactId>
  20.   <version>2.4.6</version>
  21. </dependency>

编写Elasticsearch基类

Es的基本信息的获取和封装

  1. package *.*.common.es.common;
  2.  
  3. import java.util.ArrayList;
  4. import java.util.Arrays;
  5. import java.util.List;
  6. import java.util.StringJoiner;
  7.  
  8. import org.apache.http.HttpHost;
  9. import org.slf4j.Logger;
  10. import org.slf4j.LoggerFactory;
  11.  
  12. import *.*.common.config.ConfigGroupFile;
  13. import *.*.common.config.ConfigService;
  14.  
  15. public class EsConfig {
  16. 	@SuppressWarnings("unused")
  17. 	private static final Logger logger = LoggerFactory.getLogger(EsConfig.class);
  18.  
  19. 	private static EsConfig config;
  20.  
  21. 	/**
  22. 	 * 节点列表
  23. 	 */
  24. 	private HttpHost[] hosts;
  25.  
  26. 	/**
  27. 	 * 用户名
  28. 	 */
  29. 	private String username;
  30.  
  31. 	/**
  32. 	 * 密码
  33. 	 */
  34. 	private String password;
  35.  
  36. 	// 调优参数
  37.  
  38. 	/**
  39. 	 * 连接超时
  40. 	 */
  41. 	private int connTimeout;
  42.  
  43. 	/**
  44. 	 * 读写超时
  45. 	 */
  46. 	private int socketTimeout;
  47.  
  48. 	/**
  49. 	 * 请求超时
  50. 	 */
  51. 	private int connRequestTimeout;
  52.  
  53. 	/**
  54. 	 * 允许建立多少个连接，超出以后所有的连接都会进入等待队列
  55. 	 */
  56. 	private int maxConnTotal;
  57.  
  58. 	/**
  59. 	 * 同一个 host:port 最多有几个活跃连接
  60. 	 */
  61. 	private int maxConnPerRoute;
  62.  
  63. 	/**
  64. 	 * 返回 ES 配置信息。
  65. 	 *
  66. 	 * @return ES配置
  67. 	 */
  68. 	public static EsConfig getConfig() {
  69. 		if (config == null) {
  70. 			config = doCreate();
  71. 		}
  72. 		return config;
  73. 	}
  74.  
  75. 	/*
  76. 	 * 创建配置实例。
  77. 	 */
  78. 	private static EsConfig doCreate() {
  79. 		EsConfig config = new EsConfig();
  80. 		config.hosts = buildHosts();
  81. 		
  82. 		config.username = ConfigService.getPersionString("elasticsearch.http.username", ConfigGroupFile.getEsConfig());
  83. 		config.password = ConfigService.getPersionString("elasticsearch.http.password", ConfigGroupFile.getEsConfig());
  84.  
  85. 		// 超时参数
  86. 		config.connTimeout = ConfigService.getInt("crab.elasticsearch.http.conn_timeout", 5000);
  87. 		config.socketTimeout = ConfigService.getInt("crab.elasticsearch.http.socket_timeout", 30000);
  88. 		config.connRequestTimeout = ConfigService.getInt("crab.elasticsearch.http.conn_request_timeout", 5000);
  89.  
  90. 		// 连接数
  91. 		config.maxConnTotal = ConfigService.getInt("crab.elasticsearch.http.max_conn_total", 400);
  92. 		config.maxConnPerRoute = ConfigService.getInt("crab.elasticsearch.http.max_conn_per_route", 200);
  93. 		return config;
  94. 	}
  95.  
  96. 	/*
  97. 	 * 装装主机信息。
  98. 	 */
  99. 	private static HttpHost[] buildHosts() {
  100. 		String hosts = ConfigService.getPersionString("elasticsearch.http.hosts", ConfigGroupFile.getEsConfig());
  101. 		if (hosts == null) {
  102. 			return new HttpHost[0];
  103. 		}
  104.  
  105. 		List<HttpHost> hostList = new ArrayList<>();
  106. 		for (String host : hosts.split(",")) {
  107. 			String[] hostPort = host.split(":");
  108. 			if (hostPort.length == 1) {
  109. 				hostList.add(new HttpHost(hostPort[0])); // 10.1.1.1
  110. 			} else if (hostPort.length == 2) {
  111. 				hostList.add(new HttpHost(hostPort[0], Integer.parseInt(hostPort[1])));  // 10.1.1.1:9200
  112. 			} else if (hostPort.length == 3) {
  113. 				hostList.add(new HttpHost(hostPort[1], Integer.parseInt(hostPort[2]), hostPort[0])); // https:10.1.1.1:9200
  114. 			}
  115. 		}
  116. 		return hostList.toArray(new HttpHost[hostList.size()]);
  117. 	}
  118.  
  119. 	/**
  120. 	 * 获取节点列表。
  121. 	 *
  122. 	 * @return 节点名
  123. 	 */
  124. 	public static HttpHost[] getHosts() {
  125. 		return getConfig().hosts;
  126. 	}
  127.  
  128. 	/**
  129. 	 * 获取用户名
  130. 	 *
  131. 	 * @return 用户名
  132. 	 */
  133. 	public static String getUsername() {
  134. 		return getConfig().username;
  135. 	}
  136.  
  137. 	/**
  138. 	 * 获取密码。
  139. 	 *
  140. 	 * @return 密码
  141. 	 */
  142. 	public static String getPassword() {
  143. 		return getConfig().password;
  144. 	}
  145.  
  146. 	public static int getConnTimeout() {
  147. 		return getConfig().connTimeout;
  148. 	}
  149.  
  150. 	public static int getSocketTimeout() {
  151. 		return getConfig().socketTimeout;
  152. 	}
  153.  
  154. 	public static int getConnRequestTimeout() {
  155. 		return getConfig().connRequestTimeout;
  156. 	}
  157.  
  158. 	public static int getMaxConnTotal() {
  159. 		return getConfig().maxConnTotal;
  160. 	}
  161.  
  162. 	public static int getMaxConnPerRoute() {
  163. 		return getConfig().maxConnPerRoute;
  164. 	}
  165.  
  166. 	@Override
  167. 	public String toString() {
  168. 		StringJoiner sj = new StringJoiner(", ", "{", "}");
  169. 		sj.add("hosts:" + Arrays.toString(hosts));
  170. 		sj.add("username:" + username);
  171. 		sj.add("connTimeout:" + connTimeout);
  172. 		sj.add("socketTimeout:" + socketTimeout);
  173. 		sj.add("connRequestTimeout:" + connRequestTimeout);
  174. 		sj.add("maxConnTotal:" + maxConnTotal);
  175. 		sj.add("maxConnPerRoute:" + maxConnPerRoute);
  176. 		return sj.toString();
  177. 	}
  178. }
  1. package *.*.common.es.common;
  2.  
  3. import java.io.IOException;
  4.  
  5. import org.apache.http.auth.AuthScope;
  6. import org.apache.http.auth.UsernamePasswordCredentials;
  7. import org.apache.http.impl.client.BasicCredentialsProvider;
  8. import org.elasticsearch.client.Node;
  9. import org.elasticsearch.client.RestClient;
  10. import org.elasticsearch.client.RestClientBuilder;
  11. import org.elasticsearch.client.RestHighLevelClient;
  12. import org.slf4j.Logger;
  13. import org.slf4j.LoggerFactory;
  14.  
  15. import *.*.api.error.ServerException;
  16.  
  17. public class EsClients {
  18. 	private static final Logger logger = LoggerFactory.getLogger(EsClients.class);
  19.  
  20. 	private RestHighLevelClient client;
  21.  
  22. 	// 单例
  23. 	private static class Singleton {
  24. 		public static final EsClients INSTANCE = new EsClients();
  25. 		private Singleton() {}
  26. 	}
  27.  
  28. 	private EsClients() {
  29. 		client = new RestHighLevelClient(createClientBuilder());
  30. 	}
  31. 	
  32. 	public void reBuildEsClients() {
  33. 		if (client != null) {
  34. 			try {
  35. 				client.close();
  36. 			} catch (Exception e) {
  37. 				logger.debug(e.getMessage(), e);
  38. 			}
  39. 		}
  40. 		client = new RestHighLevelClient(createClientBuilder());
  41. 	}
  42. 	
  43. 	public static RestHighLevelClient getClient(){
  44. 		return Singleton.INSTANCE.client;
  45. 	}
  46.  
  47. 	/**
  48. 	 * ES 执行操作请求的回调类。
  49. 	 */
  50. 	@FunctionalInterface
  51. 	public interface ClientRequest<T> {
  52. 		T doRequest(RestHighLevelClient client) throws IOException;
  53. 	}
  54. 	
  55. 	/**
  56. 	 * 执行 ES 的操作请求。
  57. 	 * 
  58. 	 * @param request 操作的请求
  59. 	 * @return 执行结果
  60. 	 */
  61. 	public static <T> T executeRequest(ClientRequest<T> request) {
  62. 		try {
  63. 			return request.doRequest(Singleton.INSTANCE.client);
  64. 		} catch (Exception e) {
  65. 			logger.error(e.getMessage());
  66. 			String message = e.getMessage();
  67. 			if (message != null && message.contains("I/O reactor status: STOPPED")) {
  68. 				return reQuery(request);
  69. 			}
  70. 			throw new ServerException("", e);
  71. 		}
  72. 	}
  73.  
  74. 	private static <T> T reQuery(ClientRequest<T> request) {
  75. 		logger.warn("we encounter a problem I/O reactor status: STOPPED");
  76.  
  77. 		Singleton.INSTANCE.reBuildEsClients();
  78. 		try {
  79. 			return request.doRequest(Singleton.INSTANCE.client);
  80. 		} catch (Exception e) {
  81. 			throw new ServerException("new query", e);
  82. 		}
  83. 	}
  84.  
  85. 	/*
  86. 	 * 构造 builder 。
  87. 	 */
  88. 	private static RestClientBuilder createClientBuilder() {
  89. 		RestClientBuilder builder = RestClient.builder(EsConfig.getHosts());
  90.  
  91. 		// 连接超时时间
  92. 		builder.setRequestConfigCallback(requestConfigBuilder -> {
  93. 				requestConfigBuilder.setConnectTimeout(EsConfig.getConnTimeout());
  94. 				requestConfigBuilder.setSocketTimeout(EsConfig.getSocketTimeout());
  95. 				requestConfigBuilder.setConnectionRequestTimeout(EsConfig.getConnRequestTimeout());
  96. 				return requestConfigBuilder;
  97. 		});
  98.  
  99. 		// 连接信息
  100. 		builder.setHttpClientConfigCallback(httpClientBuilder -> {
  101. 			httpClientBuilder.setMaxConnTotal(EsConfig.getMaxConnTotal());
  102. 			httpClientBuilder.setMaxConnPerRoute(EsConfig.getMaxConnPerRoute());
  103.  
  104. 			final BasicCredentialsProvider provider = new BasicCredentialsProvider();
  105. 			provider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(EsConfig.getUsername(), EsConfig.getPassword()));
  106. 			return httpClientBuilder.setDefaultCredentialsProvider(provider);
  107. 		});
  108.  
  109. 		builder.setFailureListener(new RestClient.FailureListener() {
  110. 			@Override
  111. 			public void onFailure(Node node) {
  112. 				logger.debug("failed to execute on node: {}", node);
  113. 			}
  114. 		});
  115. 		return builder;
  116. 	}
  117. }

ES索引的Setting统一设置 ，包括分片信息，数据集，map的字段个数，是否开启大消息忽略

  1. package *.*.common.es.entity;
  2.  
  3. import *.*.common.config.ConfigService;
  4.  
  5. /**
  6.  * 
  7.  * @author zhanglb 
  8.  * Create at 2022年2月17日 下午5:37:26
  9.  */
  10. public class EsSettingEntity {
  11.  
  12. 	/**
  13. 	 * 分片信息
  14. 	 */
  15. 	private int indexShard = 3;
  16. 	/**
  17. 	 * 数据集
  18. 	 */
  19. 	private int indexReplicas = 2;
  20. 	/**
  21. 	 * map的字段个数
  22. 	 */
  23. 	private int fieldLength = 5000;//默认5000
  24. 	/**
  25. 	 * 是否开启大小写忽略
  26. 	 */
  27. 	private boolean openLowercase = ConfigService.getInstance().isOpenLowercase();
  28. 	
  29. 	public EsSettingEntity() {
  30. 		
  31. 	}
  32. 	
  33. 	public EsSettingEntity(int indexShard, int indexReplicas, boolean openLowercase) {
  34. 		this.indexShard = indexShard;
  35. 		this.indexReplicas =  indexReplicas;
  36. 		this.openLowercase = openLowercase;
  37. 	}
  38.  
  39. 	/**
  40. 	 * @return indexShard
  41. 	 */
  42. 	public int getIndexShard() {
  43. 		return indexShard;
  44. 	}
  45.  
  46. 	/**
  47. 	 * @param indexShard
  48. 	 */
  49. 	public void setIndexShard(int indexShard) {
  50. 		this.indexShard = indexShard;
  51. 	}
  52.  
  53. 	/**
  54. 	 * @return indexReplicas
  55. 	 */
  56. 	public int getIndexReplicas() {
  57. 		return indexReplicas;
  58. 	}
  59.  
  60. 	/**
  61. 	 * @param indexReplicas
  62. 	 */
  63. 	public void setIndexReplicas(int indexReplicas) {
  64. 		this.indexReplicas = indexReplicas;
  65. 	}
  66.  
  67. 	/**
  68. 	 * @return fieldLength
  69. 	 */
  70. 	public int getFieldLength() {
  71. 		return fieldLength;
  72. 	}
  73.  
  74. 	/**
  75. 	 * @param fieldLength
  76. 	 */
  77. 	public void setFieldLength(int fieldLength) {
  78. 		this.fieldLength = fieldLength;
  79. 	}
  80.  
  81. 	/**
  82. 	 * @return openLowercase
  83. 	 */
  84. 	public boolean isOpenLowercase() {
  85. 		return openLowercase;
  86. 	}
  87.  
  88. 	/**
  89. 	 * @param openLowercase
  90. 	 */
  91. 	public void setOpenLowercase(boolean openLowercase) {
  92. 		this.openLowercase = openLowercase;
  93. 	}
  94. 	
  95. 	
  96. }

ES Setting的初始化公共类 

  1. package *.*.common.es.support;
  2.  
  3. import java.util.ArrayList;
  4. import java.util.List;
  5.  
  6. import org.elasticsearch.common.settings.Settings;
  7.  
  8. import *.*.common.es.entity.EsSettingEntity;
  9.  
  10. /**
  11.  * 
  12.  * @author zhanglb 
  13.  * Create at 2022年2月17日 下午5:14:23
  14.  */
  15. public class EsSetting {
  16. 	
  17. 	public EsSetting() {
  18. 		
  19. 	}
  20. 	
  21. 	/**
  22. 	 * 初始化setting
  23. 	 * @param entity
  24. 	 * @return
  25. 	 */
  26. 	public static Settings.Builder initSetting(EsSettingEntity entity) {
  27. 		Settings.Builder setting = Settings.builder().put("index.number_of_shards", entity.getIndexShard()).put("index.number_of_replicas", entity.getIndexReplicas())
  28. 				.put("index.mapping.total_fields.limit", entity.getFieldLength());
  29. 		if (entity.isOpenLowercase()) {
  30. 			setLowercase(setting);
  31. 		}
  32. 		return setting;
  33. 	}
  34. 	
  35. 	/**
  36. 	 * 设置大小写不敏感
  37. 	 * @param setting
  38. 	 * @return
  39. 	 */
  40. 	public static Settings.Builder setLowercase(Settings.Builder setting) {
  41. 		List<String> filter = new ArrayList<>();
  42. 		filter.add("lowercase");
  43. 		setting.put("index.analysis.normalizer.lowercase.type",EsActivityMapping.FIELD_TYPE_CUSTOM).putList("index.analysis.normalizer.lowercase.filter", filter);
  44. 		return setting;
  45. 	}
  46. }

ES查询 数据的基础类

  1. package *.*.common.es.support;
  2.  
  3. import java.util.List;
  4. import java.util.concurrent.TimeUnit;
  5.  
  6. import org.elasticsearch.ElasticsearchException;
  7. import org.elasticsearch.action.search.SearchRequest;
  8. import org.elasticsearch.action.search.SearchResponse;
  9. import org.elasticsearch.client.RequestOptions;
  10. import org.elasticsearch.client.core.CountRequest;
  11. import org.elasticsearch.common.unit.TimeValue;
  12. import org.elasticsearch.search.builder.SearchSourceBuilder;
  13. import org.slf4j.Logger;
  14. import org.slf4j.LoggerFactory;
  15.  
  16. import *.*.api.error.ClientException;
  17. import *.*.api.error.ServerException;
  18. import *.*.common.config.ConfigService;
  19. import *.*.common.es.common.EsClients;
  20. import *.*.i18n.source.error.I18nErrorMessages;
  21.  
  22. /**
  23.  * 
  24.  * 从Es中查询数据工具类
  25.  * 
  26.  * @author zhanglb
  27.  *
  28.  */
  29. public class EsDataReader {
  30. 	private static final Logger logger = LoggerFactory.getLogger(EsDataReader.class);
  31. 	private static boolean UNIT_TEST = ConfigService.getBoolean("crab.unit.test.open", false);
  32.  
  33. 	public EsDataReader() {
  34. 	}
  35.  
  36. 	/**
  37. 	 * 执行 count 请求。
  38. 	 *
  39. 	 * @param queryParams 过滤条件
  40. 	 * @return 累计数
  41. 	 */
  42. 	public static long count(List<String> index, SearchSourceBuilder sourceBuilder) {
  43. 		if (UNIT_TEST) {
  44. 			logger.debug("es count has been suppressed for unit test.");
  45. 			return 0l;
  46. 		}
  47. 		
  48. 		try {
  49. 			CountRequest countReq = new CountRequest(index.toArray(new String[index.size()]), sourceBuilder);
  50. 			return EsClients.executeRequest(client -> client.count(countReq, RequestOptions.DEFAULT).getCount());
  51. 		} catch (ElasticsearchException e) {
  52. 			logger.error("", e);
  53. 			throw new ServerException("Failed to execute count: " + e.getMessage(), null);
  54. 		}
  55. 	}
  56.  
  57. 	/**
  58. 	 * 根据指定的查询条件，执行数据操作。
  59. 	 *
  60. 	 * @param queryParams 查询条件对象
  61. 	 * @return 查询结果
  62. 	 */
  63. 	public static SearchResponse query(List<String> index, SearchSourceBuilder sourceBuilder) {
  64. 		if (UNIT_TEST) {
  65. 			logger.debug("es query has been suppressed for unit test.");
  66. 			return null;
  67. 		}
  68. 		
  69. 		SearchResponse response;
  70. 		SearchRequest request;
  71. 		try {
  72. 			// 设置超时时间
  73. 			sourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS));
  74. 			request = new SearchRequest(index.toArray(new String[index.size()]), sourceBuilder);
  75. 			if (logger.isDebugEnabled()) {
  76. 				logger.debug("query with condition: {}", request.source().query());
  77. 			}
  78. 			response = EsClients.executeRequest(client -> client.search(request, RequestOptions.DEFAULT));
  79. 		} catch (ElasticsearchException ee) {
  80. 			Throwable cause = ee.getCause();
  81. 			if (cause != null && cause.getMessage().contains("Result window is too large")) {
  82. 				logger.debug("", ee);
  83. 				throw ClientException.create(I18nErrorMessages.ERROR_QUERY_FAIELD_NEED_LIMIT);
  84. 			} else {
  85. 				logger.error("", ee);
  86. 				throw new ServerException("Failed to execute query: " + ee.getMessage(), null);
  87. 			}
  88. 		}
  89.  
  90. 		if (response.isTimedOut()) {
  91. 			throw new ServerException("Query with timeout", null);
  92. 		}
  93. 		return response;
  94. 	}
  95. }

ES的写入数据的基础类 

  1. package *.*.common.es.support;
  2.  
  3. import java.util.List;
  4. import java.util.Map;
  5.  
  6. import org.apache.commons.lang3.StringUtils;
  7. import org.elasticsearch.action.ActionListener;
  8. import org.elasticsearch.action.bulk.BulkItemResponse;
  9. import org.elasticsearch.action.bulk.BulkRequest;
  10. import org.elasticsearch.action.bulk.BulkResponse;
  11. import org.elasticsearch.action.delete.DeleteRequest;
  12. import org.elasticsearch.action.index.IndexRequest;
  13. import org.elasticsearch.action.update.UpdateRequest;
  14. import org.elasticsearch.client.RequestOptions;
  15. import org.slf4j.Logger;
  16. import org.slf4j.LoggerFactory;
  17.  
  18. import *.*.common.config.ConfigService;
  19. import *.*.common.es.common.EsClients;
  20. import *.*.common.es.common.EsIndexUtils;
  21. import *.*.common.util.ConstantFields;
  22.  
  23. /**
  24.  * 把动态记录写入到es中工具类
  25.  * 
  26.  * @author zhanglb
  27.  *
  28.  */
  29. public class EsDataWriter {
  30. 	private static final Logger logger = LoggerFactory.getLogger(EsDataWriter.class);
  31. 	private static boolean UNIT_TEST = ConfigService.getBoolean("crab.unit.test.open", false);
  32.  
  33. 	public EsDataWriter() {
  34. 	}
  35.  
  36. 	public static boolean sync(String index, List<Map<String, Object>> map) {
  37. 		if (UNIT_TEST) {
  38. 			logger.debug("es sync has been suppressed for unit test.");
  39. 			return true;
  40. 		}
  41.  
  42. 		if (StringUtils.isBlank(index) || map == null)
  43. 			return false;
  44.  
  45. 		boolean storeIndex = false;
  46. 		if (!map.isEmpty()) {
  47. 			if (!EsIndexUtils.exists(index)) {
  48. 				logger.info("create index:" + index);
  49. 				EsIndexUtils.createIndex(index, null, null, EsActivityMapping.buildActivityMapping());
  50. 				storeIndex = true;
  51. 			}
  52. 			EsDataWriter.createObjectsAsync(index, map);
  53. 		}
  54. 		return storeIndex;
  55. 	}
  56.  
  57. 	/**
  58. 	 * 写操作类型。
  59. 	 */
  60. 	public enum WriteAction {
  61. 		ADD, UPDATE, UPSERT, DELETE;
  62. 	}
  63.  
  64. 	/**
  65. 	 * 异步创建资源对象。
  66. 	 *
  67. 	 * @param objects 待新建的资源对象
  68. 	 */
  69. 	public static void createObjectsAsync(String index, List<Map<String, Object>> objects) {
  70. 		if (UNIT_TEST) {
  71. 			logger.debug("es create object has been suppressed for unit test.");
  72. 			return;
  73. 		}
  74. 		
  75. 		writerObjectsAsync(index, objects, WriteAction.ADD);
  76. 	}
  77.  
  78. 	/**
  79. 	 * 异步更新资源对象。
  80. 	 *
  81. 	 * @param objects 待更新的资源对象（只需提供 ID 以及 变更的字段）
  82. 	 */
  83. 	public static void updateObjectsAsync(String index, List<Map<String, Object>> objects) {
  84. 		if (UNIT_TEST) {
  85. 			logger.debug("es update object has been suppressed for unit test.");
  86. 			return;
  87. 		}
  88. 		
  89. 		writerObjectsAsync(index, objects, WriteAction.UPDATE);
  90. 	}
  91.  
  92. 	/**
  93. 	 * 异步保存（根据 ID 判断新建或更新）资源对象。
  94. 	 *
  95. 	 * @param objects 待保存的资源对象
  96. 	 */
  97. 	public static void saveObjectsAsync(String index, List<Map<String, Object>> objects) {
  98. 		if (UNIT_TEST) {
  99. 			logger.debug("es save object has been suppressed for unit test.");
  100. 			return;
  101. 		}
  102. 		
  103. 		writerObjectsAsync(index, objects, WriteAction.UPSERT);
  104. 	}
  105.  
  106. 	/**
  107. 	 * 异步保存（根据 ID 判断新建或更新）资源对象。
  108. 	 *
  109. 	 * @param objects 待保存的资源对象
  110. 	 */
  111. 	public static void delObjectsAsync(String index, List<Map<String, Object>> objects) {
  112. 		if (UNIT_TEST) {
  113. 			logger.debug("es delete object has been suppressed for unit test.");
  114. 			return;
  115. 		}
  116. 		
  117. 		writerObjectsAsync(index, objects, WriteAction.DELETE);
  118. 	}
  119.  
  120. 	/**
  121. 	 * 以异步的形式批量写资源对象。
  122. 	 *
  123. 	 * @param writeRequests 写请求
  124. 	 */
  125. 	public static void writerObjectsAsync(String index, List<Map<String, Object>> writeRequests, WriteAction action) {
  126. 		if (UNIT_TEST) {
  127. 			logger.debug("es write object has been suppressed for unit test.");
  128. 			return;
  129. 		}
  130. 		
  131. 		if (writeRequests == null || writeRequests.isEmpty())
  132. 			return;
  133.  
  134. 		final BulkRequest request = new BulkRequest();
  135. 		for (Map<String, Object> writeRequest : writeRequests) {
  136. 			switch (action) {
  137. 			case ADD:
  138. 				request.add(buildCreateRequest(index, writeRequest));
  139. 				break;
  140. 			case UPDATE:
  141. 				UpdateRequest reqUp = buildUpdateRequest(index, writeRequest, true);
  142. 				if (reqUp == null)
  143. 					break;
  144. 				request.add(reqUp);
  145. 				break;
  146. 			case UPSERT:
  147. 				UpdateRequest reqU = buildUpdateRequest(index, writeRequest, true);
  148. 				if (reqU == null)
  149. 					break;
  150. 				request.add(reqU);
  151. 				break;
  152. 			case DELETE:
  153. 				DeleteRequest reqD = buildDeleteRequest(index, writeRequest.get(ConstantFields.FIELD_ID));
  154. 				if (reqD == null)
  155. 					break;
  156. 				request.add(reqD);
  157. 				break;
  158. 			}
  159. 		}
  160.  
  161. 		EsClients.executeRequest(client -> {
  162. 			client.bulkAsync(request, RequestOptions.DEFAULT, new BulkWriteCallback());
  163. 			return null;
  164. 		});
  165. 	}
  166.  
  167. 	/*
  168. 	 * 新建资源请求。
  169. 	 *
  170. 	 * @return 请求对象
  171. 	 */
  172. 	private static IndexRequest buildCreateRequest(String index, Map<String, Object> map) {
  173. 		IndexRequest create = new IndexRequest(index);
  174. 		if (map.containsKey(ConstantFields.FIELD_ID)) {
  175. 			create.id(map.get(ConstantFields.FIELD_ID).toString());
  176. 		}
  177. 		create.source(map);
  178. 		return create;
  179. 	}
  180.  
  181. 	/*
  182. 	 * 更新资源请求。
  183. 	 *
  184. 	 * @param upsert 如果不存在是否新建
  185. 	 * 
  186. 	 * @return 请求对象
  187. 	 */
  188. 	private static UpdateRequest buildUpdateRequest(String index, Map<String, Object> map, boolean upsert) {
  189. 		if (!map.containsKey(ConstantFields.FIELD_ID)) {
  190. 			return null;
  191. 		}
  192. 		UpdateRequest update = new UpdateRequest(index, map.get(ConstantFields.FIELD_ID).toString()).retryOnConflict(3);
  193. 		update.docAsUpsert(upsert);
  194. 		update.doc(map);
  195. 		return update;
  196. 	}
  197.  
  198. 	/*
  199. 	 * 删除资源请求。
  200. 	 *
  201. 	 * @return 请求对象
  202. 	 */
  203. 	private static DeleteRequest buildDeleteRequest(String index, Object id) {
  204. 		if (id == null)
  205. 			return null;
  206.  
  207. 		return new DeleteRequest(index, id.toString());
  208. 	}
  209.  
  210. 	/*
  211. 	 * 回调。
  212. 	 */
  213. 	private static class BulkWriteCallback implements ActionListener<BulkResponse> {
  214.  
  215. 		@Override
  216. 		public void onFailure(Exception e) {
  217. 			logger.error("Write data with error", e);
  218. 		}
  219.  
  220. 		@Override
  221. 		public void onResponse(BulkResponse bulkItemResponses) {
  222. 			if (bulkItemResponses.hasFailures()) {
  223. 				processError(bulkItemResponses.getItems());
  224. 			} else {
  225. 				logger.debug("Write data successfully, count: {}", bulkItemResponses.getItems().length);
  226. 			}
  227. 		}
  228.  
  229. 		private void processError(BulkItemResponse[] responses) {
  230. 			for (BulkItemResponse response : responses) {
  231. 				if (!response.isFailed() || StringUtils.contains(response.getFailureMessage(), "document_missing_exception")) {
  232. 					continue;
  233. 				}
  234. 				logger.error("Write data with error: {}", response.getFailureMessage());
  235. 			}
  236. 		}
  237. 	}
  238. }

ES创建索引，修改别名等公共方法类 

  1. package *.*.common.es.common;
  2.  
  3. import java.util.Collections;
  4. import java.util.List;
  5. import java.util.Set;
  6.  
  7. import org.apache.commons.lang3.StringUtils;
  8. import org.elasticsearch.ElasticsearchException;
  9. import org.elasticsearch.action.admin.indices.alias.Alias;
  10. import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
  11. import org.elasticsearch.action.admin.indices.alias.get.GetAliasesRequest;
  12. import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
  13. import org.elasticsearch.action.support.master.AcknowledgedResponse;
  14. import org.elasticsearch.client.GetAliasesResponse;
  15. import org.elasticsearch.client.RequestOptions;
  16. import org.elasticsearch.client.indices.CreateIndexRequest;
  17. import org.elasticsearch.client.indices.CreateIndexResponse;
  18. import org.elasticsearch.client.indices.GetIndexRequest;
  19. import org.elasticsearch.client.indices.PutIndexTemplateRequest;
  20. import org.elasticsearch.common.settings.Settings;
  21. import org.elasticsearch.common.xcontent.XContentBuilder;
  22. import org.elasticsearch.rest.RestStatus;
  23. import org.slf4j.Logger;
  24. import org.slf4j.LoggerFactory;
  25.  
  26. import *.*.common.config.ConfigService;
  27. import *.*.common.es.entity.EsSettingEntity;
  28. import *.*.common.es.support.EsSetting;
  29. import *.*.common.util.ObjectUtils;
  30. import *.*.api.error.ServerException;
  31.  
  32. public class EsIndexUtils {
  33. 	private static final Logger logger = LoggerFactory.getLogger(EsIndexUtils.class);
  34. 	private static boolean UNIT_TEST = ConfigService.getBoolean("crab.unit.test.open", false);
  35. 	// 是否删除旧索引
  36. 	private static boolean deleteOldIndex = ConfigService.getBoolean("crab.es.delete.old.index", true);
  37. 	
  38.  
  39. 	private static final String INDEX_PREX = "*_activity_";
  40.  
  41. 	private EsIndexUtils() {
  42. 	}
  43.  
  44. 	/**
  45. 	 * 创建指定的 index。
  46. 	 *
  47. 	 * @param idxName 索引名
  48. 	 * @param aliasName 索引的别名
  49. 	 * @param mappingBuilder 索引的映射配置
  50. 	 */
  51. 	public static void createIndex(String idxName, String aliasName, Settings.Builder setting, XContentBuilder mappingBuilder) {
  52. 		if (UNIT_TEST) {
  53. 			logger.debug("es create index has been suppressed for unit test.");
  54. 			return;
  55. 		}
  56.  
  57. 		//当setting为null时，初始化默认的setting
  58. 		if (setting == null) {
  59. 			setting = EsSetting.initSetting(new EsSettingEntity());
  60. 		}
  61. 		
  62. 		CreateIndexRequest request = new CreateIndexRequest(idxName);
  63. 		request.settings(setting);
  64.  
  65. 		if (aliasName != null) {
  66. 			request.alias(new Alias(aliasName));
  67. 		}
  68.  
  69. 		if (mappingBuilder != null) {
  70. 			request.mapping(mappingBuilder);
  71. 		}
  72.  
  73. 		CreateIndexResponse response = EsClients.executeRequest(client -> client.indices().create(request, RequestOptions.DEFAULT));
  74. 		if (!response.isAcknowledged()) {
  75. 			logger.error("Failed to create index '{}' with alias: {}", idxName, aliasName);
  76. 		}
  77. 	}
  78.  
  79. 	/**
  80. 	 * 创建指定的 index。
  81. 	 *
  82. 	 * @param idxName 索引名
  83. 	 */
  84. 	public static void createIndexByTemplate(String idxName) {
  85. 		CreateIndexRequest request = new CreateIndexRequest(idxName);
  86. 		CreateIndexResponse response = EsClients.executeRequest(client -> client.indices().create(request,
  87. 				RequestOptions.DEFAULT));
  88.  
  89. 		if (!response.isAcknowledged()) {
  90. 			logger.error("Failed to create index '{}'", idxName);
  91. 		}
  92. 	}
  93.  
  94. 	/**
  95. 	 * 迁移 index 的别名到新的。
  96. 	 *
  97. 	 * @param indexAlias 索引别名
  98. 	 * @param oldIndexName 旧索引
  99. 	 * @param newIndexName 新索引
  100. 	 */
  101. 	public static void aliases(String indexAlias, String oldIndexName, String newIndexName) {
  102. 		if (UNIT_TEST) {
  103. 			logger.debug("es aliases name has been suppressed for unit test.");
  104. 			return;
  105. 		}
  106.  
  107. 		IndicesAliasesRequest request = new IndicesAliasesRequest();
  108.  
  109. 		// 先移除，再添加
  110. 		request.addAliasAction(
  111. 				new IndicesAliasesRequest.AliasActions(IndicesAliasesRequest.AliasActions.Type.REMOVE).index(oldIndexName).alias(indexAlias));
  112. 		request.addAliasAction(
  113. 				new IndicesAliasesRequest.AliasActions(IndicesAliasesRequest.AliasActions.Type.ADD).index(newIndexName).alias(indexAlias));
  114.  
  115. 		AcknowledgedResponse response = EsClients.executeRequest(client -> client.indices().updateAliases(request, RequestOptions.DEFAULT));
  116. 		if (!response.isAcknowledged()) {
  117. 			logger.error("Failed to aliases index: {}", indexAlias);
  118. 		}
  119. 	}
  120. 	
  121. 	/**
  122. 	 * 设置索引别名。
  123. 	 *
  124. 	 * @param indexAlias 索引别名
  125. 	 * @param newIndexName 新索引
  126. 	 * @param dropOldIndeces 被删除的别名所关联的索引
  127. 	 */
  128. 	public static Set<String> aliases(String indexAlias, String newIndexName, boolean dropOldIndeces) {
  129. 		Set<String> oldIndices = Collections.emptySet();
  130. 		if (dropOldIndeces) {
  131. 			oldIndices = getIndexByAlias(indexAlias);
  132. 		}
  133. 		if (!oldIndices.isEmpty()) {
  134. 			oldIndices.remove(newIndexName); // 保留最新创建的索引
  135. 		}
  136. 		
  137. 		IndicesAliasesRequest request = new IndicesAliasesRequest();
  138.  
  139. 		// 关联新索引
  140. 		request.addAliasAction(new IndicesAliasesRequest.AliasActions(IndicesAliasesRequest.AliasActions.Type.ADD)
  141. 				.index(newIndexName).alias(indexAlias).writeIndex(true)); // 将新增的索引设置为可写
  142. 		
  143. 		// 删除指定索引
  144. 		if (!oldIndices.isEmpty()) {
  145. 			request.addAliasAction(new IndicesAliasesRequest.AliasActions(IndicesAliasesRequest.AliasActions.Type.REMOVE_INDEX)
  146. 					.indices(oldIndices.toArray(new String[oldIndices.size()])));
  147. 		}
  148.  
  149. 		AcknowledgedResponse response = EsClients.executeRequest(client -> client.indices().updateAliases(request, RequestOptions.DEFAULT));
  150. 		if (!response.isAcknowledged()) {
  151. 			logger.error("Failed to aliases index: {}", indexAlias);
  152. 			return Collections.emptySet();
  153. 		}
  154. 		// 删除旧索引
  155. 		deleteOldIndexes(oldIndices);
  156. 		
  157. 		return oldIndices;
  158. 	}
  159. 	
  160. 	private static void deleteOldIndexes(Set<String> oldIndexes) {
  161. 		if (ObjectUtils.isNullOrEmpty(oldIndexes) || !deleteOldIndex) {
  162. 			return;
  163. 		}
  164. 		for (String oldIndex : oldIndexes) {
  165. 			dropIndex(oldIndex, true);
  166. 		}
  167. 	}
  168.  
  169. 	/**
  170. 	 * 返回指定别名所关联的所有索引。
  171. 	 * 
  172. 	 * @param indexAlias 索引别名
  173. 	 * @return 别名所关联的索引列表
  174. 	 */
  175. 	public static Set<String> getIndexByAlias(String indexAlias) {
  176. 		GetAliasesRequest getAliasReq = new GetAliasesRequest();
  177. 		getAliasReq.aliases(indexAlias);
  178. 		GetAliasesResponse getAliasRsp = EsClients.executeRequest(client -> client.indices().getAlias(getAliasReq, RequestOptions.DEFAULT));
  179. 		return getAliasRsp.getAliases().keySet();
  180. 	}
  181.  
  182.  
  183. 	/**
  184. 	 * 判断指定的 index 是否存在。
  185. 	 * 
  186. 	 * @param indexName 索引名
  187. 	 * @return 如果存在返回 true，否则返回 false
  188. 	 */
  189. 	public static boolean exists(String indexName) {
  190. 		if (UNIT_TEST) {
  191. 			logger.debug("es check index has been suppressed for unit test.");
  192. 			return false;
  193. 		}
  194.  
  195. 		GetIndexRequest getRequest = new GetIndexRequest(indexName);
  196. 		return EsClients.executeRequest(client -> client.indices().exists(getRequest, RequestOptions.DEFAULT));
  197. 	}
  198.  
  199. 	/**
  200. 	 * 删除指定的 index。
  201. 	 *
  202. 	 * @param indexName 索引名
  203. 	 */
  204. 	public static void dropIndex(String indexName, boolean ignoreIfNotExists) {
  205. 		if (UNIT_TEST) {
  206. 			logger.debug("es drop index has been suppressed for unit test.");
  207. 			return;
  208. 		}
  209.  
  210. 		try {
  211. 			DeleteIndexRequest deleteRequest = new DeleteIndexRequest(indexName);
  212. 			EsClients.executeRequest(client -> client.indices().delete(deleteRequest, RequestOptions.DEFAULT));
  213. 		} catch (ElasticsearchException e) {
  214. 			if (ignoreIfNotExists && e.status() == RestStatus.NOT_FOUND) {
  215. 				return; // ignore
  216. 			}
  217. 			logger.error("Failed to drop index: " + indexName, e);
  218. 			throw new ServerException("Failed to drop index: " + indexName + ", reason: " + e.getMessage(), null);
  219. 		}
  220. 	}
  221.  
  222. 	public static String fillIndex(String index) {
  223. 		if (StringUtils.isBlank(index))
  224. 			return null;
  225.  
  226. 		StringBuffer sb = new StringBuffer(INDEX_PREX);
  227. 		if (index.startsWith(INDEX_PREX)) {
  228. 			return index;
  229. 		}
  230.  
  231. 		return sb.append(index).toString();
  232. 	}
  233.  
  234. 	/**
  235. 	 * 创建指定的 index 模板。
  236. 	 *
  237. 	 * @param tplName 索引模板名
  238. 	 * @param idxAliasName 索引的别名
  239. 	 * @param patterns 被应用模板的索引模式
  240. 	 * @param mappingBuilder 模板的索引映射配置
  241. 	 */
  242. 	public static void createIndexTemplate(String tplName, String idxAliasName, List<String> patterns,
  243. 																				 XContentBuilder mappingBuilder) {
  244. 		PutIndexTemplateRequest request = new PutIndexTemplateRequest(tplName);
  245. 		request.settings(getIndexSettings(idxAliasName)).patterns(patterns);
  246.  
  247. 		if (idxAliasName != null) {
  248. 			request.alias(new Alias(idxAliasName));
  249. 		}
  250.  
  251. 		if (mappingBuilder != null) {
  252. 			request.mapping(mappingBuilder);
  253. 		}
  254.  
  255. 		AcknowledgedResponse response = EsClients.executeRequest(client -> client.indices().putTemplate(request,
  256. 				RequestOptions.DEFAULT));
  257.  
  258. 		if (!response.isAcknowledged()) {
  259. 			logger.error("Failed to create index template'{}' with alias: {}", tplName, idxAliasName);
  260. 		}
  261. 	}
  262.  
  263. 	/*
  264. 	 * 返回索引的定义。
  265. 	 */
  266. 	private static Settings.Builder getIndexSettings(String aliasName) {
  267. 		final String paramPrefix = "pacific.es.index.";
  268. 		int shards = ConfigService.getInt(paramPrefix + aliasName + ".number_of_shards", 3);
  269. 		int replicas = ConfigService.getInt(paramPrefix + aliasName + ".number_of_replicas", 2);
  270. 		int fields = ConfigService.getInt(paramPrefix + aliasName + ".total_fields", 5000);
  271. 		return Settings.builder().put("index.number_of_shards", shards)
  272. 				.put("index.number_of_replicas", replicas).put("index.mapping.total_fields.limit", fields);
  273. 	}
  274. }

Elasticsearch索引动态字段基础类

  1. package *.*.common.es.support;
  2.  
  3. import java.io.IOException;
  4. import java.util.List;
  5. import java.util.Map;
  6. import java.util.Map.Entry;
  7. import java.util.stream.Collectors;
  8. import java.util.stream.Stream;
  9.  
  10. import org.elasticsearch.common.xcontent.XContentBuilder;
  11. import org.elasticsearch.common.xcontent.XContentFactory;
  12.  
  13. import *.*.common.config.ConfigService;
  14. import *.*.common.util.ConstantFields;
  15. import *.*.api.error.ServerException;
  16.  
  17. /**
  18.  * Es中对应的几种动态记录的mapping 
  19.  * 
  20.  * @author zhanglb
  21.  *
  22.  */
  23. public class EsActivityMapping {
  24. 	public static final String FIELD_TYPE_KEYWORD = "keyword";
  25. 	public static final String FIELD_TYPE_DATE = "date";
  26. 	public static final String FIELD_TYPE_TEXT = "text";
  27. 	public static final String FIELD_TYPE_LONG = "long";
  28. 	public static final String FIELD_TYPE_OBJECT = "object";
  29. 	public static final String FIELD_TYPE_CUSTOM = "custom";
  30. 	public static final List<String> NEED_LOWERCASE_TYPE = Stream.of(FIELD_TYPE_KEYWORD, FIELD_TYPE_TEXT).collect(Collectors.toList());
  31. 	public static final String CONTENT = "Total-OAO-context-OBO";
  32. 	public static final boolean OPEN_LOWERCASE = ConfigService.getInstance().isOpenLowercase();
  33.  
  34. 	private EsActivityMapping() {
  35. 	}
  36. 	
  37. 	public static XContentBuilder buildMapping(Map<String, String> propertiesMap) {
  38. 		try (XContentBuilder builder = XContentFactory.jsonBuilder();) {
  39. 			builder.startObject();
  40. 			{
  41. 				builder.startObject("properties");
  42. 				if (propertiesMap != null && !propertiesMap.isEmpty()) {
  43. 					for (Entry<String, String> entry : propertiesMap.entrySet()) {
  44. 						buildField(builder, entry.getKey(), entry.getValue());
  45. 					}
  46. 				}
  47. 				builder.endObject();
  48. 			}
  49. 			return builder.endObject();
  50. 		} catch (IOException e) {
  51. 			throw new ServerException("Failed to build index ci mapping", e);
  52. 		}
  53. 	}
  54.  
  55. 	public static XContentBuilder buildCIMapping() {
  56. 		try (XContentBuilder builder = XContentFactory.jsonBuilder();) {
  57. 			builder.startObject();
  58. 			{
  59. 				builder.startObject("properties");
  60. 				buildField(builder, CONTENT, FIELD_TYPE_KEYWORD);
  61. 				buildField(builder, ConstantFields.FIELD_ID, FIELD_TYPE_KEYWORD);
  62. 				buildField(builder, ConstantFields.FIELD_CLASSCODE, FIELD_TYPE_KEYWORD);
  63. 				buildField(builder, ConstantFields.RES_CIRCLE_FIELD, FIELD_TYPE_KEYWORD);
  64. 				buildField(builder, ConstantFields.CREATE_TIME, FIELD_TYPE_DATE);
  65. 				buildField(builder, ConstantFields.UPDATE_TIME, FIELD_TYPE_DATE);
  66. 				buildField(builder, ConstantFields.RES_ATTRVALUES, FIELD_TYPE_OBJECT);
  67. 			
  68. 				builder.endObject();
  69. 			}
  70. 			return builder.endObject();
  71. 		} catch (IOException e) {
  72. 			throw new ServerException("Failed to build index ci mapping", e);
  73. 		}
  74. 	}
  75. 	
  76. 	/**
  77. 	 * 创建配置项动态记录字段映射。
  78. 	 */
  79. 	public static XContentBuilder buildActivityMapping() {
  80. 		try (XContentBuilder builder = XContentFactory.jsonBuilder();) {
  81. 			builder.startObject();
  82. 			{
  83. 				builder.startObject("properties");
  84. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_ID, FIELD_TYPE_KEYWORD);
  85. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_CLASSCODE, FIELD_TYPE_KEYWORD);
  86. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_CLASSNAME, FIELD_TYPE_KEYWORD);
  87. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_ACTYPE, FIELD_TYPE_KEYWORD);
  88. 				buildField(builder, ConstantFields.FIELD_TYPE, FIELD_TYPE_KEYWORD);
  89. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_TIME, FIELD_TYPE_DATE);
  90. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_TENANTID, FIELD_TYPE_KEYWORD);
  91. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_VERSION, FIELD_TYPE_LONG);
  92. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_CIRCLEID, FIELD_TYPE_KEYWORD);
  93. 				buildField(builder, ConstantFields.FIELD_USERID, FIELD_TYPE_KEYWORD);
  94. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_USERNAME, FIELD_TYPE_TEXT);
  95. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_OPER, FIELD_TYPE_KEYWORD);
  96. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_OPERNAME, FIELD_TYPE_KEYWORD);
  97. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_ENTITYID, FIELD_TYPE_KEYWORD);
  98. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_ENTITYNAME, FIELD_TYPE_TEXT);
  99. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_DESC, FIELD_TYPE_TEXT);
  100. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_PARENTCINAME, FIELD_TYPE_TEXT);
  101. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_TARGETCINAME, FIELD_TYPE_TEXT);
  102. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_ITSMINFO, FIELD_TYPE_OBJECT);
  103. 				buildField(builder, ConstantFields.ACTIVITY_FIELD_RELATIONNAME, FIELD_TYPE_TEXT);
  104. 				buildField(builder, ConstantFields.CI_RECORD_FIELD_SOURCE, FIELD_TYPE_KEYWORD);
  105. 				buildField(builder, ConstantFields.FIELD_STATE, FIELD_TYPE_KEYWORD);
  106. 				buildField(builder, ConstantFields.CI_RECORD_FIELD_AUTO_CLAIM, FIELD_TYPE_KEYWORD);
  107. 				builder.endObject();
  108. 			}
  109. 			return builder.endObject();
  110. 		} catch (IOException e) {
  111. 			throw new ServerException("Failed to build index mapping", e);
  112. 		}
  113. 	}
  114.  
  115. 	private static void buildField(XContentBuilder builder, String field, String type) throws IOException {
  116. 		builder.startObject(field);
  117. 		{
  118. 			builder.field("type", type);
  119. 			if (OPEN_LOWERCASE && NEED_LOWERCASE_TYPE.contains(type)) {
  120. 				builder.field("normalizer", "lowercase");
  121. 			}
  122. 		}
  123. 		builder.endObject();
  124. 	}
  125. }

Elasticsearch的资源增删改查

写入数据

  1. 	/**
  2. 	 * 
  3. 	 * @param list
  4. 	 * @param testData
  5. 	 * @param syncByAlias 是否用别名
  6. 	 */
  7. 	@SuppressWarnings({ "unchecked" })
  8. 	public void insertDates(List<Test> list, CommonTestData testData, boolean syncByAlias) {
  9. 		ObjectMapper mapper = new ObjectMapper();
  10. 		String id = null;
  11. 		StringJoiner content;
  12. 		Map<String, TestClass> testClassMap = testData.getClassMap();
  13. 		Map<String, TestAttribute> testAttributeMap = testData.getAttrMap();
  14. 		if (resClassMap.isEmpty()) {
  15. 			return;
  16. 		}
  17. 		try {
  18. 			List<Map<String, Object>> objects = new ArrayList<>();
  19. 			for (Test entity : list) {
  20. 				TestClass testClass = testClassMap.get(entity.getClassCode());
  21. 				if (testClass == null) {
  22. 					continue;
  23. 				}
  24. 				List<TestAttribute> attrs = new ArrayList<>();
  25. 				for (String code : testClass.getAttrCodes()) {
  26. 					TestAttribute attr = testAttributeMap.get(code);
  27. 					if (attr != null) {
  28. 						attrs.add(attr);
  29. 					}
  30. 				}
  31.  
  32. 				content = new StringJoiner(ElasticsearchUtil.SPLIT_SYMBOL);
  33. 				String json = mapper.writeValueAsString(ElasticsearchUtil.converVo(entity, content, testClass, attrs, testData.getDictMap()));
  34. 				Map<String, Object> productMap = mapper.readValue(json, Map.class);
  35. 				id = productMap.get(TestAttrFieldCodes.BASE_FIELD_ID).toString();
  36. 				if (productMap.containsKey(ATTRVALUES)) {
  37. 					Map<String, Object> result = new HashMap<>();
  38. 					Map<String, Object> attr = (Map<String, Object>) productMap.get(ATTRVALUES);
  39. 					for (Map.Entry<String, Object> entry : attr.entrySet()) {
  40. 						if (entry.getValue() == null || StringUtils.isBlank(entry.getValue().toString())) {
  41. 							continue;
  42. 						}
  43. 						TestAttribute att = testAttributeMap.get(entry.getKey().toString());
  44.  
  45. 						if (att != null) {
  46. 							result.put(att.getType() + SPLIT + entry.getKey(), entry.getValue());
  47. 							content.add(String.valueOf(entry.getValue()));
  48. 						}
  49.  
  50. 					}
  51. 					productMap.put(ATTRVALUES, result);
  52. 				}
  53. 				productMap.put(ElasticsearchUtil.CONTENT, ElasticsearchUtil.SPLIT_SYMBOL + content.toString() + ElasticsearchUtil.SPLIT_SYMBOL);
  54. 				objects.add(productMap);
  55. 			}
  56. 			if (syncByAlias) {
  57. 				EsDataWriter.saveObjectsAsync(ElasticsearchUtil.getIndexAliasName(), objects);
  58. 			} else {
  59. 				EsDataWriter.saveObjectsAsync(ElasticsearchUtil.getIndexName(), objects);
  60. 			}
  61. 			
  62. 		} catch (JsonParseException e) {
  63. 			resData.clearData();
  64. 			logger.error("Failed to parse JSON data id= {} , message : {}", id, e);
  65. 			throw new ServerException("Failed to parse JSON data" + id, e);
  66. 		} catch (JsonMappingException e) {
  67. 			resData.clearData();
  68. 			logger.error("Failed to parse JSON data id= {} , message : {}", id, e);
  69. 			throw new ServerException("Failed to process JSON data" + id, e);
  70. 		} catch (IOException e) {
  71. 			resData.clearData();
  72. 			logger.error("Failed to parse JSON data id= {} , message : {}", id, e);
  73. 			throw new ServerException("Failed to execute" + id, e);
  74. 		}
  75. 	}

删除数据

  1. @Override
  2. 	public void delById(String id, String type) {
  3. 		try {
  4. 			DeleteRequest deleteRequest = new DeleteRequest(ElasticsearchUtil.getIndexAliasName(), id);
  5. 			EsClients.getClient().delete(deleteRequest, RequestOptions.DEFAULT);
  6. 		} catch (Exception e) {
  7. 			e.printStackTrace();
  8. 		}
  9.  
  10. 	}

查询数据

  1. public Map<String, Object> getSearchData(String index, String type, String value, boolean isPreciseSearch, Integer page, Integer size) {
  2. 		Map<String, Object> result = new HashMap<>();
  3. 		Integer from = 0;
  4. 		if (page - 1 >= 0) {
  5. 			from = (page - 1) * size;
  6. 		}
  7. 		try {
  8. 			if (isPreciseSearch) {
  9. 				value = "*" + ElasticsearchUtil.SPLIT_SYMBOL + value.substring(1, value.length() - 1) + ElasticsearchUtil.SPLIT_SYMBOL + "*";
  10. 			}
  11.  
  12. 			if (value.length() == 1) {
  13. 				value = CommonUtil.replaceRegexChars(value);
  14. 			} else {
  15. 				value = CommonUtil.replaceRegexChars(value).replace("\\*", "*");
  16. 			}
  17. 			SearchResponse response = this.getResponse(index, type, value, from, size);
  18. 			
  19. 			long totalRecords = response.getHits().getTotalHits().value;
  20. 			List<Map<String, Object>> list = new ArrayList<>();
  21. 			SearchHit[] ciHits = response.getHits().getHits();
  22. 			CacheContext context = new CacheContext();
  23. 			for (int i = 0; i < ciHits.length; i++) {
  24. 				String objId = ciHits[i].getId();
  25. 				Map<String, Object> map = this.filter(ciHits[i].getSourceAsMap(), objId, context);
  26. 				map.put(TestAttrFieldCodes.BASE_FIELD_ID, objId);
  27. 				list.add(map);
  28. 			}
  29. 			context.clear();
  30. 			result.put("totalRecords", totalRecords);
  31. 			result.put("list", list);
  32. 			return result;
  33. 		} catch (Exception e) {// NOSONAR 此处增加代码稳定性
  34. 			throw new ServerException("Failed to search data", e);
  35. 		}
  36. 	}
  37.  
  38. private SearchResponse getResponse(String index, String type, String value, int from, int size) {
  39. 		SearchSourceBuilder sb = new SearchSourceBuilder();
  40. 		if (ConfigService.getInstance().isEsQueryNeedPointTotalOAOContxtObo()) {
  41. 			BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery()
  42. 					.must(QueryBuilders.queryStringQuery("*" + value + "*").defaultField(ElasticsearchUtil.getQueryContent()));
  43. 			// 是否开启分词查询
  44. 			if (ConfigService.getInstance().isOpenEsParticipleQuery()) {
  45. 				queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.matchPhraseQuery(ElasticsearchUtil.getQueryContent(), value));
  46. 			}
  47. 			queryBuilder.filter(
  48. 					QueryBuilders.boolQuery().must(QueryBuilders.termQuery(ResAttrFieldCodes.BASE_FIELD_TENANT, ThreadLocalContext.getTenantId())));
  49. 			sb.query(queryBuilder);
  50. 		} else {
  51. 			BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.queryStringQuery(value + "*")).filter(
  52. 					QueryBuilders.boolQuery().must(QueryBuilders.termQuery(ResAttrFieldCodes.BASE_FIELD_TENANT, ThreadLocalContext.getTenantId())));
  53. 			sb.query(queryBuilder);
  54. 		}
  55.  
  56. 		// 分页信息
  57. 		sb.from(from);
  58. 		sb.size(size);
  59.  
  60. 		// 是否统计总数
  61. 		sb.trackTotalHits(true);
  62. 		sb.sort(ConstantFields.UPDATE_TIME, SortOrder.DESC);
  63. 		List<String> indexs = new ArrayList<>();
  64. 		indexs.add(index);
  65. 		return EsDataReader.query(indexs, sb);
  66. 	}

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/你好赵伟/article/detail/68684
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号