发布于 由admin

Spring Boot整合Elasticsearch写入延迟高的批量优化方案

Spring Boot整合Elasticsearch写入延迟高的批量优化方案

大家好，今天我们来聊聊在使用Spring Boot整合Elasticsearch时，遇到的批量写入延迟高的问题以及相应的优化方案。这个问题在数据量较大的场景下尤为突出，直接影响系统的性能和用户体验。

一、问题分析：为什么批量写入会慢？

在深入优化之前，我们需要了解导致批量写入缓慢的几个主要原因：

  1. 网络延迟： Spring Boot应用和Elasticsearch集群之间的网络延迟是不可避免的。每一次请求都需要经过网络传输，批量写入实际上是将多次单个写入操作合并成一次，但如果网络状况不佳，整体延迟仍然会很高。

  2. 序列化/反序列化开销： Java对象需要序列化成JSON格式才能发送给Elasticsearch，而Elasticsearch接收到JSON后需要反序列化成其内部的数据结构。频繁的序列化/反序列化操作会消耗大量的CPU资源。

  3. Elasticsearch的资源限制： Elasticsearch集群的资源（CPU、内存、磁盘I/O）是有限的。如果集群负载过高，写入性能自然会下降。此外，Elasticsearch的线程池、队列大小等配置也可能成为瓶颈。

  4. refresh策略： Elasticsearch默认的refresh策略是refresh_interval: 1s，这意味着数据写入后需要1秒钟才能被搜索到。每次refresh都会消耗大量的资源。

  5. 索引设置不合理： 例如，索引的分片数量过多或过少，字段类型设置不当，使用了不必要的analyzer等，都会影响写入性能。

  6. 数据量和复杂度： 写入的数据量越大，字段越多，字段类型越复杂，Elasticsearch的处理时间自然越长。

  7. 硬件限制： 磁盘I/O是写入速度的一个重要瓶颈。如果使用传统的机械硬盘，写入速度会受到很大的限制。

二、优化方案：多管齐下提升性能

针对上述问题，我们可以采取一系列的优化措施，提高批量写入的效率。

  1. 调整Elasticsearch的配置

    • 刷新间隔(refresh_interval)

      Elasticsearch的refresh操作负责将内存中的数据刷新到磁盘，使其可以被搜索到。频繁的refresh会消耗大量的资源。对于写入量大但实时性要求不高的场景，可以适当增加refresh间隔，甚至设置为-1禁用自动refresh，然后在写入完成后手动refresh。

      PUT /your_index/_settings
{
  "index": {
    "refresh_interval": "-1"
  }
}

      写入完成后手动refresh：

      @Autowired
private RestHighLevelClient client;

public void refreshIndex(String indexName) throws IOException {
  RefreshRequest request = new RefreshRequest(indexName);
  RefreshResponse response = client.indices().refresh(request, RequestOptions.DEFAULT);
  if (response.getShardFailures().length > 0) {
    // 处理refresh失败的情况
    System.err.println("Refresh failed: " + Arrays.toString(response.getShardFailures()));
  }
}

// 在批量写入完成后调用refreshIndex方法
refreshIndex("your_index");

    • 调整Bulk请求的大小

      Bulk请求的大小会影响Elasticsearch的处理效率。过小的Bulk请求会导致频繁的网络交互，而过大的Bulk请求可能会导致内存溢出。需要根据实际情况调整Bulk请求的大小，找到一个平衡点。一般来说，每个Bulk请求包含几百到几千条数据是比较合适的。

      在Spring Data Elasticsearch中，可以通过配置bulkSize属性来控制Bulk请求的大小。

      @Configuration
public class ElasticsearchConfig {

