devbox
小小林熬夜学编程
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

JAVA开发与JAVA（一文学会使用ElasticSearch）_java elasticsearch

作者：小小林熬夜学编程 | 2024-05-13 22:07:35

赞

踩

java elasticsearch

        在web网站的架设中特别是数据量大的网站或者APP小程序需要搜索或者全文检索的场景，几乎都需要借助ElasticSearch来作为全文检索引擎，以提高网站的搜索效率和性能。

这一节，我们通过一篇文章介绍，使大家通过一文就学会使用ElasticSearch。

一、ElasticSearch介绍：

 

ElasticSearch是一个基于Lucene的搜索服务器。它提供了一个分布式多用户能力的全文搜索引擎，基于RESTful web接口。Elasticsearch是用Java开发的，并作为Apache许可条款下的开放源码发布，是当前流行的企业级搜索引擎。设计用于云计算中，能够达到实时搜索，稳定，可靠，快速，安装使用方便。

ElasticSearch相关概念：

a）、索引index，相当于数据库中的database。
b）、类型type相当于数据库中的table。
c）、主键id相当于数据库中记录的主键，是唯一的。
d）、文档 document （相当于一条数据）
文档是ElasticSearch的基本单位。在Es中文档以JSON格式来表示
向es中的index下面的type中存储json类型的数据。
e) 、字段是文档中的field 属性，需要对每一个属性定义索引和被搜索的方式

二、ElasticSearch的安装：

1、先安装jdk

2、安装ElasticSearch

直接进入elasticsearch的官网，下载最新的安装包：https://www.elastic.co/downloads/elasticsearch，此教程使用的是5.1.1版本。

将下载的安装包上传到centos，或者直接在centos使用wget命令下载。

解压：

unzip elasticsearch-5.1.1.zip

运行：

  1. cd bin
  2. ./elasticsearch

三、java语言操作ElasticSearch：

1、maven依赖

  1.         <dependency>
  2.             <groupId>org.elasticsearch</groupId>
  3.             <artifactId>elasticsearch</artifactId>
  4.             <version>7.6.2</version>
  5.         </dependency>
  6.  
  7.         <dependency>
  8.             <groupId>org.elasticsearch.client</groupId>
  9.             <artifactId>elasticsearch-rest-high-level-client</artifactId>
  10.             <version>7.6.2</version>
  11.         </dependency>

2、连接ElasticSearch

  1. import org.apache.http.HttpHost;
  2. import org.elasticsearch.client.RestClient;
  3. import org.elasticsearch.client.RestHighLevelClient;
  4.  
  5. import java.io.IOException;
  6.  
  7. public class EsClientTest {
  8.  
  9.     public static void main(String[] args) throws IOException {
  10.         RestHighLevelClient esClient = new RestHighLevelClient(
  11.                 RestClient.builder(new HttpHost("IP",9200,"http"))
  12.         );
  13.         System.out.println("success");
  14.         esClient.close();
  15.     }
  16.  
  17. }

 3、连接的相关api

  1.     public static RestHighLevelClient esClient;
  2.  
  3.     static {
  4.         esClient = new RestHighLevelClient(
  5.                 RestClient.builder(new HttpHost("IP", 9200, "http"))
  6.         );
  7.     }

4、创建索引操作：

  1.     /**
  2.      * 创建索引
  3.      * @throws IOException
  4.      */
  5.     public static void createIndex() throws IOException {
  6.         CreateIndexRequest createIndexRequest = new CreateIndexRequest("user");
  7.         CreateIndexResponse indexResponse = esClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
  8.         boolean acknowledged = indexResponse.isAcknowledged();
  9.         System.out.println("索引创建状态:" + acknowledged);
  10.     }

5、获取索引：

  1. /**
  2.      * 索引信息查询
  3.      * @throws IOException
  4.      */
  5.     public static void getIndex() throws IOException {
  6.         GetIndexRequest getIndexRequest = new GetIndexRequest("user");
  7.         GetIndexResponse getIndexResponse = esClient.indices().get(getIndexRequest, RequestOptions.DEFAULT);
  8.         System.out.println(getIndexResponse.getAliases());
  9.         System.out.println(getIndexResponse.getMappings());
  10.         System.out.println(getIndexResponse.getSettings());
  11.     }

