MinIO分片上传超大文件（纯服务端）

一、MinIO快速搭建

这里简单先容一下通过docker方式快速搭建MinIO的大要流程。
1.1、拉取docker镜像

起首直接实验拉取：

1. docker pull minio/minio

复制代码

如果拉不到，试图更改docker镜像源：

1. echo '{
2.     "registry-mirrors": [
3.         "https://4xxwxhl6.mirror.aliyuncs.com",
4.         "https://mirror.iscas.ac.cn",
5.         "https://docker.rainbond.cc",
6.         "https://docker.nju.edu.cn",
7.         "https://6kx4zyno.mirror.aliyuncs.com",
8.         "https://mirror.baidubce.com",
9.         "https://docker.m.daocloud.io",
10.         "https://dockerproxy.com"
11.     ]
12. }' | sudo tee /etc/docker/daemon.json > /dev/null

复制代码

接侧重启docker服务，使新设置见效：

1. sudo systemctl restart docker

复制代码

末了再次拉取即可。
1.2、启动docker容器

起首创创建置和数据目次：

1. mkdir -p /opt/minio/config
2. mkdir -p /opt/minio/data

复制代码

接着启动：

1. docker run -p 9000:9000 -p 9001:9001 --net=host --name minio -d --restart=always -e "MINIO_ACCESS_KEY=minio" -e "MINIO_SECRET_KEY=minio123" -v /opt/minio/data:/data -v /opt/minio/config:/root/.minio minio/minio server /data --console-address ":9001" -address ":9000"

复制代码

末了进入MinIO控制台http://192.168.2.195:9001，简单做点存储桶、用户、用户组等设置即可。好比创建新用户名minioUser，暗码minioUser123。
二、分片上传大文件到MinIO

2.1、添加依靠

这里必要留意minio 8.3.3必须依靠okhttp的版本不小于4.8.1。

1. // minio 8.3.3 Must use okhttp >= 4.8.1
2. implementation 'io.minio:minio:8.3.3'
3. implementation 'com.squareup.okhttp3:okhttp:4.12.0'

复制代码

2.2、实现MinioClient

参考S3官方文档https://docs.aws.amazon.com/AmazonS3/latest/userguide/mpuoverview.html#mpu-process，大文件的分片上传，告急分三步实现：

• initMultiPartUpload创建一个大文件分片上传任务
  
• uploadMultiPart逐个上传分片
  
• mergeMultipartUpload归并分片
  

通过继承默认的MinioClient，将一些干系的告急方法暴袒露来，以便利用。

1. package com.szh.minio;
3. import com.google.common.collect.Multimap;
4. import io.minio.*;
5. import io.minio.errors.*;
6. import io.minio.messages.Part;
8. import java.io.IOException;
9. import java.security.InvalidKeyException;
10. import java.security.NoSuchAlgorithmException;
12. public class CustomMinioClient extends MinioClient {
14.     /**
15.      * 继承父类
16.      */
17.     public CustomMinioClient(MinioClient client) {
18.         super(client);
19.     }
21.     /**
22.      * 初始化分片上传即获取uploadId
23.      */
24.     public String initMultiPartUpload(String bucket, String region, String object, Multimap<String, String> headers, Multimap<String, String> extraQueryParams) throws IOException, InvalidKeyException, NoSuchAlgorithmException, InsufficientDataException, ServerException, InternalException, XmlParserException, InvalidResponseException, ErrorResponseException {
25.         CreateMultipartUploadResponse response = this.createMultipartUpload(bucket, region, object, headers, extraQueryParams);
26.         return response.result().uploadId();
27.     }
29.     /**
30.      * 上传单个分片
31.      */
32.     public UploadPartResponse uploadMultiPart(String bucket, String region, String object, Object data,
33.                                               long length,
34.                                               String uploadId,
35.                                               int partNumber,
36.                                               Multimap<String, String> headers,
37.                                               Multimap<String, String> extraQueryParams) throws IOException, InvalidKeyException, NoSuchAlgorithmException, InsufficientDataException, ServerException, InternalException, XmlParserException, InvalidResponseException, ErrorResponseException {
38.         return this.uploadPart(bucket, region, object, data, length, uploadId, partNumber, headers, extraQueryParams);
39.     }
41.     /**
42.      * 合并分片
43.      */
44.     public ObjectWriteResponse mergeMultipartUpload(String bucketName, String region, String objectName, String uploadId, Part[] parts, Multimap<String, String> extraHeaders, Multimap<String, String> extraQueryParams) throws IOException, NoSuchAlgorithmException, InsufficientDataException, ServerException, InternalException, XmlParserException, InvalidResponseException, ErrorResponseException, ServerException, InvalidKeyException {
45.         return this.completeMultipartUpload(bucketName, region, objectName, uploadId, parts, extraHeaders, extraQueryParams);
46.     }
48.     public void cancelMultipartUpload(String bucketName, String region, String objectName, String uploadId, Multimap<String, String> extraHeaders, Multimap<String, String> extraQueryParams) throws ServerException, InsufficientDataException, ErrorResponseException, NoSuchAlgorithmException, IOException, InvalidKeyException, XmlParserException, InvalidResponseException, InternalException {
49.         this.abortMultipartUpload(bucketName, region, objectName, uploadId, extraHeaders, extraQueryParams);
50.     }
52.     /**
53.      * 查询当前上传后的分片信息
54.      */
55.     public ListPartsResponse listMultipart(String bucketName, String region, String objectName, Integer maxParts, Integer partNumberMarker, String uploadId, Multimap<String, String> extraHeaders, Multimap<String, String> extraQueryParams) throws NoSuchAlgorithmException, InsufficientDataException, IOException, InvalidKeyException, ServerException, XmlParserException, ErrorResponseException, InternalException, InvalidResponseException {
56.         return this.listParts(bucketName, region, objectName, maxParts, partNumberMarker, uploadId, extraHeaders, extraQueryParams);
57.     }
58. }

