分片导出是一种常见的处理大数据量导出的方法，通过将数据分成多个片段（分片），然后使用多个线程分别处理不同的数据段，最后将结果合并。这种方法可以充分利用多核 CPU 的性能，提高导出效率。

以下是基于 CompletableFuture 和线程池的分片导出实现示例：

实现思路

1. 数据分片：将总数据量分成多个片段（分片），每个片段由一个线程处理。
2. 异步处理：使用 CompletableFuture 异步处理每个分片的数据。
3. 结果合并：将每个分片的处理结果写入文件或合并到最终结果中。
4. 线程池管理：使用线程池控制并发线程数量，避免资源耗尽。

代码实现

import java.io.FileWriter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
​
public class ShardedExcelExporter {
​
    private static final int TOTAL_RECORDS = 10000; // 总记录数
    private static final int SHARD_SIZE = 1000; // 每个分片的大小
    private static final int THREAD_POOL_SIZE = 10; // 线程池大小
    private static final ExecutorService executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE); // 线程池
​
    public static void main(String[] args) {
        DataService dataService = new DataService();
        String filePath = "Sharded_output.csv";
​
        try (FileWriter writer = new FileWriter(filePath)) {
            // 写入表头
            writer.write("ID,Data\n");
​
            // 计算分片数量
            int shardCount = (TOTAL_RECORDS + SHARD_SIZE - 1) / SHARD_SIZE;
​
            // 创建分片任务
            List<CompletableFuture<Void>> futures = new ArrayList<>();
            for (int shardIndex = 0; shardIndex < shardCount; shardIndex++) {
                int start = shardIndex * SHARD_SIZE;
                int end = Math.min(start + SHARD_SIZE, TOTAL_RECORDS);
​
                // 异步处理每个分片
                CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                    List<String> data = dataService.fetchData(start, end - start);
                    writeDataToFile(writer, data, start);
                }, executor);
​
                futures.add(future);
            }
​
            // 等待所有分片任务完成
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
​
            System.out.println("导出完成！");
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            executor.shutdown(); // 关闭线程池
        }
    }
​
    /**
     * 将数据写入文件
     */
    private static void writeDataToFile(FileWriter writer, List<String> data, int startOffset) {
        try {
            for (int i = 0; i < data.size(); i++) {
                writer.write((startOffset + i + 1) + "," + data.get(i) + "\n");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
​
/**
 * 模拟数据服务
 */
class DataService {
    /**
     * 模拟从数据库中获取数据
     *
     * @param offset 起始位置
     * @param limit  获取的记录数
     * @return 数据列表
     */
    public List<String> fetchData(int offset, int limit) {
        List<String> data = new ArrayList<>();
        for (int i = 0; i < limit; i++) {
            data.add("Data " + (offset + i + 1));
        }
        return data;
    }
}

代码说明

1. 分片逻辑：
   • 将总数据量 TOTAL_RECORDS 分成多个分片，每个分片的大小为 SHARD_SIZE。
   • 计算每个分片的起始位置 start 和结束位置 end。
2. 异步处理：
   • 使用 CompletableFuture.runAsync 异步处理每个分片的数据。
   • 每个分片的任务是调用 DataService.fetchData 获取数据，并将数据写入文件。
3. 线程池：
   • 使用 Executors.newFixedThreadPool 创建固定大小的线程池，控制并发线程数量。
4. 结果合并：
   • 每个分片的任务将数据写入同一个文件。由于文件写入是顺序的，需要通过同步机制（如锁）来避免并发写入冲突。
   • 在本例中，FileWriter 的写入操作是线程安全的，因此不需要额外同步。
5. 等待所有任务完成：
   • 使用 CompletableFuture.allOf 等待所有分片任务完成。

优化点

1. 文件写入性能：
   • 如果文件写入成为性能瓶颈，可以将每个分片的数据写入临时文件，最后再合并所有临时文件。
2. 动态分片大小：
   • 根据数据量和系统资源动态调整分片大小，以平衡负载和性能。
3. 异常处理：
   • 增加异常处理逻辑，确保某个分片任务失败时不会影响其他任务。
4. 资源释放：
   • 确保线程池和文件写入流在任务完成后正确释放。

示例输出

运行程序后，生成的 Sharded_output.csv 文件内容如下：

复制

ID,Data
1,Data 1
2,Data 2
3,Data 3
...
10000,Data 10000

总结

通过分片导出和异步处理，可以高效地处理大数据量的导出任务。CompletableFuture 和线程池的结合使得任务调度和并发控制更加灵活和高效。在实际应用中，可以根据具体需求进一步优化分片大小、线程池配置和文件写入逻辑。

本文作者 xulin（海鸥技术部落），原载于团队 GitBook 技术 Wiki。