6、删除索引：

  1.     /**
  2.      * 删除索引
  3.      * @throws IOException
  4.      */
  5.     public static void deleteIndex() throws IOException {
  6.         DeleteIndexRequest getIndexRequest = new DeleteIndexRequest("user");
  7.         AcknowledgedResponse delete = esClient.indices().delete(getIndexRequest, RequestOptions.DEFAULT);
  8.         System.out.println("索引删除状态:" + delete.isAcknowledged());
  9.     }

7、添加数据：

  1.     /**
  2.      * 添加数据
  3.      * @throws Exception
  4.      */
  5.     public static void add() throws Exception{
  6.         IndexRequest indexRequest = new IndexRequest();
  7.         indexRequest.index("user").id("1008");
  8.         User user = new User();
  9.         user.setName("茅河野人");
  10.         user.setAge(28);
  11.         user.setSex("男");
  12.         user.setSalary(50000);
  13.  
  14.         String userData = objectMapper.writeValueAsString(user);
  15.         indexRequest.source(userData,XContentType.JSON);
  16.         //插入数据
  17.         IndexResponse response = esClient.index(indexRequest, RequestOptions.DEFAULT);
  18.         System.out.println(response.status());
  19.         System.out.println(response.getResult());
  20.     }

8、修改数据：

  1.     /**
  2.      * 修改数据
  3.      * @throws Exception
  4.      */
  5.     public static void update() throws Exception{
  6.         UpdateRequest request = new UpdateRequest();
  7.         request.index("user").id("1008");
  8.         request.doc(XContentType.JSON,"name","茅河野人");
  9.         //插入数据
  10.         UpdateResponse response = esClient.update(request, RequestOptions.DEFAULT);
  11.         System.out.println(response.getResult());
  12.     }

9、删除数据：

  1.     /**
  2.      * 删除
  3.      * @throws Exception
  4.      */
  5.     public static void delete() throws Exception{
  6.         DeleteRequest request = new DeleteRequest();
  7.         request.index("user").id("1008");
  8.         //插入数据
  9.         DeleteResponse delete = esClient.delete(request, RequestOptions.DEFAULT);
  10.         System.out.println(delete.getResult());
  11.     }

10、批量添加数据：

  1.     /**
  2.      * 批量添加
  3.      * @throws Exception
  4.      */
  5.     public static void batchInsert() throws Exception{
  6.  
  7.         BulkRequest bulkRequest = new BulkRequest();
  8.  
  9.         User user1 = new User("关羽","男",33,5500);
  10.         String userData1 = objectMapper.writeValueAsString(user1);
  11.         IndexRequest indexRequest1 = new IndexRequest().index("user").id("1002").source(userData1, XContentType.JSON);
  12.  
  13.         bulkRequest.add(indexRequest1);
  14.  
  15.         User user2 = new User("黄忠","男",50,8000);
  16.         String userData2 = objectMapper.writeValueAsString(user2);
  17.         IndexRequest indexRequest2 = new IndexRequest().index("user").id("1003").source(userData2, XContentType.JSON);
  18.         bulkRequest.add(indexRequest2);
  19.  
  20.         User user3 = new User("黄忠2","男",49,10000);
  21.         String userData3 = objectMapper.writeValueAsString(user3);
  22.         IndexRequest indexRequest3 = new IndexRequest().index("user").id("1004").source(userData3, XContentType.JSON);
  23.         bulkRequest.add(indexRequest3);
  24.  
  25.         User user4 = new User("赵云","男",33,12000);
  26.         String userData4 = objectMapper.writeValueAsString(user4);
  27.         IndexRequest indexRequest4 = new IndexRequest().index("user").id("1005").source(userData4, XContentType.JSON);
  28.         bulkRequest.add(indexRequest4);
  29.  
  30.         User user5 = new User("马超","男",38,20000);
  31.         String userData5 = objectMapper.writeValueAsString(user5);
  32.         IndexRequest indexRequest5 = new IndexRequest().index("user").id("1006").source(userData5, XContentType.JSON);
  33.         bulkRequest.add(indexRequest5);
  34.  
  35.         User user6 = new User("关羽","男",41,27000);
  36.         String userData6 = objectMapper.writeValueAsString(user6);
  37.         IndexRequest indexRequest6 = new IndexRequest().index("user").id("1007").source(userData6, XContentType.JSON);
  38.         bulkRequest.add(indexRequest6);
  39.  
  40.         BulkResponse bulkResponse = esClient.bulk(bulkRequest, RequestOptions.DEFAULT);
  41.         System.out.println(bulkResponse.status());
  42.         System.out.println(bulkResponse.getItems());
  43.     }