复制代码

2.3、实现分片上传

2.3.0、初始化MinioClient

毗连到minio，并确生存储桶的存在。

1. static CustomMinioClient minioClient = new CustomMinioClient(MinioClient.builder()
2.         .endpoint("http://192.168.2.195:9000")
3.         .credentials("minioUser", "minioUser123")
4.         .build());
5. // 测试桶
6. static String bucketName = "test";
7. static {
8.     try {
9.         boolean found = minioClient.bucketExists(BucketExistsArgs.builder().bucket(bucketName).build());
10.         if (!found) {
11.             minioClient.makeBucket(MakeBucketArgs.builder().bucket(bucketName).build());
12.         }
13.     } catch (Exception e) {
14.         throw new RuntimeException(e);
15.     }
16. }

复制代码

2.3.1、预备分片上传

创建一个大文件分片上传任务。

1. String contentType = "application/octet-stream";
2. HashMultimap<String, String> headers = HashMultimap.create();
3. headers.put("Content-Type", contentType);
4. String uploadId = minioClient.initMultiPartUpload(bucketName, null, file.getName(), headers, null);
5. System.out.println("uploadId: " + uploadId);

复制代码

2.3.2、分片并上传

本文是利用纯服务端举行分片和上传，而实际项目中更保举由后端起首调用minio的接口getPresignedObjectUrl，逐个天生每个分片的署名后的上传url，然后前端直接以此上传到minio，即可省去后端服务的网络IO开销。
📢 后者方案请见：MinIO分片上传超大文件（非纯服务端）
2.3.2.1、设置分片巨细

一方面，必要留意单个分片巨细最小5MB，如果每个分片设置小于5MB，则minio或S3底层在归并时报错：code = EntityTooSmall, message = Your proposed upload is smaller than the minimum allowed object size。
另一方面，在调解分片巨细时，必要留意minio或S3底层答应的分片范围[1,10000]。
2.3.2.2、分片

一方面，为了包管分片的服从，借助线程池的并发，以及RandomAccessFile的文件随机访问本领，更快地完身分片的流程。固然，可控制并发数和分片巨细以防止并发分片中的OOM。
另一方面，思量到分片全部完成之后，另有末了的归并操纵，以是借助CountDownLatch来确保全部门片上传之后，再去实行归并。
2.3.3、分片归并

归并全部已上传的分片。

1. Part[] parts = new Part[(int) chunkCount];
2. // 查询上传后的分片数据。S3最大允许10000，且从1开始
3. ListPartsResponse partResult = minioClient.listMultipart(bucketName, null, file.getName(), 10000, 0, uploadId, null, null);
4. int partNumber = 1;
5. for (Part part : partResult.result().partList()) {
6.     parts[partNumber - 1] = new Part(partNumber, part.etag());
7.     partNumber++;
8. }
9. ObjectWriteResponse objectWriteResponse = minioClient.mergeMultipartUpload(bucketName, null, file.getName(), uploadId, parts, null, null);

复制代码

三、测试

3.1、完备测试代码

