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

java 使用springBoot 操作Elasticsearch组件_springboot es bulkrequest bulkprocessor

作者：你好赵伟 | 2024-03-29 06:40:35

赞

踩

springboot es bulkrequest bulkprocessor

简介

  该代码是以组件的形式写入，使用时在其他方法中注入  ElasticSearchUtil 类进行使用，主要实现了动态的增删改查，分页模糊查询，以指定字段查询与全部字段进行全差

should和must的比较

  1. should模式：就算有一个字段没有匹配的也会有结果返回
  2. must模式：必须要求所有字段都匹配到，只要有一个字段不匹配就没有搜索结果

版本信息

  • Elasticsearch：2.x
  • jdk  1.8

maven 依赖

  1.  
  2. 	<parent>
  3. 		<groupId>org.springframework.boot</groupId>
  4. 		<artifactId>spring-boot-starter-parent</artifactId>
  5. 		<version>1.5.8.RELEASE</version>
  6. 	</parent>
  7.  
  8.  
  9. 	<properties>
  10. 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  11. 		<java.version>1.8</java.version>
  12. 	</properties>
  13.  
  14. 	<dependencies>
  15. 		<!-- elasticsearch start -->
  16. 		<dependency>
  17. 			<groupId>org.springframework.boot</groupId>
  18. 			<artifactId>spring-boot-starter-data-elasticsearch</artifactId>
  19. 		</dependency>
  20. 		<!-- springBoot start -->
  21.  
  22. 		<dependency>
  23. 			<groupId>org.springframework.boot</groupId>
  24. 			<artifactId>spring-boot-starter-test</artifactId>
  25. 			<scope>test</scope>
  26. 		</dependency>
  27. 		<dependency>
  28. 			<groupId>com.sun.jna</groupId>
  29. 			<artifactId>jna</artifactId>
  30. 			<version>3.0.9</version>
  31. 		</dependency>
  32. 		<dependency>
  33. 			<groupId>org.springframework.boot</groupId>
  34. 			<artifactId>spring-boot-starter-jetty</artifactId>
  35. 		</dependency>
  36. 		<dependency>
  37. 			<groupId>org.springframework.boot</groupId>
  38. 			<artifactId>spring-boot-starter-aop</artifactId>
  39. 		</dependency>
  40. 		<dependency>
  41. 			<groupId>org.springframework.boot</groupId>
  42. 			<artifactId>spring-boot-starter-web</artifactId>
  43. 		</dependency>
  44. 		<!-- springBoot end -->
  45.  
  46. 		<!-- json start -->
  47. 		<dependency>
  48. 			<groupId>com.alibaba</groupId>
  49. 			<artifactId>fastjson</artifactId>
  50. 			<version>1.2.4</version>
  51. 		</dependency>
  52. 		<!-- json end -->
  53.  
  54. 	</dependencies>