11、批量删除数据：

  1.     /**
  2.      * 批量删除
  3.      * @throws Exception
  4.      */
  5.     public static void batchDelete() throws Exception{
  6.         BulkRequest bulkRequest = new BulkRequest();
  7.         DeleteRequest indexRequest1 = new DeleteRequest().index("user").id("1002");
  8.         DeleteRequest indexRequest2 = new DeleteRequest().index("user").id("1003");
  9.         DeleteRequest indexRequest3 = new DeleteRequest().index("user").id("1004");
  10.         DeleteRequest indexRequest4 = new DeleteRequest().index("user").id("1005");
  11.         DeleteRequest indexRequest5 = new DeleteRequest().index("user").id("1006");
  12.         DeleteRequest indexRequest6 = new DeleteRequest().index("user").id("1007");
  13.  
  14.         bulkRequest.add(indexRequest1);
  15.         bulkRequest.add(indexRequest2);
  16.         bulkRequest.add(indexRequest3);
  17.         bulkRequest.add(indexRequest4);
  18.         bulkRequest.add(indexRequest5);
  19.         bulkRequest.add(indexRequest6);
  20.  
  21.         BulkResponse bulkResponse = esClient.bulk(bulkRequest, RequestOptions.DEFAULT);
  22.         System.out.println(bulkResponse.status());
  23.         System.out.println(bulkResponse.getItems());
  24.     }

13、删除某个索引下所有数据：

  1.     /**
  2.      * 查询某个索引下的所有数据
  3.      * @throws Exception
  4.      */
  5.     public static void searchIndexAll() throws Exception{
  6.         SearchRequest request = new SearchRequest();
  7.         request.indices("user");
  8.         // 索引中的全部数据查询
  9.         SearchSourceBuilder query = new SearchSourceBuilder().query(QueryBuilders.matchAllQuery());
  10.         request.source(query);
  11.         SearchResponse response = esClient.search(request, RequestOptions.DEFAULT);
  12.         SearchHits hits = response.getHits();
  13.         for (SearchHit searchHit : hits){
  14.             System.out.println(searchHit.getSourceAsString());
  15.         }
  16.     }

 14、根据条件查询：

  1.         TermQueryBuilder ageQueryBuilder = QueryBuilders.termQuery("sex", "女");
  2.         SearchSourceBuilder query = new SearchSourceBuilder().query(ageQueryBuilder);
  3.         request.source(query);
  4.         SearchResponse response = esClient.search(request, RequestOptions.DEFAULT);
  5.         System.out.println(response.getHits().getHits());
  6.         System.out.println(response.getHits().getTotalHits());
  7.         SearchHits hits = response.getHits();
  8.         for (SearchHit searchHit : hits){
  9.             System.out.println(searchHit.getSourceAsString());
  10.         }

15、分页查询：

  1.         SearchSourceBuilder sourceBuilder = new                         
  2.         SearchSourceBuilder().query(QueryBuilders.matchAllQuery());
  3.         sourceBuilder.from(0).size(3);
  4.         request.source(sourceBuilder);
  5.         SearchResponse response = esClient.search(request, RequestOptions.DEFAULT);
  6.         System.out.println(response.getHits().getHits());
  7.         System.out.println(response.getHits().getTotalHits());
  8.         SearchHits hits = response.getHits();
  9.         for (SearchHit searchHit : hits){
  10.             System.out.println(searchHit.getSourceAsString());
  11.         }