1. package com.szh.minio;
3. import com.google.common.collect.HashMultimap;
4. import io.minio.*;
5. import io.minio.messages.Part;
6. import lombok.Getter;
7. import lombok.Setter;
9. import java.io.File;
10. import java.io.IOException;
11. import java.io.RandomAccessFile;
12. import java.util.concurrent.CountDownLatch;
13. import java.util.concurrent.ExecutorService;
14. import java.util.concurrent.Executors;
16. @Setter
17. @Getter
18. public class MinioMain {
19.     static CustomMinioClient minioClient = new CustomMinioClient(MinioClient.builder()
20.             .endpoint("http://192.168.2.195:9000")
21.             .credentials("minioUser", "minioUser123")
22.             .build());
23.     // 测试桶
24.     static String bucketName = "test";
25.     static {
26.         try {
27.             boolean found = minioClient.bucketExists(BucketExistsArgs.builder().bucket(bucketName).build());
28.             if (!found) {
29.                 minioClient.makeBucket(MakeBucketArgs.builder().bucket(bucketName).build());
30.             }
31.         } catch (Exception e) {
32.             throw new RuntimeException(e);
33.         }
34.     }
36.     // 需要被分片上传的大文件
37.     static String filePath = "C:\\tmp\\psi_result.csv";
38.     static File file = new File(filePath);
40.     // 单个分片大小5MB，如果每个分片设置小于5MB，则minio或S3底层在合并时报错：
41.     // code = EntityTooSmall, message = Your proposed upload is smaller than the minimum allowed object size.
42.     static final long CHUNK_SIZE = 5 * 1024 * 1024;
44.     // 当前分片号，minio或S3底层允许的分片范围[1,10000]
45.     // https://docs.aws.amazon.com/AmazonS3/latest/userguide/mpuoverview.html#mpu-process
46.     private int chunkIndex;
48.     // 用于得知所有分片都传输成功后的时刻，进而进行合并
49.     private static CountDownLatch countDownLatch;
51.     public static void main(String[] args) throws Exception {
52.         // 第一步：准备分片上传
53.         String contentType = "application/octet-stream";
54.         HashMultimap<String, String> headers = HashMultimap.create();
55.         headers.put("Content-Type", contentType);
56.         String uploadId = minioClient.initMultiPartUpload(bucketName, null, file.getName(), headers, null);
57.         System.out.println("uploadId: " + uploadId);
59.         // 第二步：分片并上传
60.         // ps：实际项目中可由后端先getPresignedObjectUrl逐个生成每个分片的签名后的上传url，前端直接以此上传到minio，即可省去后端服务的网络开销
61.         long totalLength = file.length();
62.         System.out.println("totalLength: " + totalLength + " Byte");
63.         // 计算分片数量
64.         long chunkCount = (totalLength + CHUNK_SIZE - 1) / CHUNK_SIZE;
65.         System.out.println("chunkCount: " + chunkCount);
66.         countDownLatch = new CountDownLatch((int) chunkCount);
67.         // 5个核心线程并发上传分片
68.         ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);
69.         for (long i = 0; i < chunkCount; i++) {
70.             long position = i * CHUNK_SIZE;
71.             int bytesRead = (int) Math.min(CHUNK_SIZE, totalLength - position);
73.             MinioMain minioMain = new MinioMain();
74.             // S3分片号从1开始
75.             minioMain.setChunkIndex((int) i + 1);
76.             fixedThreadPool.submit(new Runnable() {
77.                 @Override
78.                 public void run() {
79.                     try {
80.                         // 上传分片
81.                         minioMain.processChunk(filePath, position, bytesRead, uploadId);
82.                     } catch (Exception e) {
83.                         throw new RuntimeException(e);
84.                     }
85.                 }
86.             });
87.         }
88.         countDownLatch.await();
89.         fixedThreadPool.shutdownNow();
91.         // 第三步：合并分片
92.         System.out.println("ready to merge <" + file.getName() + " - " + uploadId + " - " + bucketName + ">");
93.         Part[] parts = new Part[(int) chunkCount];
94.         // 查询上传后的分片数据。S3最大允许10000，且从1开始
95.         ListPartsResponse partResult = minioClient.listMultipart(bucketName, null, file.getName(), 10000, 0, uploadId, null, null);
96.         int partNumber = 1;
97.         for (Part part : partResult.result().partList()) {
98.             parts[partNumber - 1] = new Part(partNumber, part.etag());
99.             partNumber++;
100.         }
101.         ObjectWriteResponse objectWriteResponse = minioClient.mergeMultipartUpload(bucketName, null, file.getName(), uploadId, parts, null, null);
102.         System.out.println("mergeMultipartUpload resp etag: " + objectWriteResponse.etag());
103.         StatObjectResponse statObjectResponse = minioClient.statObject(StatObjectArgs.builder().bucket(bucketName).object(file.getName()).build());
104.         System.out.println("etag: " + statObjectResponse.etag() + " size: " + statObjectResponse.size() + " lastModified: " + statObjectResponse.lastModified());
105.     }
107.     private void processChunk(String filePath, long position, int bytesRead, String uploadId) {
108.         // 可控制并发数和分片大小以防止OOM
109.         byte[] buffer = new byte[bytesRead];
110.         RandomAccessFile raf = null;
111.         try {
112.             int chunkIndex = this.getChunkIndex();
113.             raf = new RandomAccessFile(filePath, "r");
114.             // 定位到指定位置
115.             raf.seek(position);
116.             // 读取bytesRead字节长度作为分片
117.             raf.readFully(buffer);
118.             String contentType = "application/octet-stream";
119.             HashMultimap<String, String> headers = HashMultimap.create();
120.             headers.put("Content-Type", contentType);
121.             UploadPartResponse uploadPartResponse = minioClient.uploadMultiPart(bucketName, null, file.getName(),
122.                     buffer, bytesRead,
123.                     uploadId, chunkIndex, headers, null);
124.             System.out.println("chunk[" + chunkIndex + "] buffer size: [" + buffer.length + " Byte] upload etag: [" + uploadPartResponse.etag() + "]");
125.         } catch (Exception e) {
126.             e.printStackTrace();
127.         } finally {
128.             if (raf != null) {
129.                 try {
130.                     raf.close();
131.                 } catch (IOException e) {
132.                     e.printStackTrace();
133.                 }
134.             }
135.             countDownLatch.countDown();
136.         }
137.     }
138. }

