基于Golang的Couchbase批量Upsert：阻塞与错误处理求助

阻塞场景的行为逻辑 #

当批量Upsert遇到阻塞（比如集群负载过高、网络延迟）时，默认会先进入阻塞等待状态，直到你设置的Timeout超时。一旦超时，Do方法会抛出错误。
如果操作过程中遇到不可恢复的错误（比如权限不足、文档格式非法），则不会等待，会立即返回错误。

错误处理与失败文档定位 #

Do方法返回的err若不为空，你可以通过类型断言判断是否为批量操作的部分失败——gocb的批量操作支持部分成功，*gocb.BulkError会包含所有失败操作的详情：

if bulkErr, ok := err.(*gocb.BulkError); ok {
    for _, opErr := range bulkErr.Errors {
        // 从失败操作中提取文档ID等关键信息
        if upsertOp, ok := opErr.Op.(*gocb.UpsertOp); ok {
            docID := upsertOp.ID
            fmt.Printf("文档%s Upsert失败，错误：%v\n", docID, opErr.Err)
        }
    }
} else if err != nil {
    // 整个批量操作完全失败（比如连接中断、全局超时）
    fmt.Printf("批量操作整体失败：%v\n", err)
}

你代码中设置的Timeout超时错误，也会被包含在BulkError中，对应每个超时的操作都会标记具体错误信息。

失败文档的重试策略 #

定位到失败文档后，可以收集对应的失败操作，重新构建批量容器进行重试：

// 收集需要重试的Upsert操作
var retryOps []gocb.BulkOp
if bulkErr, ok := err.(*gocb.BulkError); ok {
    for _, opErr := range bulkErr.Errors {
        if upsertOp, ok := opErr.Op.(*gocb.UpsertOp); ok {
            retryOps = append(retryOps, upsertOp)
        }
    }
}

// 执行重试（建议添加重试次数限制与指数退避）
if len(retryOps) > 0 {
    retryErr := cl.scope.Collection(collectionName).Do(retryOps, &gocb.BulkOpOptions{
        Context: ctx,
        Timeout: waitingTimeout,
    })
    // 可继续处理重试后的错误，比如再次收集失败操作或记录日志
}

注意：重试时建议添加指数退避机制（如第一次等待100ms，第二次200ms，以此类推），避免给集群造成额外压力；同时要限制重试次数，防止无限循环。

你的代码优化示例 #

当前代码缺少错误处理逻辑，补充后的完整示例如下：

bulkContainer.AddUpsertOp(convertToDocumentId(oraData.IdentifierType, oraData.Identifier), &model.CouchbaseCustomer{Customer: oraData})

err := cl.scope.Collection(collectionName).Do(bulkContainer.ops, &gocb.BulkOpOptions{Context: ctx, Timeout: waitingTimeout})
if err != nil {
    if bulkErr, ok := err.(*gocb.BulkError); ok {
        var retryOps []gocb.BulkOp
        for _, opErr := range bulkErr.Errors {
            if upsertOp, ok := opErr.Op.(*gocb.UpsertOp); ok {
                fmt.Printf("文档%s Upsert失败：%v\n", upsertOp.ID, opErr.Err)
                retryOps = append(retryOps, upsertOp)
            }
        }
        // 执行重试
        if len(retryOps) > 0 {
            retryErr := cl.scope.Collection(collectionName).Do(retryOps, &gocb.BulkOpOptions{
                Context: ctx,
                Timeout: waitingTimeout,
            })
            if retryErr != nil {
                fmt.Printf("重试失败：%v\n", retryErr)
            }
        }
    } else {
        fmt.Printf("批量操作整体异常：%v\n", err)
    }
}

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