四、在springboot中的运用

1、maven依赖

  1.      <dependency>
  2.             <groupId>org.springframework.boot</groupId>
  3.             <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
  4.       </dependency>
  5.

2、yml配置文件：

  1. # es 服务地址
  2. elasticsearch.host=IP
  3. # es 服务端口
  4. elasticsearch.port=9200
  5. # 配置日志级别,开启 debug 日志
  6. logging.level.com.congge=debug

3、实际例子：

创建一个实体类

  1. import lombok.AllArgsConstructor;
  2. import lombok.Data;
  3. import lombok.NoArgsConstructor;
  4. import lombok.ToString;
  5. import org.springframework.data.annotation.Id;
  6. import org.springframework.data.elasticsearch.annotations.Document;
  7. import org.springframework.data.elasticsearch.annotations.Field;
  8. import org.springframework.data.elasticsearch.annotations.FieldType;
  9.  
  10. @Data
  11. @NoArgsConstructor
  12. @AllArgsConstructor
  13. @ToString
  14. @Document(indexName = "shopping", shards = 3, replicas = 1)
  15. public class Product {
  16.     //必须有 id,这里的 id 是全局唯一的标识，等同于 es 中的"_id"
  17.     @Id
  18.     private Long id;//商品唯一标识
  19.  
  20.     /**
  21.      * type : 字段数据类型
  22.      * analyzer : 分词器类型
  23.      * index : 是否索引(默认:true)
  24.      * Keyword : 短语,不进行分词
  25.      */
  26.     @Field(type = FieldType.Text, analyzer = "ik_max_word")
  27.     private String title;//商品名称
  28.  
  29.     @Field(type = FieldType.Keyword)
  30.     private String category;//分类名称
  31.  
  32.     @Field(type = FieldType.Double)
  33.     private Double price;//商品价格
  34.  
  35.     @Field(type = FieldType.Keyword, index = false)
  36.     private String images;//图片地址
  37. }

提供接口：

  1. import com.congge.entity.Product;
  2. import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
  3. import org.springframework.stereotype.Repository;
  4.  
  5. @Repository
  6. public interface ProductDao extends ElasticsearchRepository<Product, Long>{
  7.  
  8. }

配置类：

  1. import lombok.Data;
  2. import org.apache.http.HttpHost;
  3. import org.elasticsearch.client.RestClient;
  4. import org.elasticsearch.client.RestClientBuilder;
  5. import org.elasticsearch.client.RestHighLevelClient;
  6. import org.springframework.boot.context.properties.ConfigurationProperties;
  7. import org.springframework.context.annotation.Configuration;
  8. //import org.springframework.data.elasticsearch.config.AbstractElasticsearchConfiguration;
  9.  
  10. @ConfigurationProperties(prefix = "elasticsearch")
  11. @Configuration
  12. @Data
  13. public class EsConfig extends com.congge.config.AbstractElasticsearchConfiguration {
  14.  
  15.     private String host ;
  16.     private Integer port ;
  17.  
  18.     //重写父类方法
  19.     @Override
  20.     public RestHighLevelClient elasticsearchClient() {
  21.         RestClientBuilder builder = RestClient.builder(new HttpHost(host, port));
  22.         RestHighLevelClient restHighLevelClient = new
  23.                 RestHighLevelClient(builder);
  24.         return restHighLevelClient;
  25.     }
  26. }
  1. import org.elasticsearch.client.RestHighLevelClient;
  2. import org.springframework.context.annotation.Bean;
  3. import org.springframework.data.elasticsearch.config.ElasticsearchConfigurationSupport;
  4. import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
  5. import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
  6. import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;
  7.  
  8. public abstract class AbstractElasticsearchConfiguration extends ElasticsearchConfigurationSupport {
  9.  
  10.     //需重写本方法
  11.     public abstract RestHighLevelClient elasticsearchClient();
  12.  
  13.     @Bean(name = { "elasticsearchOperations", "elasticsearchTemplate" })
  14.     public ElasticsearchOperations elasticsearchOperations(ElasticsearchConverter elasticsearchConverter) {
  15.         return new ElasticsearchRestTemplate(elasticsearchClient(), elasticsearchConverter);
  16.     }
  17. }