    @Bean
    public ElasticsearchOperations elasticsearchTemplate(RestHighLevelClient client) {
        ElasticsearchRestTemplate template = new ElasticsearchRestTemplate(client);
        // 设置Bulk请求的大小，例如500
        template.setBulkSize(500);
        return template;
    }
}

    • 调整translog的设置

      Elasticsearch的translog用于保证数据的可靠性。默认情况下，每次index、bulk、delete等操作都会立即flush translog到磁盘，以防止数据丢失。这种机制虽然保证了数据的安全性，但也会降低写入性能。

      可以调整index.translog.durability参数来控制translog的flush策略。

      • request: 每次请求后flush (默认值，最安全，但性能最差)
      • async: 异步flush (性能较好，但可能会丢失数据)
      PUT /your_index/_settings
{
  "index": {
    "translog": {
      "durability": "async",
      "sync_interval": "5s" // 异步flush的间隔时间，默认为5秒
    }
  }
}

      注意： 将index.translog.durability设置为async会降低数据的安全性，需要根据实际情况权衡。

    • 禁用refresh interval和调整translog参数的影响:

      配置项 默认值 修改建议 影响
      index.refresh_interval 1s 增加到30s或-1（禁用），然后在批量写入后手动refresh 提升写入速度： 显著提升。 数据可见性： 延迟数据可见性，直到手动refresh。 适用场景： 适用于非实时性要求高的批量写入场景。 风险： 如果Elasticsearch崩溃，未refresh的数据会丢失。
      index.translog.durability request 修改为async 提升写入速度： 显著提升。 数据安全性： 降低数据安全性，因为translog的flush是异步的。 适用场景： 适用于对数据安全性要求不高的场景。 风险： 如果Elasticsearch崩溃，最近一次异步flush间隔内的数据可能会丢失。

    • 调整索引缓冲区大小

      indices.memory.index_buffer_size参数控制用于索引的内存缓冲区大小。默认情况下，Elasticsearch会将JVM堆内存的10%分配给所有节点上的所有索引缓冲区。可以适当增加这个值，以提高索引性能。

      # elasticsearch.yml
indices.memory.index_buffer_size: 30%

      注意： 不要将indices.memory.index_buffer_size设置得过大，否则可能会导致JVM内存溢出。

    • 增加分片数量

      增加分片数量可以提高写入的并行度，从而提高写入性能。但是，过多的分片也会增加资源消耗。需要根据集群的规模和数据量来选择合适的分片数量。一般来说，每个分片的大小在30-50GB之间是比较合适的。

      可以在创建索引时指定分片数量：

      PUT /your_index
{
  "settings": {
    "number_of_shards": 5,
    "number_of_replicas": 1
  }
}

    • 禁用不必要的分析器(Analyzer)

      如果某些字段不需要进行分词分析，可以将其index属性设置为false，或者使用keyword类型。这样可以减少Elasticsearch的计算量，提高写入性能。

      PUT /your_index/_mapping
{
  "properties": {
    "field1": {
      "type": "text",
      "analyzer": "standard"
    },
    "field2": {
      "type": "keyword" // 使用keyword类型，不进行分词
    },
    "field3": {
      "type": "text",
      "index": false // 不进行索引
    }
  }
}

  2. 优化Spring Boot代码

    • 使用RestHighLevelClient

      RestHighLevelClient是Elasticsearch官方推荐的Java客户端，相比于TransportClient，它更加稳定和高效。

      在Spring Boot中，可以通过以下方式注入RestHighLevelClient：

      @Configuration
public class ElasticsearchConfig {

    @Bean
    public RestHighLevelClient client(@Value("${elasticsearch.host}") String host,
                                     @Value("${elasticsearch.port}") int port) {
        RestClientBuilder builder = RestClient.builder(
                new HttpHost(host, port, "http"));
        return new RestHighLevelClient(builder);
    }
}

    • 使用BulkRequest进行批量写入

      BulkRequest可以将多个index、update、delete操作合并成一个请求发送给Elasticsearch，减少网络交互的次数。

      @Autowired
private RestHighLevelClient client;

public void bulkInsert(List<YourData> dataList, String indexName) throws IOException {
    BulkRequest request = new BulkRequest();
    for (YourData data : dataList) {
        IndexRequest indexRequest = new IndexRequest(indexName)
                .source(objectMapper.writeValueAsString(data), XContentType.JSON);
        request.add(indexRequest);
    }

    BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);

    if (bulkResponse.hasFailures()) {
        // 处理bulk写入失败的情况
        for (BulkItemResponse bulkItemResponse : bulkResponse) {
            if (bulkItemResponse.isFailed()) {
                BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                System.err.println("Error indexing document: " + failure);
            }
        }
    }
}

    • 使用线程池进行并发写入

      可以使用线程池将批量写入任务分解成多个子任务，并发执行，以提高写入速度。

      @Autowired