config 底层操作类

  1. package *.config;
  2.  
  3. import java.lang.reflect.Field;
  4. import java.util.Map;
  5. import java.util.Set;
  6.  
  7. import javax.annotation.PostConstruct;
  8. import javax.annotation.PreDestroy;
  9. import org.elasticsearch.action.bulk.BackoffPolicy;
  10. import org.elasticsearch.action.bulk.BulkProcessor;
  11. import org.elasticsearch.action.bulk.BulkRequest;
  12. import org.elasticsearch.action.bulk.BulkResponse;
  13. import org.elasticsearch.action.delete.DeleteResponse;
  14. import org.elasticsearch.action.get.GetResponse;
  15. import org.elasticsearch.action.index.IndexResponse;
  16. import org.elasticsearch.action.search.SearchRequestBuilder;
  17. import org.elasticsearch.action.search.SearchResponse;
  18. import org.elasticsearch.action.update.UpdateResponse;
  19. import org.elasticsearch.client.Client;
  20. import org.elasticsearch.common.unit.ByteSizeUnit;
  21. import org.elasticsearch.common.unit.ByteSizeValue;
  22. import org.elasticsearch.common.unit.TimeValue;
  23. import org.elasticsearch.index.query.BoolQueryBuilder;
  24. import org.elasticsearch.index.query.QueryBuilders;
  25. import org.elasticsearch.search.SearchHit;
  26. import org.slf4j.Logger;
  27. import org.slf4j.LoggerFactory;
  28. import org.springframework.beans.factory.annotation.Autowired;
  29. import org.springframework.stereotype.Component;
  30. import com.alibaba.fastjson.JSONArray;
  31. import com.alibaba.fastjson.JSONObject;
  32.  
  33. /**
  34.  * 底层操作类
  35.  * @ClassName: ElasticConfigration 
  36.  * @date 2018年11月20日
  37.  * @author tang wang
  38.  *
  39.  */
  40. @Component
  41. public class ElasticConfigration {
  42.  
  43. 	private final Logger logger = LoggerFactory.getLogger(ElasticConfigration.class);
  44.  
  45. 	@Autowired
  46. 	private Client client;
  47.  
  48. 	private BulkProcessor bulkProcessor;
  49. 	@PostConstruct
  50. 	public void initBulkProcessor() {
  51.  
  52. 		bulkProcessor = BulkProcessor.builder(client, new BulkProcessor.Listener() {
  53.  
  54. 			@Override
  55. 			public void beforeBulk(long executionId, BulkRequest request) {
  56. 				logger.info("序号：{} 开始执行{} 条记录保存", executionId, request.numberOfActions());
  57. 			}
  58.  
  59. 			@Override
  60. 			public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
  61. 				logger.error(String.format("序号：%s 执行失败; 总记录数：%s", executionId, request.numberOfActions()), failure);
  62. 			}
  63.  
  64. 			@Override
  65. 			public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
  66. 				logger.info("序号：{} 执行{}条记录保存成功,耗时：{}毫秒,", executionId,
  67. 						request.numberOfActions(), response.getTookInMillis());
  68. 			}
  69. 		}).setBulkActions(1000)
  70. 				.setBulkSize(new ByteSizeValue(10, ByteSizeUnit.MB))
  71. 				.setConcurrentRequests(4)
  72. 				.setFlushInterval(TimeValue.timeValueSeconds(5))
  73. 				/**
  74. 				 * 失败后等待多久及重试次数
  75. 				 */
  76. 				.setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(500), 3))  
  77. 				.build();
  78. 	}
  79.  
  80.  
  81. 	@PreDestroy
  82. 	public void closeBulk() {
  83. 		if (bulkProcessor != null) {
  84. 			try {
  85. 				bulkProcessor.close();
  86. 			} catch (Exception e) {
  87. 				logger.error("close bulkProcessor exception", e);
  88. 			}
  89. 		}
  90. 	}
  91.  
  92.  
  93. 	/**
  94. 	 * 批量添加,性能最好
  95. 	 * 
  96. 	 */
  97. 	public void addDocumentToBulkProcessor(String indices, String type, Object object) {
  98. 		bulkProcessor.add(client.prepareIndex(indices, type).setSource(JSONObject.toJSONString(object)).request());
  99. 	}
  100.  
  101. 	/**
  102. 	 * 添加数据 
  103. 	 * @Title:  ElasticConfigration   
  104. 	 * @Description:
  105. 	 * @param indices 索引名字 
  106. 	 * @param type    索引类型
  107. 	 * @param object  索引数据
  108. 	 * @author: tao wang 
  109. 	 * @date:   2018年11月14日
  110. 	 * @throws
  111. 	 */
  112. 	public void addDocument(String indices, String type, Object object) {
  113.  
  114. 		IndexResponse resp = client.prepareIndex(indices, type).setSource(JSONObject.toJSONString(object)).get();
  115. 		logger.info("添加结果：{}", resp.toString());
  116. 	}
  117.  
  118. 	/**
  119. 	 * 按照Id 进行删除
  120. 	 * @Title:  ElasticConfigration   
  121. 	 * @Description:
  122. 	 * @param index  索引名称
  123. 	 * @param type   索引类型
  124. 	 * @param id     数据Id
  125. 	 * @author: tao wang 
  126. 	 * @date:   2018年11月14日
  127. 	 * @throws
  128. 	 */
  129. 	public void deleteDocumentById(String index, String type, String id) {
  130. 		// new DeleteByQueryRequest(search);
  131. 		DeleteResponse resp = client.prepareDelete(index, type, id).get();
  132. 		logger.info("删除结果：{}", resp.toString());
  133. 	}
  134.  
  135. 	/**
  136. 	 * 查询单个数据
  137. 	 * @param  index 引擎名称
  138. 	 * @param  type  引擎类型
  139. 	 * @param  id    引擎Id  
  140. 	 * @return   
  141. 	 * @auth tao wang
  142. 	 */
  143.  
  144. 	public JSONObject getDocmentById(String index, String type, String id) {
  145. 		JSONObject obj = new JSONObject();
  146. 		GetResponse getResponse = client.prepareGet(index, type, id).get();
  147. 		Map<String, Object> map = getResponse.getSource();
  148. 		obj.put("id", id);
  149. 		obj.put("value", map);
  150. 		return JSONObject.parseObject(JSONObject.toJSONString(map));
  151. 	}
  152.  
  153.  
  154. 	/**
  155. 	 * 按ID更新
  156. 	 * @Title:  ElasticConfigration   
  157. 	 * @Description:
  158. 	 * @param indices  索引名称
  159. 	 * @param type    索引类型
  160. 	 * @param id      数据Id
  161. 	 * @param object  数据值
  162. 	 * @author: tao wang 
  163. 	 * @date:   2018年11月14日
  164. 	 * @throws
  165. 	 */
  166. 	public void updateDocument(String indices, String type, String id, Object object) {
  167. 		UpdateResponse resp = client.prepareUpdate(indices, type, id).setDoc(JSONObject.toJSONString(object)).get();
  168. 		logger.info("更新结果：{}", resp.toString());
  169. 	}
  170.  
  171. 	/**
  172. 	 * 查询所有字段,默认分页
  173. 	 * @Title:  ElasticConfigration   
  174. 	 * @Description:
  175. 	 * @param indices 索引名称
  176. 	 * @param type    索引类型
  177. 	 * @param clazz   返回的集合对象
  178. 	 * @return
  179. 	 * @author: tao wang 
  180. 	 * @date:   2018年11月14日
  181. 	 * @throws
  182. 	 */
  183. 	public JSONArray queryDocumentByParam(String indices, String type) {
  184. 		SearchRequestBuilder builder = buildRequest(indices, type);
  185. 		SearchResponse resp = builder.get();
  186. 		return convertResponse(resp);
  187. 	}
  188.  
  189. 	/**
  190. 	 * 进行分页数据查询
  191. 	 * @param indices  索引名称
  192. 	 * @param type    索引类型
  193. 	 * @param key     模糊字段
  194. 	 * @param value   模糊字段
  195. 	 * @param pageNumber  当前页数
  196. 	 * @param pageSize    每页显示数据
  197. 	 * @return   
  198. 	 * @auth tao wang
  199. 	 * @date 2018年11月26日
  200. 	 */
  201. 	public JSONArray queryDocumentByParam(String indices, String type, String key, String value, Integer pageNumber, Integer pageSize) {
  202. 		SearchRequestBuilder builder = buildRequest(indices, type);
  203. 		BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
  204. 		/**
  205. 		 * must 满足所有才返回
  206. 		 */
  207. 		boolQueryBuilder.must(QueryBuilders.wildcardQuery(key, ("*" + value + "*").toLowerCase()));
  208. 		builder.setQuery(boolQueryBuilder);
  209. 		builder.setFrom(pageNumber).setSize(pageSize);
  210. 		SearchResponse resp = builder.get();
  211. 		return convertResponse(resp);
  212. 	}
  213.  
  214. 	/**
  215. 	 * 进行模糊查询所有，不进行分页
  216. 	 * @param indices  索引名称
  217. 	 * @param type    索引类型
  218. 	 * @param value   模糊字段
  219. 	 * @param obj     类
  220. 	 * @return   
  221. 	 * @auth tao wang
  222. 	 * @date 2018年11月26日
  223. 	 */
  224. 	@SuppressWarnings("rawtypes")
  225. 	public JSONArray queryDocumentByParam(String indices, String type, String value, Class clazz) {
  226.  
  227. 		SearchRequestBuilder builder = buildRequest(indices, type);
  228. 		BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
  229.  
  230.  
  231. 		// 获取实体类的所有属性信息，返回Field数组  
  232. 		Field[] fields = clazz.getDeclaredFields();  
  233. 		System.err.println(fields.toString());
  234. 		for (Field field : fields) {  
  235. 			/**
  236. 			 * should 满足一个都返回
  237. 			 */
  238. 			boolQueryBuilder.should(QueryBuilders.wildcardQuery(field.getName(), ("*" + value + "*").toLowerCase()));
  239. 			System.err.println(field.getName());  
  240. 		}  			
  241. 		builder.setQuery(boolQueryBuilder);
  242. 		SearchResponse resp = builder.get();
  243. 		return convertResponse(resp);
  244. 	}
  245.  
  246. 	/**
  247. 	 * 进行模糊查询所有，不进行分页
  248. 	 * @param indices  索引名称
  249. 	 * @param type    索引类型
  250. 	 * @param value   模糊字段
  251. 	 * @param setKey     类的属性名称
  252. 	 * @return   
  253. 	 * @auth tao wang
  254. 	 * @date 2018年11月26日
  255. 	 */
  256. 	public JSONArray queryDocumentByParam(String indices, String type, String value, Set<String> setKey) {
  257.  
  258. 		SearchRequestBuilder builder = buildRequest(indices, type);
  259. 		BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
  260. 		for (String string : setKey) {
  261. 			boolQueryBuilder.should(QueryBuilders.wildcardQuery(string, ("*" + value + "*").toLowerCase()));	
  262. 		}
  263. 		builder.setQuery(boolQueryBuilder);
  264. 		SearchResponse resp = builder.get();
  265. 		return convertResponse(resp);
  266. 	}
  267.  
  268. 	/**
  269. 	 * 通用的装换返回结果
  270. 	 * @Title:  ElasticConfigration   
  271. 	 * @Description:
  272. 	 * @param response 数据
  273. 	 * @param clazz    实体类
  274. 	 * @return
  275. 	 * @author: tao wang 
  276. 	 * @date:   2018年11月14日
  277. 	 * @throws
  278. 	 */
  279. 	public JSONArray convertResponse(SearchResponse response) {
  280. 		JSONArray list = new JSONArray();
  281. 		if (response != null && response.getHits() != null) {
  282. 			for (SearchHit hit : response.getHits()) {
  283. 				Map<String, Object> source = hit.getSource();
  284. 				String result = JSONObject.toJSONString(source);
  285. 				if (org.springframework.util.StringUtils.hasText(result)) {
  286. 					JSONObject obj = new JSONObject();
  287. 					obj.put("id", hit.getId());
  288. 					obj.put("value", result);
  289. 					list.add(obj);
  290. 				}
  291. 			}
  292. 		}
  293. 		return list;
  294. 	}
  295.  
  296. 	/**
  297. 	 * 进行数据数据绑定
  298. 	 * @Title:  ElasticConfigration   
  299. 	 * @Description: 
  300. 	 * @param indices  索引名称
  301. 	 * @param type    索引类型
  302. 	 * @return
  303. 	 * @author: tao wang 
  304. 	 * @date:   2018年11月14日
  305. 	 * @throws
  306. 	 */
  307. 	public SearchRequestBuilder buildRequest(String indices, String type) {
  308. 		return client.prepareSearch(indices).setTypes(type);
  309. 	}
  310. }