测试1：

  1. import com.congge.entity.Product;
  2. import org.junit.Test;
  3. import org.junit.runner.RunWith;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.boot.test.context.SpringBootTest;
  6. import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
  7. import org.springframework.test.context.junit4.SpringRunner;
  8.  
  9. @RunWith(SpringRunner.class)
  10. @SpringBootTest
  11. public class EsIndexTest {
  12.  
  13.     //注入 ElasticsearchRestTemplate
  14.     @Autowired
  15.     private ElasticsearchRestTemplate elasticsearchRestTemplate;
  16.  
  17.     //创建索引并增加映射配置
  18.     @Test
  19.     public void createIndex(){
  20.         //创建索引，系统初始化会自动创建索引
  21.         System.out.println("创建索引");
  22.     }
  23.  
  24.     @Test
  25.     public void deleteIndex(){
  26.         //创建索引，系统初始化会自动创建索引
  27.         boolean flg = elasticsearchRestTemplate.deleteIndex(Product.class);
  28.         System.out.println("删除索引 = " + flg);
  29.     }
  30.  
  31. }

测试2：

  1. import com.congge.dao.ProductDao;
  2. import com.congge.entity.Product;
  3. import org.elasticsearch.index.query.QueryBuilders;
  4. import org.elasticsearch.index.query.TermQueryBuilder;
  5. import org.junit.Test;
  6. import org.junit.runner.RunWith;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.boot.test.context.SpringBootTest;
  9. import org.springframework.data.domain.Page;
  10. import org.springframework.data.domain.PageRequest;
  11. import org.springframework.data.domain.Sort;
  12. import org.springframework.test.context.junit4.SpringRunner;
  13.  
  14. import java.util.ArrayList;
  15. import java.util.List;
  16.  
  17. @RunWith(SpringRunner.class)
  18. @SpringBootTest
  19. public class EsDocTest {
  20.  
  21.  
  22.     @Autowired
  23.     private ProductDao productDao;
  24.  
  25.     /**
  26.      * 新增
  27.      */
  28.     @Test
  29.     public void save() {
  30.         Product product = new Product();
  31.         product.setId(2L);
  32.         product.setTitle("ipad mini");
  33.         product.setCategory("ipad");
  34.         product.setPrice(1998.0);
  35.         product.setImages("http://ipad.jpg");
  36.         productDao.save(product);
  37.     }
  38.  
  39.  
  40.     //修改
  41.     @Test
  42.     public void update(){
  43.         Product product = new Product();
  44.         product.setId(2L);
  45.         product.setTitle("iphone");
  46.         product.setCategory("mobile");
  47.         product.setPrice(6999.0);
  48.         product.setImages("http://www.phone.jpg");
  49.         productDao.save(product);
  50.     }
  51.  
  52.     //根据 id 查询
  53.     @Test
  54.     public void findById(){
  55.         Product product = productDao.findById(2L).get();
  56.         System.out.println(product);
  57.     }
  58.  
  59.     //查询所有
  60.     @Test
  61.     public void findAll(){
  62.         Iterable<Product> products = productDao.findAll();
  63.         for (Product product : products) {
  64.             System.out.println(product);
  65.         }
  66.     }
  67.  
  68.     //删除
  69.     @Test
  70.     public void delete(){
  71.         Product product = new Product();
  72.         product.setId(2L);
  73.         productDao.delete(product);
  74.     }
  75.  
  76.     //批量新增
  77.     @Test
  78.     public void saveAll(){
  79.         List<Product> productList = new ArrayList<>();
  80.         for (int i = 0; i < 10; i++) {
  81.             Product product = new Product();
  82.             product.setId(Long.valueOf(i));
  83.             product.setTitle("iphone" + i);
  84.             product.setCategory("mobile");
  85.             product.setPrice(5999.0 + i);
  86.             product.setImages("http://www.phone.jpg");
  87.             productList.add(product);
  88.         }
  89.         productDao.saveAll(productList);
  90.     }
  91.  
  92.     //分页查询
  93.     @Test
  94.     public void findByPageable(){
  95.         //设置排序(排序方式，正序还是倒序，排序的 id)
  96.         Sort sort = Sort.by(Sort.Direction.DESC,"id");
  97.         int currentPage=0;//当前页，第一页从 0 开始， 1 表示第二页
  98.         int pageSize = 5;//每页显示多少条
  99.         //设置查询分页
  100.         PageRequest pageRequest = PageRequest.of(currentPage, pageSize,sort);
  101.         //分页查询
  102.         Page<Product> productPage = productDao.findAll(pageRequest);
  103.         for (Product Product : productPage.getContent()) {
  104.             System.out.println(Product);
  105.         }
  106.     }
  107.  
  108.     /**
  109.      * term 查询
  110.      * search(termQueryBuilder) 调用搜索方法，参数查询构建器对象
  111.      */
  112.     @Test
  113.     public void termQuery(){
  114.         TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery("title", "iphone");
  115.         Iterable<Product> products = productDao.search(termQueryBuilder);
  116.         for (Product product : products) {
  117.             System.out.println(product);
  118.         }
  119.     }
  120.  
  121.     /**
  122.      * term 查询加分页
  123.      */
  124.     @Test
  125.     public void termQueryByPage(){
  126.         int currentPage= 0 ;
  127.         int pageSize = 5;
  128.         //设置查询分页
  129.         PageRequest pageRequest = PageRequest.of(currentPage, pageSize);
  130.         TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery("title", "phone");
  131.         Iterable<Product> products =
  132.                 productDao.search(termQueryBuilder,pageRequest);
  133.         for (Product product : products) {
  134.             System.out.println(product);
  135.         }
  136.     }
  137.  
  138. }