private RestHighLevelClient client;

private final ExecutorService executor = Executors.newFixedThreadPool(10); // 创建一个固定大小的线程池

public void bulkInsertAsync(List<YourData> dataList, String indexName) {
    List<List<YourData>> partitions = Lists.partition(dataList, 1000); // 将数据分成多个小批次

    for (List<YourData> partition : partitions) {
        executor.submit(() -> {
            try {
                bulkInsert(partition, indexName); // 调用上面的bulkInsert方法
            } catch (IOException e) {
                System.err.println("Error during bulk insert: " + e.getMessage());
            }
        });
    }
}

public void shutdownExecutor() {
    executor.shutdown();
    try {
        if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
            executor.shutdownNow();
        }
    } catch (InterruptedException e) {
        executor.shutdownNow();
    }
}

// 使用示例：
// bulkInsertAsync(yourDataList, "your_index");
// shutdownExecutor(); // 在程序结束时关闭线程池

    • 优化序列化过程

      选择合适的JSON序列化库，例如Jackson或Gson，并进行优化配置，可以减少序列化/反序列化的开销。

      例如，使用Jackson时，可以禁用自动检测功能，避免不必要的属性扫描。

      ObjectMapper objectMapper = new ObjectMapper();
objectMapper.setVisibility(PropertyAccessor.FIELD, Visibility.ANY); // 允许访问私有字段
objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); // 忽略未知属性

    • 减少不必要的字段

      只存储需要的字段，避免存储冗余数据，可以减少Elasticsearch的存储空间和处理时间。

    • 批量删除使用BulkRequest

      @Autowired
private RestHighLevelClient client;

public void bulkDelete(List<String> ids, String indexName) throws IOException {
    BulkRequest request = new BulkRequest();
    for (String id : ids) {
        DeleteRequest deleteRequest = new DeleteRequest(indexName, id);
        request.add(deleteRequest);
    }

    BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);

    if (bulkResponse.hasFailures()) {
        // 处理bulk删除失败的情况
        for (BulkItemResponse bulkItemResponse : bulkResponse) {
            if (bulkItemResponse.isFailed()) {
                BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                System.err.println("Error deleting document: " + failure);
            }
        }
    }
}

    • 监控和分析

      使用Elasticsearch的监控工具（例如Kibana）和Spring Boot Actuator监控应用的性能，及时发现和解决问题。

  3. 硬件升级

    如果以上优化措施都无法满足需求，可以考虑升级硬件，例如使用SSD硬盘，增加CPU和内存。

  4. 代码示例：完整的Spring Boot整合Elasticsearch批量写入示例

import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.http.HttpHost;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@SpringBootApplication
public class ElasticsearchBulkExampleApplication {

    public static void main(String[] args) {
        SpringApplication.run(ElasticsearchBulkExampleApplication.class, args);
    }

    // 模拟数据类
    static class YourData {
        private String id;
        private String name;
        private String description;

        public YourData(String id, String name, String description) {
            this.id = id;
            this.name = name;
            this.description = description;
        }

        public String getId() {
            return id;
        }

        public String getName() {
            return name;
        }

        public String getDescription() {
            return description;
        }

        public void setId(String id) {
            this.id = id;
        }

        public void setName(String name) {
            this.name = name;
        }

        public void setDescription(String description) {
            this.description = description;
        }
    }

    @Configuration
    static class ElasticsearchConfig {

        @Value("${elasticsearch.host}")
        private String host;

        @Value("${elasticsearch.port}")
        private int port;

        @Bean(destroyMethod = "close")
        public RestHighLevelClient client() {
            return new RestHighLevelClient(
                    RestClient.builder(new HttpHost(host, port, "http"))
            );
        }

        @Bean
        public ObjectMapper objectMapper() {
            ObjectMapper objectMapper = new ObjectMapper();
            objectMapper.setVisibility(PropertyAccessor.FIELD, Visibility.ANY);
            objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
            return objectMapper;
        }
    }

