如何在CouchDB中实现大文件分块上传及下载合并功能

实现思路 #

• 采用两类文档存储结构：
  • 文件元数据文档：存储文件基础信息，包括文件名、总大小、分块大小、总分块数等，不存储实际二进制内容
  • 分块内容文档：每个文档存储单块二进制内容，关联对应元数据文档的id，以及当前分块的序号
• 分块大小推荐设置为1MB，平衡单文档大小和请求次数，避免超过CouchDB单文档大小阈值

前置定义 #

// 文件元数据实体
public class CouchDbFileMeta
{
    public string _id { get; set; }
    public string _rev { get; set; }
    public string FileName { get; set; }
    public long TotalSize { get; set; }
    public int ChunkSize { get; set; }
    public int TotalChunks { get; set; }
}

// 分块内容实体
public class CouchDbFileChunk
{
    public string _id { get; set; }
    public string _rev { get; set; }
    public string ParentFileId { get; set; }
    public int ChunkIndex { get; set; }
    public byte[] Data { get; set; }
}

分块上传实现 #

替换原有InsertDataToCouchDb方法，默认采用异步调用避免线程死锁：

public async Task<string> InsertDataToCouchDb(string dbName, string fileId, string filename, byte[] fileContent)
{
    var connection = System.Configuration.ConfigurationManager.ConnectionStrings["CouchDb"].ConnectionString;
    // 分块大小设置为1MB，可根据业务调整
    const int chunkSize = 1 * 1024 * 1024;
    using var db = new MyCouchClient(connection, dbName);
    
    // 计算总分块数
    var totalChunks = (int)Math.Ceiling((double)fileContent.Length / chunkSize);
    var uploadedChunkIds = new List<string>();

    try
    {
        // 逐个上传分块
        for (var i = 0; i < totalChunks; i++)
        {
            var start = i * chunkSize;
            var end = Math.Min(start + chunkSize, fileContent.Length);
            var chunkData = new byte[end - start];
            Buffer.BlockCopy(fileContent, start, chunkData, 0, chunkData.Length);

            var chunk = new CouchDbFileChunk
            {
                // 分块id按规则生成，方便后续查询
                _id = $"{fileId}_chunk_{i}",
                ParentFileId = fileId,
                ChunkIndex = i,
                Data = chunkData
            };

            var chunkResp = await db.Entities.PutAsync(chunk);
            uploadedChunkIds.Add(chunk.Id);
        }

        // 所有分块上传成功后，存储元数据
        var meta = new CouchDbFileMeta
        {
            _id = fileId,
            FileName = filename,
            TotalSize = fileContent.Length,
            ChunkSize = chunkSize,
            TotalChunks = totalChunks
        };

        var metaResp = await db.Entities.PutAsync(meta);
        return metaResp.Id;
    }
    catch (Exception)
    {
        // 上传失败，清理已上传的分块，避免垃圾数据
        foreach (var chunkId in uploadedChunkIds)
        {
            await db.Documents.DeleteAsync(chunkId);
        }
        throw;
    }
}

分块下载实现 #

替换原有FetchDataFromCouchDb方法：

public async Task<byte[]> FetchDataFromCouchDb(string dbName, string fileId)
{
    var connection = System.Configuration.ConfigurationManager.ConnectionStrings["CouchDb"].ConnectionString;
    using var db = new MyCouchClient(connection, dbName);
    
    // 先拉取元数据
    var metaResp = await db.Entities.GetAsync<CouchDbFileMeta>(fileId);
    var meta = metaResp.Content;
    if (meta == null)
    {
        throw new KeyNotFoundException("指定文件不存在");
    }

    // 按序号拉取所有分块
    var chunks = new byte[meta.TotalChunks][];
    for (var i = 0; i < meta.TotalChunks; i++)
    {
        var chunkId = $"{fileId}_chunk_{i}";
        var chunkResp = await db.Entities.GetAsync<CouchDbFileChunk>(chunkId);
        chunks[i] = chunkResp.Content.Data;
    }

    // 合并分块为完整字节数组
    using var ms = new MemoryStream();
    foreach (var chunk in chunks)
    {
        await ms.WriteAsync(chunk, 0, chunk.Length);
    }
    return ms.ToArray();
}

注意事项 #

• 原有代码中直接使用.Result调用异步方法存在线程死锁风险，推荐统一改为async/await异步调用模式，若必须保留同步方法签名，可自行调整为同步调用
• 若文件大小超过100MB，可适当调大分块大小到2MB-4MB，减少请求次数
• 生产环境可给分块文档的ParentFileId字段建立视图索引，方便批量查询分块以及后续的垃圾数据清理

内容的提问来源于stack exchange，提问作者Hardik Barot