业务实现类：

 

  1. package *.util;
  2.  
  3. import java.util.Set;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.stereotype.Component;
  6. import com.alibaba.fastjson.JSONArray;
  7. import com.alibaba.fastjson.JSONObject;
  8. import *.config.ElasticConfigration;
  9.  
  10. /**
  11.  * 业务实现接口
  12.  * @ClassName: ElasticSearchUtil 
  13.  * @date 2018年11月20日
  14.  * @author tang wang
  15.  *
  16.  */
  17. @Component
  18. public class ElasticSearchUtil {
  19.  
  20. 	@Autowired
  21. 	private  ElasticConfigration elasticConfigration;
  22.  
  23. 	/**
  24. 	 * 添加指定的索引名称和类型数据
  25. 	 * @Title:  ElasticConfigUtil   
  26. 	 * @Description:
  27. 	 * @param indices 索引名称
  28. 	 * @param type    索引类型
  29. 	 * @param object  JSON 对象
  30. 	 * @return
  31. 	 * @author: tao wang 
  32. 	 * @date:   2018年11月14日
  33. 	 * @throws
  34. 	 */
  35. 	public  boolean   insert(String indices, String type, JSONObject object) {
  36. 		elasticConfigration.addDocument(indices.toLowerCase(), type.toLowerCase(), object);
  37. 		return true;
  38. 	}
  39.  
  40. 	/**
  41. 	 * 添加多条数据
  42. 	 * @param indices 索引名称
  43. 	 * @param type    索引类型
  44. 	 * @param array   索引的对象集合
  45. 	 * @return
  46. 	 */
  47. 	public  boolean  insert(String indices, String type, JSONArray array) {
  48. 		for (Object object : array) {
  49. 			elasticConfigration.addDocument(indices.toLowerCase(), type.toLowerCase(), object);
  50. 		}
  51. 		return true;
  52. 	}
  53.  
  54. 	/**
  55. 	 * 删除单条数据
  56. 	 * @param @param indices 索引名称
  57. 	 * @param @param type   索引类型
  58. 	 * @param @param id    索引Id
  59. 	 * @param @return 
  60. 	 * @return   
  61. 	 * @auth tao wang
  62. 	 */
  63. 	public boolean delete(String indices, String type, String id) {
  64. 		elasticConfigration.deleteDocumentById(indices.toLowerCase(), type.toLowerCase(), id);
  65. 		return true;
  66. 	}
  67.  
  68. 	/**
  69. 	 * 删除多条数据
  70. 	 * @param @param indices 索引名称
  71. 	 * @param @param type    索引类型
  72. 	 * @param @param ids     id 集合
  73. 	 * @param @return 
  74. 	 * @return   
  75. 	 * @auth tao wang
  76. 	 */
  77. 	public boolean delete(String indices, String type, Set<String> ids) {
  78. 		for (String id : ids) {
  79. 			elasticConfigration.deleteDocumentById(indices.toLowerCase(), type.toLowerCase(), id);
  80. 		}
  81. 		return true;
  82. 	}
  83.  
  84. 	/**
  85. 	 * 获取单个数据
  86. 	 * @param @param index 索引名称
  87. 	 * @param @param type 索引类型
  88. 	 * @param @param id   索引Id
  89. 	 * @param @return  返回数据
  90. 	 * @return   
  91. 	 * @auth tao wang
  92. 	 */
  93. 	public JSONObject getDocmentById(String indices, String type, String id) {
  94. 		return elasticConfigration.getDocmentById(indices.toLowerCase(), type.toLowerCase(), id);
  95. 	}
  96.  
  97. 	/**
  98. 	 * 修改单条数据
  99. 	 * @param @param indices 索引名称
  100. 	 * @param @param type   索引类型
  101. 	 * @param @param id     索引Id
  102. 	 * @param @param object  修改数据 数据不修改传原值
  103. 	 * @param @return 
  104. 	 * @return   
  105. 	 * @auth tao wang
  106. 	 */
  107. 	public boolean  update(String indices, String type, String id, JSONObject object) {
  108. 		elasticConfigration.updateDocument(indices.toLowerCase(), type.toLowerCase(), id, object);
  109. 		return true;
  110. 	}
  111.  
  112. 	/**
  113. 	 * 分页查询搜索引擎里面的数据
  114. 	 * @Title:  ElasticSearchUtil   
  115. 	 * @Description:
  116. 	 * @param pageNumber 当前页数
  117. 	 * @param pageSize  每页条数
  118. 	 * @param key    要模糊字段名称
  119. 	 * @param value  要模糊字段值
  120. 	 * @param indices 索引名称
  121. 	 * @param type    索引类型
  122. 	 * @param 
  123. 	 * @return
  124. 	 * @author: tao wang 
  125. 	 * @date:   2018年11月12日
  126. 	 * @throws
  127. 	 */
  128.  
  129. 	public JSONArray searchEmployee(String indices, String type, Integer pageNumber, 
  130. 			Integer pageSize, String key, String value) {
  131. 		return elasticConfigration.queryDocumentByParam(indices.toLowerCase(), type.toLowerCase(),
  132. 				key, value, pageNumber, pageSize);
  133. 	}
  134.  
  135.  
  136. 	/**
  137. 	 * 查询搜索里面的数据
  138. 	 * @param indices 索引名称
  139. 	 * @param type   索引类型
  140. 	 * @param value  模糊字段名称
  141. 	 * @param obj    类属性
  142. 	 * @return   
  143. 	 * @auth tao wang
  144. 	 * @date 2018年11月26日
  145. 	 */
  146. 	@SuppressWarnings("rawtypes")
  147. 	public JSONArray searchEmployee(String indices, String type, String value, Class  clazz) {
  148. 		return elasticConfigration.queryDocumentByParam(indices.toLowerCase(), type.toLowerCase(), value, clazz);
  149. 	}
  150.  
  151. 	/**
  152. 	 * 查询搜索里面的数据
  153. 	 * @param indices  索引名称
  154. 	 * @param type    索引类型
  155. 	 * @param value   模糊字段名称
  156. 	 * @param setKsy  类属性名称集合
  157. 	 * @return   
  158. 	 * @auth tao wang
  159. 	 * @date 2018年11月26日
  160. 	 */
  161. 	public JSONArray searchEmployee(String indices, String type, String value, Set<String> setKsy) {
  162. 		return elasticConfigration.queryDocumentByParam(indices.toLowerCase(), type.toLowerCase(), value, setKsy);
  163. 	}
  164.  
  165.  
  166. 	/**
  167. 	 * 不进行分页查询
  168. 	 * @Title:  ElasticSearchUtil   
  169. 	 * @Description:
  170. 	 * @param key    要模糊字段名称，多个中间逗号分开
  171. 	 * @param value  要模糊字段值
  172. 	 * @param indices 索引名称
  173. 	 * @param type    索引类型
  174. 	 * @param clazz   要返回的类集合
  175. 	 * @return
  176. 	 * @author: tao wang 
  177. 	 * @date:   2018年11月12日
  178. 	 * @throws
  179. 	 */
  180.  
  181. 	public JSONArray searchEmployee(String indices, String type) {
  182. 		return elasticConfigration.queryDocumentByParam(indices.toLowerCase(), type.toLowerCase());
  183. 	}
  184.  
  185. }

配置类：application.properties

  1.  
  2.  
  3.  
  4. ##es的名称 默认为elasticsearch
  5. spring.data.elasticsearch.cluster-name=*
  6. #配置es节点信息，逗号分隔，如果没有指定，则启动ClientNode
  7. spring.data.elasticsearch.cluster-nodes=*:9300

 

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

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

  

闽ICP备14008679号