    @Service
    static class ElasticsearchBulkService {

        @Autowired
        private RestHighLevelClient client;

        @Autowired
        private ObjectMapper objectMapper;

        private final ExecutorService executor = Executors.newFixedThreadPool(10);

        public void bulkInsertAsync(List<YourData> dataList, String indexName) {
            // 将数据分成多个小批次
            int batchSize = 1000;
            for (int i = 0; i < dataList.size(); i += batchSize) {
                int end = Math.min(i + batchSize, dataList.size());
                List<YourData> partition = dataList.subList(i, end);

                executor.submit(() -> {
                    try {
                        bulkInsert(partition, indexName);
                    } catch (IOException e) {
                        System.err.println("Error during bulk insert: " + e.getMessage());
                    }
                });
            }
        }

        public void bulkInsert(List<YourData> dataList, String indexName) throws IOException {
            BulkRequest request = new BulkRequest();
            for (YourData data : dataList) {
                IndexRequest indexRequest = new IndexRequest(indexName)
                        .source(objectMapper.writeValueAsString(data), XContentType.JSON);
                request.add(indexRequest);
            }

            BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);

            if (bulkResponse.hasFailures()) {
                for (BulkItemResponse bulkItemResponse : bulkResponse) {
                    if (bulkItemResponse.isFailed()) {
                        BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                        System.err.println("Error indexing document: " + failure);
                    }
                }
            } else {
              System.out.println("Bulk insert success, took " + bulkResponse.getTook());
            }
        }

        public void shutdownExecutor() {
            executor.shutdown();
            try {
                if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                    executor.shutdownNow();
                }
            } catch (InterruptedException e) {
                executor.shutdownNow();
            }
        }
    }

    // 使用示例
    @org.springframework.boot.CommandLineRunner
    public class MyRunner implements org.springframework.boot.CommandLineRunner {

        @Autowired
        private ElasticsearchBulkService bulkService;

        @Override
        public void run(String... args) throws Exception {
            // 准备模拟数据
            List<YourData> dataList = new ArrayList<>();
            for (int i = 0; i < 5000; i++) {
                dataList.add(new YourData("id-" + i, "name-" + i, "description-" + i));
            }

            // 使用批量异步插入
            long startTime = System.currentTimeMillis();
            bulkService.bulkInsertAsync(dataList, "your_index"); // 替换为你的索引名称

            // 等待所有任务完成并关闭线程池
            bulkService.shutdownExecutor();
            while (!bulkService.executor.isTerminated()) {
                // 等待所有任务完成
            }
            long endTime = System.currentTimeMillis();
            System.out.println("Total time taken: " + (endTime - startTime) + "ms");
        }
    }
}

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.17</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.example</groupId>
    <artifactId>elasticsearch-bulk-example</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>elasticsearch-bulk-example</name>
    <description>elasticsearch-bulk-example</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

        <!-- Elasticsearch REST High Level Client -->
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>7.17.11</version>
        </dependency>

        <!-- Elasticsearch REST Client -->
        <dependency>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch</artifactId>
            <version>7.17.11</version>
        </dependency>

        <!-- Jackson for JSON serialization/deserialization -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

application.properties

elasticsearch.host=localhost
elasticsearch.port=9200

三、总结与建议

本次分享主要讨论了Spring Boot整合Elasticsearch批量写入延迟高的问题，并从Elasticsearch配置、Spring Boot代码和硬件升级三个方面提出了优化方案。关键在于找到性能瓶颈，并针对性地进行优化。根据实际业务场景和数据特点，灵活运用这些优化措施，才能有效地提高批量写入的效率。

四、实践是检验真理的唯一标准

理论知识固然重要，但实践才是检验真理的唯一标准。建议大家在实际项目中尝试这些优化方案，并根据实际效果进行调整，找到最适合自己的解决方案。

五、持续监控，及时调整

系统的性能是一个动态变化的过程，需要持续监控，及时发现和解决问题。只有这样，才能保证系统的稳定性和高效性。

六、综合优化，提升整体性能

优化是一个系统工程，需要从多个方面入手，综合考虑。只有这样，才能最大程度地提升整体性能。

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注