五、将mysql数据写入Elasticsearch例子

  1. package com.example.esdemo.service.impl;
  2.  
  3. import com.example.esdemo.config.DBHelper;
  4. import com.example.esdemo.imports.ImportDb2Es;
  5. import com.example.esdemo.service.ImportService;
  6. import org.apache.logging.log4j.LogManager;
  7. import org.apache.logging.log4j.Logger;
  8. import org.elasticsearch.action.ActionListener;
  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.index.IndexRequest;
  14. import org.elasticsearch.client.RequestOptions;
  15. import org.elasticsearch.client.RestHighLevelClient;
  16. import org.elasticsearch.common.unit.ByteSizeUnit;
  17. import org.elasticsearch.common.unit.ByteSizeValue;
  18. import org.elasticsearch.common.unit.TimeValue;
  19. import org.springframework.beans.factory.annotation.Autowired;
  20. import org.springframework.stereotype.Component;
  21.  
  22. import java.sql.*;
  23. import java.util.ArrayList;
  24. import java.util.HashMap;
  25. import java.util.concurrent.TimeUnit;
  26. import java.util.function.BiConsumer;
  27.  
  28.  
  29. /**
  30.  * 导入db2es 实现类
  31.  */
  32. @Component
  33. public class ImportServiceImpl implements ImportService {
  34.  
  35.     private static final Logger logger = LogManager.getLogger(ImportServiceImpl.class);
  36.  
  37.     @Autowired
  38.     private RestHighLevelClient client;
  39.  
  40.  
  41.     @Override
  42.     public void importDb2Es(ImportDb2Es importDb2Es) {
  43.         writeMySQLDataToES(importDb2Es.getDbTableName(),importDb2Es.getDbTableName());
  44.     }
  45.  
  46.  
  47.     private void writeMySQLDataToES(String tableName,String esIndeName) {
  48.         BulkProcessor bulkProcessor = getBulkProcessor(client);
  49.         Connection connection = null;
  50.         PreparedStatement ps = null;
  51.         ResultSet rs = null;
  52.         try {
  53.             connection = DBHelper.getConn();
  54.             logger.info("start handle data :" + tableName);
  55.             String sql = "select * from " + tableName;
  56.             ps = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
  57.             // 根据自己需要设置 fetchSize
  58.             ps.setFetchSize(20);
  59.             rs = ps.executeQuery();
  60.             ResultSetMetaData colData = rs.getMetaData();
  61.             ArrayList<HashMap<String, String>> dataList = new ArrayList<>();
  62.             HashMap<String, String> map = null;
  63.             int count = 0;
  64.             // c 就是列的名字   v 就是列对应的值
  65.             String c = null;
  66.             String v = null;
  67.             while (rs.next()) {
  68.                 count++;
  69.                 map = new HashMap<String, String>(128);
  70.                 for (int i = 1; i < colData.getColumnCount(); i++) {
  71.                     c = colData.getColumnName(i);
  72.                     v = rs.getString(c);
  73.                     map.put(c, v);
  74.                 }
  75.                 dataList.add(map);
  76.                 // 每1万条 写一次   不足的批次的数据 最后一次提交处理
  77.                 if (count % 10000 == 0) {
  78.                     logger.info("mysql handle data  number:" + count);
  79.                     // 将数据添加到 bulkProcessor
  80.                     for (HashMap<String, String> hashMap2 : dataList) {
  81.                         bulkProcessor.add(new IndexRequest(esIndeName).source(hashMap2));
  82.                     }
  83.                     // 每提交一次 清空 map 和  dataList
  84.                     map.clear();
  85.                     dataList.clear();
  86.                 }
  87.             }
  88.             // 处理 未提交的数据
  89.             for (HashMap<String, String> hashMap2 : dataList) {
  90.                 bulkProcessor.add(new IndexRequest(esIndeName).source(hashMap2));
  91.             }
  92.             bulkProcessor.flush();
  93.  
  94.         } catch (SQLException e) {
  95.             e.printStackTrace();
  96.         } finally {
  97.             try {
  98.                 rs.close();
  99.                 ps.close();
  100.                 connection.close();
  101.                 boolean terinaFlag = bulkProcessor.awaitClose(150L, TimeUnit.SECONDS);
  102.                 logger.info(terinaFlag);
  103.             } catch (Exception e) {
  104.                 e.printStackTrace();
  105.             }
  106.         }
  107.  
  108.     }
  109.  
  110.     private BulkProcessor getBulkProcessor(RestHighLevelClient client) {
  111.  
  112.         BulkProcessor bulkProcessor = null;
  113.         try {
  114.  
  115.             BulkProcessor.Listener listener = new BulkProcessor.Listener() {
  116.                 @Override
  117.                 public void beforeBulk(long executionId, BulkRequest request) {
  118.                     logger.info("Try to insert data number : "
  119.                             + request.numberOfActions());
  120.                 }
  121.  
  122.                 @Override
  123.                 public void afterBulk(long executionId, BulkRequest request,
  124.                                       BulkResponse response) {
  125.                     logger.info("************** Success insert data number : "
  126.                             + request.numberOfActions() + " , id: " + executionId);
  127.                 }
  128.  
  129.                 @Override
  130.                 public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
  131.                     logger.error("Bulk is unsuccess : " + failure + ", executionId: " + executionId);
  132.                 }
  133.             };
  134.  
  135.             BiConsumer<BulkRequest, ActionListener<BulkResponse>> bulkConsumer = (request, bulkListener) -> client
  136.                     .bulkAsync(request, RequestOptions.DEFAULT, bulkListener);
  137.  
  138.             BulkProcessor.Builder builder = BulkProcessor.builder(bulkConsumer, listener);
  139.             builder.setBulkActions(5000);
  140.             builder.setBulkSize(new ByteSizeValue(100L, ByteSizeUnit.MB));
  141.             builder.setConcurrentRequests(10);
  142.             builder.setFlushInterval(TimeValue.timeValueSeconds(100L));
  143.             builder.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1L), 3));
  144.             // 注意点：让参数设置生效
  145.             bulkProcessor = builder.build();
  146.  
  147.         } catch (Exception e) {
  148.             e.printStackTrace();
  149.             try {
  150.                 bulkProcessor.awaitClose(100L, TimeUnit.SECONDS);
  151.             } catch (Exception e1) {
  152.                 logger.error(e1.getMessage());
  153.             }
  154.         }
  155.         return bulkProcessor;
  156.     }
  157. }
  158.

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

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

  

闽ICP备14008679号