复制代码

3.2、运行日记和结果

运行日记如下：

1. uploadId: MzFiMWRmZjctMDg0Yy00YzMyLTk5NTYtMjRkZGZiMDZlYjJhLmUwZmFkNzFiLWEwZTctNDU1Yi04ZWFjLWFhODQyZjBiMmIyOXgxNzI3MzQwMjUzMTA2Njc5MTEz
2. totalLength: 3576974860 Byte
3. chunkCount: 683
4. chunk[1] buffer size: [5242880 Byte] upload etag: [97096e510d1dcda56646608345de08ea]
5. chunk[3] buffer size: [5242880 Byte] upload etag: [d8102f80f10eb79f600cdf2d378ae8fe]
6. chunk[4] buffer size: [5242880 Byte] upload etag: [b74f9b8fa2025580b4fc00449c66e271]
7. chunk[5] buffer size: [5242880 Byte] upload etag: [e77603ee49cc3f7d229f124ecd9a3f38]
8. chunk[2] buffer size: [5242880 Byte] upload etag: [b148b311ccd2b3fcd4777d56a8758c3d]
9. chunk[6] buffer size: [5242880 Byte] upload etag: [94abe5a7a2117b612d9805029398cfd9]
10. chunk[7] buffer size: [5242880 Byte] upload etag: [433b52aed0d1b1486df07a2259932a83]
11. chunk[8] buffer size: [5242880 Byte] upload etag: [2c242bd205f9b3c4546454fe2d0abef4]
12. ...
13. chunk[679] buffer size: [5242880 Byte] upload etag: [8492b0573cc74ec55cb6d2a86aee0f69]
14. chunk[678] buffer size: [5242880 Byte] upload etag: [4aa5c01b4f7aea95952ec62d71ee9996]
15. chunk[681] buffer size: [5242880 Byte] upload etag: [ac0b739044bfd2644fc8da97fc03a1a9]
16. chunk[680] buffer size: [5242880 Byte] upload etag: [d95ee210ac774b3ca26e091941c66e20]
17. chunk[682] buffer size: [5242880 Byte] upload etag: [75e78df64c1fad0839ba8a1583cd93ec]
18. chunk[683] buffer size: [1330700 Byte] upload etag: [2f30c8d65e23d266c7f10f051854bc6a]
19. ready to merge <psi_result.csv - MzFiMWRmZjctMDg0Yy00YzMyLTk5NTYtMjRkZGZiMDZlYjJhLmUwZmFkNzFiLWEwZTctNDU1Yi04ZWFjLWFhODQyZjBiMmIyOXgxNzI3MzQwMjUzMTA2Njc5MTEz - test>
20. mergeMultipartUpload resp etag: "ff6ebd330b3cb224ade84463dd14df82-683"
21. etag: ff6ebd330b3cb224ade84463dd14df82-683 size: 3576974860 lastModified: 2024-09-26T09:09Z

复制代码

上传后的控制台：