Spring Boot + Elasticsearch 实现索引批量写入

傷城~ 2023-07-25 08:42 10阅读 0赞

# [Spring Boot + Elasticsearch 实现索引批量写入][Spring Boot _ Elasticsearch] #

在使用Eleasticsearch进行索引维护的过程中，如果你的应用场景需要频繁的大批量的索引写入，再使用上篇中提到的维护方法的话显然效率是低下的，此时推荐使用bulkIndex来提升效率。批写入数据块的大小取决于你的数据集及集群的配置。

下面我们以Spring Boot结合Elasticsearch创建一个示例项目，从基本的pom配置开始

1.  `<dependency>`
2.  `   <groupId>com.google.code.gson</groupId>`
3.  `   <artifactId>gson</artifactId>`
4.  `   <version>1.4</version>`
5.  `</dependency>`
6.  `<dependency>`
7.  `   <groupId>org.springframework.boot</groupId>`
8.  `   <artifactId>spring-boot-starter-data-elasticsearch</artifactId>`
9.  `</dependency>`

application.properties配置

1.  `#elasticsearch config`
2.  `spring.data.elasticsearch.cluster-name:elasticsearch`
3.  `spring.data.elasticsearch.cluster-nodes:192.168.1.105:9300`
4.   
5.  `#application config`
6.  `server.port=8080`
7.  `spring.application.name=esp-app`

我们需要定义域的实体和一个Spring data的基本的CRUD支持库类。用id注释定义标识符字段，如果你没有指定ID字段，Elasticsearch不能索引你的文件。同时需要指定索引名称类型，@Document注解也有助于我们设置分片和副本数量。

1.  `@Data`
2.  `@Document(indexName = "carIndex", type = "carType", shards = 1, replicas = 0)`
3.  `public class Car implements Serializable {`
4.  `   /**`
5.  `    * serialVersionUID:`
6.  `    * @since JDK 1.6`
7.  `    */`
8.  `   private static final long serialVersionUID = 1L;`
9.  `   @Id`
10. `   private Long id;`
11. `   private String brand;`
12. `   private String model;`
13. `   private BigDecimal amount;`
14.  
15. `   public Car(Long id, String brand, String model, BigDecimal amount) {`
16. `       this.id = id;`
17. `       this.brand = brand;`
18. `       this.model = model;`
19. `       this.amount = amount;`
20. `   }`
21. `}`

接着定义一个IndexService并使用bulk请求来处理索引，操作前首先要判断索引是否存在，以免出现异常。为了更好的掌握Java API，这里采用了不同于上篇中ElasticSearchRepository的ElasticSearchTemplate工具集，相对来讲功能更加丰富。

1.  `@Service`
2.  `public class IndexerService {`
3.  `   private static final String CAR_INDEX_NAME = "car_index";`
4.  `   private static final String CAR_INDEX_TYPE = "car_type";`
5.  `   @Autowired`
6.  `   ElasticsearchTemplate elasticsearchTemplate;`
7.   
8.  `   public long bulkIndex() throws Exception {`
9.  `       int counter = 0;`
10. `       try {`
11. `           //判断索引是否存在`
12. `           if (!elasticsearchTemplate.indexExists(CAR_INDEX_NAME)) {`
13. `               elasticsearchTemplate.createIndex(CAR_INDEX_NAME);`
14. `           }`
15. `           Gson gson = new Gson();`
16. `           List<IndexQuery> queries = new ArrayList<IndexQuery>();`
17. `           List<Car> cars = assembleTestData();`
18. `           for (Car car : cars) {`
19. `               IndexQuery indexQuery = new IndexQuery();`
20. `               indexQuery.setId(car.getId().toString());`
21. `               indexQuery.setSource(gson.toJson(car));`
22. `               indexQuery.setIndexName(CAR_INDEX_NAME);`
23. `               indexQuery.setType(CAR_INDEX_TYPE);`
24. `               queries.add(indexQuery);`
25. `               //分批提交索引`
26. `               if (counter % 500 == 0) {`
27. `                   elasticsearchTemplate.bulkIndex(queries);`
28. `                   queries.clear();`
29. `                   System.out.println("bulkIndex counter : " + counter);`
30. `               }`
31. `               counter++;`
32. `           }`
33. `           //不足批的索引最后不要忘记提交`
34. `           if (queries.size() > 0) {`
35. `               elasticsearchTemplate.bulkIndex(queries);`
36. `           }`
37. `           elasticsearchTemplate.refresh(CAR_INDEX_NAME);`
38. `           System.out.println("bulkIndex completed.");`
39. `       } catch (Exception e) {`
40. `           System.out.println("IndexerService.bulkIndex e;" + e.getMessage());`
41. `           throw e;`
42. `       }`
43.  
44. `       return -1;`
45. `   }`
46.  
47. `   private List<Car> assembleTestData() {`
48. `       List<Car> cars = new ArrayList<Car>();`
49. `       //随机生成10000个索引，以便下一次批量写入`
50. `       for (int i = 0; i < 10000; i++) {`
51. `           cars.add(new Car(RandomUtils.nextLong(1, 11111), RandomStringUtils.randomAscii(20), RandomStringUtils.randomAlphabetic(15), BigDecimal.valueOf(78000)));`
52. `       }`
53. `       return cars;`
54. `   }`
55. `}`

再下面的工作就比较简单了，可以编写一个RestController接受请求来测试或者CommandLineRunner，在系统启动时就加载上面的方法。

1.  `@SpringBootApplication`
2.  `@RestController`
3.  `public class ESPApplicatoin {`
4.   
5.  `   public static void main(String[] args) {`
6.  `       SpringApplication.run(ESPApplicatoin.class, args);`
7.  `   }`
8.   
9.  `   @Autowired`
10. `   IndexerService indexService;`
11.  
12.  
13. `   @RequestMapping(value = "bulkIndex",method = RequestMethod.POST)`
14. `   public void bulkIndex(){`
15. `       try {`
16. `           indexService.bulkIndex();`
17. `       } catch (Exception e) {`
18. `           e.printStackTrace();`
19. `       }`
20. `   }`
21. `}`

CommandLineRunner方法类：

1.  `@Component`
2.  `public class AppLoader implements CommandLineRunner {`
3.  `   @Autowired`
4.  `   IndexerService indexerService;`
5.   
6.  `   @Override`
7.  `   public void run(String... strings) throws Exception {`
8.  `       indexerService.bulkIndex();`
9.  `   }`
10. `}`

结束后，就可在通过地址http://localhost:9200/car\_index/\_search/来查看索引到底有无生效。注：要特别关注版本的兼容问题，如果用Es 5+的话，显然不能采用Spring Data Elasticsearch的方式。

<table> 
 <thead> 
  <tr> 
   <th> <p>Spring Boot&nbsp;</p> <p>Version (x)</p> </th> 
   <th>Spring Data Elasticsearch Version (y)</th> 
   <th>Elasticsearch Version (z)</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>x &lt;= 1.3.5</td> 
   <td>y &lt;= 1.3.4</td> 
   <td>z &lt;= 1.7.2*</td> 
  </tr> 
  <tr> 
   <td>x &gt;= 1.4.x</td> 
   <td>2.0.0 &lt;=y &lt; 5.0.0**</td> 
   <td>2.0.0 &lt;= z &lt; 5.0.0**</td> 
  </tr> 
 </tbody> 
</table>

--------------------

(\*) - require manual change in your project pom file (solution 2.)

(\*\*) - Next big ES release with breaking changes

[Spring Boot _ Elasticsearch]: https://www.cnblogs.com/growithus/p/11012209.html