how to use oss go sdk v2 for multipart upload - Object Storage Service: Central de Documentação do Alibaba Cloud

O recurso de multipart upload do Object Storage Service (OSS) permite dividir um objeto grande em várias partes e enviá-las separadamente. Após enviar todas as partes, chame a operação CompleteMultipartUpload para combiná-las em um objeto completo.

Observações

Os códigos de exemplo neste tópico usam o ID da região cn-hangzhou para a região China (Hangzhou) e um endpoint público por padrão. Ao acessar o OSS a partir de outros serviços da Alibaba Cloud na mesma região, use um endpoint interno. Para mais informações sobre regiões e endpoints do OSS, consulte Regiões e endpoints.
Os exemplos deste tópico leem credenciais de acesso de variáveis de ambiente. Para saber como configurar credenciais de acesso, consulte Configurar credenciais de acesso.
Para realizar um multipart upload, você precisa da permissão oss:PutObject. Para mais detalhes, consulte Conceder permissões personalizadas a um usuário RAM.

Processo de multipart upload

O multipart upload consiste nas três etapas a seguir:

Inicialize um evento de multipart upload.

Chame o método Client.InitiateMultipartUpload para obter um ID de upload globalmente exclusivo do OSS.
Envie as partes.

Chame o método Client.UploadPart para enviar as partes.
Nota
- Para um determinado ID de upload, o número da parte identifica sua posição no objeto. Se você enviar uma nova parte com um número já existente, o OSS substituirá a parte anterior.
- O OSS inclui o hash MD5 da parte recebida no cabeçalho ETag da resposta.
- O OSS calcula o hash MD5 dos dados enviados e o compara com o hash calculado pelo kit de desenvolvimento de software (SDK). Se os hashes não coincidirem, o OSS retornará o código de erro InvalidDigest.
Conclua o multipart upload.

Após enviar todas as partes, chame o método Client.CompleteMultipartUpload para combiná-las em um objeto completo.

Código de exemplo

O código a seguir demonstra como dividir um arquivo local grande em várias partes, enviá-las simultaneamente para um bucket e combiná-las em um objeto completo.

package main

import (
	"bufio"
	"bytes"
	"context"
	"flag"
	"io"
	"log"
	"os"
	"sync"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
)

// Define global variables.
var (
	region     string // The region where the bucket is located.
	bucketName string // The name of the source bucket.
	objectName string // The name of the source object.

)

// The init function initializes command-line parameters.
func init() {
	flag.StringVar(&region, "region", "", "The region in which the bucket is located.")
	flag.StringVar(&bucketName, "bucket", "", "The name of the source bucket.")
	flag.StringVar(&objectName, "object", "", "The name of the source object.")
}

func main() {
	// Parse command-line parameters.
	flag.Parse()

	// Define the upload ID.
	var uploadId string

	// Check if the source bucket name is empty.
	if len(bucketName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, source bucket name required")
	}

	// Check if the region is empty.
	if len(region) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, region required")
	}

	// Check if the source object name is empty.
	if len(objectName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, source object name required")
	}

	// Load default configurations and set the credential provider and region.
	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	// Create an OSS client.
	client := oss.NewClient(cfg)

	// Initialize the multipart upload request.
	initRequest := &oss.InitiateMultipartUploadRequest{
		Bucket: oss.Ptr(bucketName),
		Key:    oss.Ptr(objectName),
	}
	initResult, err := client.InitiateMultipartUpload(context.TODO(), initRequest)
	if err != nil {
		log.Fatalf("failed to initiate multipart upload %v", err)
	}

	// Print the result of the multipart upload initialization.
	log.Printf("initiate multipart upload result:%#v\n", *initResult.UploadId)
	uploadId = *initResult.UploadId

	// Initialize the wait group and mutex.
	var wg sync.WaitGroup
	var parts []oss.UploadPart
	count := 3
	var mu sync.Mutex

	// Read the local file into memory. Replace yourLocalFile with the actual local file name and path.
	file, err := os.Open("yourLocalFile")
	if err != nil {
		log.Fatalf("failed to open local file %v", err)
	}
	defer file.Close()

	bufReader := bufio.NewReader(file)
	content, err := io.ReadAll(bufReader)
	if err != nil {
		log.Fatalf("failed to read local file %v", err)
	}
	log.Printf("file size: %d\n", len(content))

	// Calculate the size of each part.
	chunkSize := len(content) / count
	if chunkSize == 0 {
		chunkSize = 1
	}

	// Start multiple goroutines to upload parts concurrently.
	for i := 0; i < count; i++ {
		start := i * chunkSize
		end := start + chunkSize
		if i == count-1 {
			end = len(content)
		}

		wg.Add(1)
		go func(partNumber int, start, end int) {
			defer wg.Done()

			// Create an upload part request.
			partRequest := &oss.UploadPartRequest{
				Bucket:     oss.Ptr(bucketName),                 // The name of the destination bucket.
				Key:        oss.Ptr(objectName),                 // The name of the destination object.
				PartNumber: int32(partNumber),                   // The part number.
				UploadId:   oss.Ptr(uploadId),                   // The upload ID.
				Body:       bytes.NewReader(content[start:end]), // The content of the part.
			}

			// Send the upload part request.
			partResult, err := client.UploadPart(context.TODO(), partRequest)
			if err != nil {
				log.Fatalf("failed to upload part %d: %v", partNumber, err)
			}

			// Record the result of the part upload.
			part := oss.UploadPart{
				PartNumber: partRequest.PartNumber,
				ETag:       partResult.ETag,
			}

			// Use a mutex to protect shared data.
			mu.Lock()
			parts = append(parts, part)
			mu.Unlock()
		}(i+1, start, end)
	}

	// Wait for all goroutines to complete.
	wg.Wait()

	// Complete the multipart upload request.
	request := &oss.CompleteMultipartUploadRequest{
		Bucket:   oss.Ptr(bucketName),
		Key:      oss.Ptr(objectName),
		UploadId: oss.Ptr(uploadId),
		CompleteMultipartUpload: &oss.CompleteMultipartUpload{
			Parts: parts,
		},
	}
	result, err := client.CompleteMultipartUpload(context.TODO(), request)
	if err != nil {
		log.Fatalf("failed to complete multipart upload %v", err)
	}

	// Print the result of the completed multipart upload.
	log.Printf("complete multipart upload result:%#v\n", result)
}

Cenários comuns

Enviar uma string aleatória de comprimento especificado usando multipart upload

O código a seguir mostra como dividir uma string aleatória de 400 KB em três partes, enviá-las simultaneamente para um bucket e combiná-las em um objeto completo.

package main

import (
	"bufio"
	"context"
	"flag"
	"io"
	"log"
	"math/rand"
	"strings"
	"sync"
	"time"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
)

// Define global variables.
var (
	region     string                                                                     // The region where the bucket is located.
	bucketName string                                                                     // The name of the bucket.
	objectName string                                                                     // The name of the object.
	letters    = []rune("0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ") // The character set for generating random strings.
)

// The init function initializes command-line parameters.
func init() {
	flag.StringVar(&region, "region", "", "The region in which the bucket is located.")
	flag.StringVar(&bucketName, "bucket", "", "The name of the bucket.")
	flag.StringVar(&objectName, "object", "", "The name of the object.")
}

func main() {
	// Parse command-line parameters.
	flag.Parse()

	// Define the upload ID.
	var uploadId string

	// Check if the bucket name is empty.
	if len(bucketName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, bucket name required")
	}

	// Check if the region is empty.
	if len(region) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, region required")
	}

	// Check if the object name is empty.
	if len(objectName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, object name required")
	}

	// Load default configurations and set the credential provider and region.
	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	// Create an OSS client.
	client := oss.NewClient(cfg)

	// Create a request to initialize the multipart upload.
	initRequest := &oss.InitiateMultipartUploadRequest{
		Bucket: oss.Ptr(bucketName), // The name of the bucket.
		Key:    oss.Ptr(objectName), // The name of the object.
	}

	// Initialize the multipart upload and handle the result.
	initResult, err := client.InitiateMultipartUpload(context.TODO(), initRequest)
	if err != nil {
		log.Fatalf("failed to initiate multipart upload %v", err)
	}

	// Print the result of the multipart upload initialization.
	log.Printf("initiate multipart upload result:%#v\n", initResult)
	uploadId = *initResult.UploadId

	// Initialize the wait group and mutex.
	var wg sync.WaitGroup
	var parts []oss.UploadPart
	count := 3
	body := randBody(400000) // Generate a 400 KB random string.
	reader := strings.NewReader(body)
	bufReader := bufio.NewReader(reader)
	content, _ := io.ReadAll(bufReader)
	partSize := len(body) / count
	var mu sync.Mutex

	// Start multiple goroutines to upload parts concurrently.
	for i := 0; i < count; i++ {
		wg.Add(1)
		go func(partNumber int, partSize int, i int) {
			defer wg.Done()

			// Create an upload part request.
			partRequest := &oss.UploadPartRequest{
				Bucket:     oss.Ptr(bucketName),                                             // The name of the bucket.
				Key:        oss.Ptr(objectName),                                             // The name of the object.
				PartNumber: int32(partNumber),                                               // The part number.
				UploadId:   oss.Ptr(uploadId),                                               // The upload ID.
				Body:       strings.NewReader(string(content[i*partSize : (i+1)*partSize])), // The content of the part.
			}

			// Send the upload part request.
			partResult, err := client.UploadPart(context.TODO(), partRequest)
			if err != nil {
				log.Fatalf("failed to upload part %d: %v", partNumber, err)
			}

			// Record the result of the part upload.
			part := oss.UploadPart{
				PartNumber: partRequest.PartNumber,
				ETag:       partResult.ETag,
			}

			// Use a mutex to protect shared data.
			mu.Lock()
			parts = append(parts, part)
			mu.Unlock()
		}(i+1, partSize, i)
	}

	// Wait for all goroutines to complete.
	wg.Wait()

	// Print a message indicating that the parts were uploaded successfully.
	log.Println("upload part success!")

	// Create a request to complete the multipart upload.
	request := &oss.CompleteMultipartUploadRequest{
		Bucket:   oss.Ptr(bucketName),
		Key:      oss.Ptr(objectName),
		UploadId: oss.Ptr(uploadId),
		CompleteMultipartUpload: &oss.CompleteMultipartUpload{
			Parts: parts,
		},
	}

	// Complete the multipart upload and handle the result.
	result, err := client.CompleteMultipartUpload(context.TODO(), request)
	if err != nil {
		log.Fatalf("failed to complete multipart upload %v", err)
	}
	log.Printf("complete multipart upload result:%#v\n", result)
}

// randBody generates a random string of a specified length.
func randBody(n int) string {
	b := make([]rune, n)
	randMarker := rand.New(rand.NewSource(time.Now().UnixNano()))
	for i := range b {
		b[i] = letters[randMarker.Intn(len(letters))]
	}
	return string(b)
}

Cancele um evento específico de multipart upload

Nos cenários a seguir, use o método Client.AbortMultipartUpload para cancele um evento de multipart upload.

Erros no arquivo:
- Se detectar um erro no arquivo durante o envio, como corrupção ou código malicioso, cancele o upload para evitar ameaças potenciais.
Instabilidade de rede:
- Quando a conexão de rede estiver instável ou for interrompida, partes podem ser perdidas ou danificadas. Cancele o upload e reinicie-o para garantir a integridade e a consistência dos dados.
Limites de recursos:
- Caso o espaço de armazenamento seja limitado e o arquivo muito grande, cancele o upload para liberar recursos de armazenamento para tarefas mais importantes.
Operação acidental:
- Se iniciar acidentalmente um upload desnecessário ou enviar a versão errada do arquivo, cancele o evento de upload.

package main

import (
	"context"
	"flag"
	"log"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
)

// Define global variables.
var (
	region     string // The region where the bucket is located.
	bucketName string // The name of the source bucket.
	objectName string // The name of the source object.

)

// The init function initializes command-line parameters.
func init() {
	flag.StringVar(&region, "region", "", "The region in which the bucket is located.")
	flag.StringVar(&bucketName, "bucket", "", "The name of the source bucket.")
	flag.StringVar(&objectName, "object", "", "The name of the source object.")
}

func main() {
	// Parse command-line parameters.
	flag.Parse()

	// Define the upload ID.
	var uploadId string

	// Check if the source bucket name is empty.
	if len(bucketName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, source bucket name required")
	}

	// Check if the region is empty.
	if len(region) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, region required")
	}

	// Check if the source object name is empty.
	if len(objectName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, source object name required")
	}

	// Load default configurations and set the credential provider and region.
	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	// Create an OSS client.
	client := oss.NewClient(cfg)

	// Initialize the multipart upload request.
	initRequest := &oss.InitiateMultipartUploadRequest{
		Bucket: oss.Ptr(bucketName),
		Key:    oss.Ptr(objectName),
	}

	// Execute the multipart upload initialization request.
	initResult, err := client.InitiateMultipartUpload(context.TODO(), initRequest)
	if err != nil {
		log.Fatalf("failed to initiate multipart upload %v", err)
	}

	// Print the result of the multipart upload initialization.
	log.Printf("initiate multipart upload result:%#v\n", *initResult.UploadId)
	uploadId = *initResult.UploadId

	// Create an AbortMultipartUploadRequest request.
	request := &oss.AbortMultipartUploadRequest{
		Bucket:   oss.Ptr(bucketName), // The name of the bucket.
		Key:      oss.Ptr(objectName), // The name of the object.
		UploadId: oss.Ptr(uploadId),   // The upload ID.
	}
	// Execute the request and handle the result.
	result, err := client.AbortMultipartUpload(context.TODO(), request)
	if err != nil {
		log.Fatalf("failed to abort multipart upload %v", err)
	}
	log.Printf("abort multipart upload result:%#v\n", result)

}

Listar partes enviadas com sucesso em um evento específico de multipart upload

Nos cenários abaixo, use o paginador Client.NewListPartsPaginator para listar as partes enviadas com sucesso em um evento de multipart upload.

Monitorar o progresso do upload:

Uploads de arquivos grandes:
- Ao enviar um arquivo muito grande, liste as partes já enviadas para garantir que o processo ocorra conforme o esperado e identificar problemas rapidamente.
Uploads retomáveis:
- Se a rede estiver instável ou o upload for interrompido, verifique as partes já enviadas para decidir se deve tentar enviar novamente as partes restantes. Isso possibilita uploads retomáveis.
Solução de problemas:
- Caso ocorra um erro durante o upload, verifique as partes enviadas para localizar rapidamente o problema, como uma parte específica cuja falha ocorreu no envio. Em seguida, resolva a questão.
Gerenciamento de recursos:
- Em cenários que exigem controle rigoroso de recursos, monitore o progresso do upload para gerencie melhor o espaço de armazenamento e a largura de banda, garantindo o uso eficiente dos recursos.

package main

import (
	"context"
	"flag"
	"log"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
)

// Define global variables.
var (
	region     string // The region where the bucket is located.
	bucketName string // The name of the source bucket.
	objectName string // The name of the source object.

)

// The init function initializes command-line parameters.
func init() {
	flag.StringVar(&region, "region", "", "The region in which the bucket is located.")
	flag.StringVar(&bucketName, "bucket", "", "The name of the source bucket.")
	flag.StringVar(&objectName, "object", "", "The name of the source object.")
}

func main() {
	// Parse command-line parameters.
	flag.Parse()

	// Define the upload ID.
	var uploadId string

	// Check if the source bucket name is empty.
	if len(bucketName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, source bucket name required")
	}

	// Check if the region is empty.
	if len(region) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, region required")
	}

	// Check if the source object name is empty.
	if len(objectName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, source object name required")
	}

	// Load default configurations and set the credential provider and region.
	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	// Create an OSS client.
	client := oss.NewClient(cfg)

	// Initialize the multipart upload request.
	initRequest := &oss.InitiateMultipartUploadRequest{
		Bucket: oss.Ptr(bucketName),
		Key:    oss.Ptr(objectName),
	}

	// Execute the multipart upload initialization request.
	initResult, err := client.InitiateMultipartUpload(context.TODO(), initRequest)
	if err != nil {
		log.Fatalf("failed to initiate multipart upload %v", err)
	}

	// Print the result of the multipart upload initialization.
	log.Printf("initiate multipart upload result:%#v\n", *initResult.UploadId)
	uploadId = *initResult.UploadId

	// Create a request to list parts.
	request := &oss.ListPartsRequest{
		Bucket:   oss.Ptr(bucketName), // The name of the bucket.
		Key:      oss.Ptr(objectName), // The name of the object.
		UploadId: oss.Ptr(uploadId),   // The upload ID.
	}

	// Create a paginator.
	p := client.NewListPartsPaginator(request)

	// Initialize the page counter.
	var i int
	log.Println("List Parts:")

	// Traverse each page in the paginator.
	for p.HasNext() {
		i++

		// Get the data of the next page.
		page, err := p.NextPage(context.TODO())
		if err != nil {
			log.Fatalf("failed to get page %v, %v", i, err)
		}

		// Print the information of each part on the page.
		for _, part := range page.Parts {
			log.Printf("Part Number: %v, ETag: %v, Last Modified: %v, Size: %v, HashCRC64: %v\n",
				part.PartNumber,
				oss.ToString(part.ETag),
				oss.ToTime(part.LastModified),
				part.Size,
				oss.ToString(part.HashCRC64))
		}
	}

}

Listar eventos de multipart upload

Nos cenários a seguir, use o paginador Client.NewListMultipartUploadsPaginator para listar todos os eventos de multipart upload em andamento em um bucket.

Cenários de monitoramento:

Gerenciamento de upload em lote de arquivos:
- Quando precisar enviar muitos arquivos, use o método ListMultipartUploads para monitorar todas as atividades de multipart upload em tempo real. Isso garante que todos os arquivos sejam enviados corretamente.
Detecção e recuperação de falhas:
- Se ocorrerem problemas de rede ou outras falhas durante um upload, algumas partes podem não ser enviadas. Ao monitorar eventos de multipart upload em andamento, é possível detectar esses problemas prontamente e tomar medidas para retomar o upload.
Otimização e gerenciamento de recursos:
- Durante uploads de arquivos em grande escala, monitorar eventos de multipart upload em andamento ajuda a otimizar a alocação de recursos. Por exemplo, ajuste o uso de largura de banda ou otimize a política de upload com base no progresso do envio.
Migração de dados:
- Ao realizar um projeto de migração de dados em grande escala, monitore todos os eventos de multipart upload em andamento para garantir que a tarefa de migração prossiga sem problemas e para identificar e resolver prontamente quaisquer problemas potenciais.

Configurações de parâmetros

Parâmetro	Descrição
Delimiter	Caractere usado para agrupar nomes de objetos. Todos os objetos com o mesmo prefixo até a primeira ocorrência do delimitador são agrupados como um único elemento.
MaxUploads	Número máximo de eventos de multipart upload a serem retornados. O valor padrão e máximo é 1000.
KeyMarker	Especifique o ponto inicial da lista. A lista inclui eventos de multipart upload para objetos cujos nomes estão lexicograficamente após o valor `KeyMarker`. Use este parâmetro com `UploadIDMarker`.
Prefix	Filtra a resposta para incluir apenas objetos cujos nomes começam com o prefixo especificado. Os nomes dos objetos retornados ainda contêm o prefixo.
UploadIDMarker	Use com `KeyMarker` para especifique o ponto inicial da lista. Se `KeyMarker` não estiver definido, o OSS ignora este parâmetro. Se `KeyMarker` estiver definido, a consulta retorna: Eventos de multipart upload para objetos cujos nomes estão lexicograficamente após o valor `KeyMarker`. Eventos de multipart upload para objetos cujos nomes correspondem ao valor `KeyMarker`, mas cujos IDs de upload estão lexicograficamente após o valor `UploadIDMarker`.

Especifique o prefixo file e retorne no máximo 100 resultados

package main

import (
	"context"
	"flag"
	"log"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
)

// Define global variables.
var (
	region     string // The region where the bucket is located.
	bucketName string // The name of the source bucket.
	objectName string // The name of the source object.

)

// The init function initializes command-line parameters.
func init() {
	flag.StringVar(&region, "region", "", "The region in which the bucket is located.")
	flag.StringVar(&bucketName, "bucket", "", "The name of the source bucket.")
	flag.StringVar(&objectName, "object", "", "The name of the source object.")
}

func main() {
	// Parse command-line parameters.
	flag.Parse()

	// Check if the source bucket name is empty.
	if len(bucketName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, source bucket name required")
	}

	// Check if the region is empty.
	if len(region) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, region required")
	}

	// Check if the source object name is empty.
	if len(objectName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, source object name required")
	}

	// Load default configurations and set the credential provider and region.
	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	// Create an OSS client.
	client := oss.NewClient(cfg)

	// Create a request to list multipart uploads.
	request := &oss.ListMultipartUploadsRequest{
		Bucket:     oss.Ptr(bucketName), // The name of the bucket.
		MaxUploads: 100,                 // Specify that a maximum of 100 results are returned.
		Prefix:     oss.Ptr("file"),     // Specify the prefix as file.
	}

	// Create a paginator.
	p := client.NewListMultipartUploadsPaginator(request)

	var i int
	log.Println("List Multipart Uploads:")

	// Traverse each page in the paginator.
	for p.HasNext() {
		i++

		// Get the data of the next page.
		page, err := p.NextPage(context.TODO())
		if err != nil {
			log.Fatalf("failed to get page %v, %v", i, err)
		}

		// Print the information of each multipart upload on the page.
		for _, u := range page.Uploads {
			log.Printf("Upload key: %v, upload id: %v, initiated: %v\n", oss.ToString(u.Key), oss.ToString(u.UploadId), oss.ToTime(u.Initiated))
		}
	}

}

Multipart upload com callbacks

O código a seguir demonstra como dividir um arquivo de 400 KB em três partes, enviá-las simultaneamente para o Alibaba Cloud OSS, combinar as partes em um objeto completo após a conclusão do upload e acionar uma notificação de callback após a combinação das partes.

package main

import (
	"bufio"
	"context"
	"encoding/base64"
	"encoding/json"
	"flag"
	"io"
	"log"
	"math/rand"
	"strings"
	"sync"
	"time"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
)

var (
	region     string
	bucketName string
	objectName string
	letters    = []rune("0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")
)

func init() {
	flag.StringVar(&region, "region", "", "The region in which the bucket is located.")
	flag.StringVar(&bucketName, "bucket", "", "The name of the bucket.")
	flag.StringVar(&objectName, "object", "", "The name of the object.")
}

func main() {
	flag.Parse()
	if len(bucketName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, bucket name required")
	}

	if len(region) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, region required")
	}

	if len(objectName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, object name required")
	}

	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	client := oss.NewClient(cfg)

	initRequest := &oss.InitiateMultipartUploadRequest{
		Bucket: oss.Ptr(bucketName),
		Key:    oss.Ptr(objectName),
	}
	initResult, err := client.InitiateMultipartUpload(context.TODO(), initRequest)

	// Define callback parameters.
	callbackMap := map[string]string{
		"callbackUrl":      "https://example.com:23450",                                                                  // Set the URL of the callback server, for example, https://example.com:23450.
		"callbackBody":     "bucket=${bucket}&object=${object}&size=${size}&my_var_1=${x:my_var1}&my_var_2=${x:my_var2}", // Set the callback request body.
		"callbackBodyType": "application/x-www-form-urlencoded",                                                          // Set the callback request body type.
	}

	// Convert the callback parameters to JSON and encode them in Base64 to pass them as callback parameters.
	callbackStr, err := json.Marshal(callbackMap)
	if err != nil {
		log.Fatalf("failed to marshal callback map: %v", err)
	}
	callbackBase64 := base64.StdEncoding.EncodeToString(callbackStr)

	callbackVarMap := map[string]string{}
	callbackVarMap["x:my_var1"] = "this is var 1"
	callbackVarMap["x:my_var2"] = "this is var 2"
	callbackVarStr, err := json.Marshal(callbackVarMap)
	if err != nil {
		log.Fatalf("failed to marshal callback var: %v", err)
	}
	callbackVarBase64 := base64.StdEncoding.EncodeToString(callbackVarStr)

	var wg sync.WaitGroup
	var parts []oss.UploadPart
	count := 3
	body := randBody(400000)
	reader := strings.NewReader(body)
	bufReader := bufio.NewReader(reader)
	content, _ := io.ReadAll(bufReader)
	partSize := len(body) / count
	var mu sync.Mutex
	for i := 0; i < count; i++ {
		wg.Add(1)
		go func(partNumber int, partSize int, i int) {
			defer wg.Done()
			partRequest := &oss.UploadPartRequest{
				Bucket:     oss.Ptr(bucketName),
				Key:        oss.Ptr(objectName),
				PartNumber: int32(partNumber),
				UploadId:   initResult.UploadId,
				Body:       strings.NewReader(string(content[i*partSize : (i+1)*partSize])),
			}
			partResult, err := client.UploadPart(context.TODO(), partRequest)
			if err != nil {
				log.Fatalf("failed to upload part %d: %v", partNumber, err)
			}
			part := oss.UploadPart{
				PartNumber: partRequest.PartNumber,
				ETag:       partResult.ETag,
			}
			mu.Lock()
			parts = append(parts, part)
			mu.Unlock()
		}(i+1, partSize, i)
	}
	wg.Wait()

	request := &oss.CompleteMultipartUploadRequest{
		Bucket:   oss.Ptr(bucketName),
		Key:      oss.Ptr(objectName),
		UploadId: initResult.UploadId,
		CompleteMultipartUpload: &oss.CompleteMultipartUpload{
			Parts: parts,
		},
		Callback:    oss.Ptr(callbackBase64), // Specify the callback parameters.
		CallbackVar: oss.Ptr(callbackVarBase64),
	}
	result, err := client.CompleteMultipartUpload(context.TODO(), request)
	if err != nil {
		log.Fatalf("failed to complete multipart upload %v", err)
	}
	log.Printf("complete multipart upload result:%#v\n", result)
}

func randBody(n int) string {
	b := make([]rune, n)
	randMarker := rand.New(rand.NewSource(time.Now().UnixNano()))
	for i := range b {
		b[i] = letters[randMarker.Intn(len(letters))]
	}
	return string(b)
}

Multipart upload com barra de progresso

package main

import (
	"bufio"
	"bytes"
	"context"
	"flag"
	"fmt"
	"io"
	"log"
	"os"
	"sync"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
)

// Define global variables.
var (
	region     string // The region where the bucket is located.
	bucketName string // The name of the source bucket.
	objectName string // The name of the source object.

)

// The init function initializes command-line parameters.
func init() {
	flag.StringVar(&region, "region", "", "The region in which the bucket is located.")
	flag.StringVar(&bucketName, "bucket", "", "The name of the source bucket.")
	flag.StringVar(&objectName, "object", "", "The name of the source object.")
}

func main() {
	// Parse command-line parameters.
	flag.Parse()

	// Define the upload ID.
	var uploadId string

	// Check if the source bucket name is empty.
	if len(bucketName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, source bucket name required")
	}

	// Check if the region is empty.
	if len(region) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, region required")
	}

	// Check if the source object name is empty.
	if len(objectName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, source object name required")
	}

	// Load default configurations and set the credential provider and region.
	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	// Create an OSS client.
	client := oss.NewClient(cfg)

	// Initialize the multipart upload request.
	initRequest := &oss.InitiateMultipartUploadRequest{
		Bucket:  oss.Ptr(bucketName),
		Key:     oss.Ptr(objectName),
	}

	// Execute the multipart upload initialization request.
	initResult, err := client.InitiateMultipartUpload(context.TODO(), initRequest)
	if err != nil {
		log.Fatalf("failed to initiate multipart upload %v", err)
	}

	// Print the result of the multipart upload initialization.
	log.Printf("initiate multipart upload result:%#v\n", *initResult.UploadId)
	uploadId = *initResult.UploadId

	// Initialize the wait group and mutex.
	var wg sync.WaitGroup
	var parts []oss.UploadPart
	count := 5
	var mu sync.Mutex

	// Read the local file into memory. Replace /Users/yourLocalPath/yourFileName with the actual path and name of your local file.
	file, err := os.Open("/Users/yourLocalPath/yourFileName")
	if err != nil {
		log.Fatalf("failed to open local file %v", err)
	}
	defer file.Close()

	bufReader := bufio.NewReader(file)
	content, err := io.ReadAll(bufReader)
	if err != nil {
		log.Fatalf("failed to read local file %v", err)
	}
	log.Printf("file size: %d\n", len(content))

	// Calculate the size of each part.
	chunkSize := len(content) / count
	if chunkSize == 0 {
		chunkSize = 1
	}

	// Start multiple goroutines to upload parts concurrently.
	for i := 0; i < count; i++ {
		start := i * chunkSize
		end := start + chunkSize
		if i == count-1 {
			end = len(content)
		}

		wg.Add(1)
		go func(partNumber int, start, end int) {
			defer wg.Done()

			// Create an upload part request.
			partRequest := &oss.UploadPartRequest{
				Bucket:     oss.Ptr(bucketName),                 // The name of the destination bucket.
				Key:        oss.Ptr(objectName),                 // The name of the destination object.
				PartNumber: int32(partNumber),                   // The part number.
				UploadId:   oss.Ptr(uploadId),                   // The upload ID.
				Body:       bytes.NewReader(content[start:end]), // The content of the part.
				ProgressFn: func(increment, transferred, total int64) {
					fmt.Printf("increment:%v, transferred:%v, total:%v\n", increment, transferred, total)
				}, // The progress callback function, used to display the upload progress.
			}

			// Send the upload part request.
			partResult, err := client.UploadPart(context.TODO(), partRequest)
			if err != nil {
				log.Fatalf("failed to upload part %d: %v", partNumber, err)
			}

			log.Printf("successfully uploaded part %d (start: %d, end: %d)", partNumber, start, end)

			// Record the result of the part upload.
			part := oss.UploadPart{
				PartNumber: partRequest.PartNumber,
				ETag:       partResult.ETag,
			}

			// Use a mutex to protect shared data.
			mu.Lock()
			parts = append(parts, part)
			mu.Unlock()
		}(i+1, start, end)
	}

	// Wait for all goroutines to complete.
	wg.Wait()

	// Complete the multipart upload request.
	request := &oss.CompleteMultipartUploadRequest{
		Bucket:   oss.Ptr(bucketName),
		Key:      oss.Ptr(objectName),
		UploadId: oss.Ptr(uploadId),
		CompleteMultipartUpload: &oss.CompleteMultipartUpload{
			Parts: parts,
		},
	}
	result, err := client.CompleteMultipartUpload(context.TODO(), request)
	if err != nil {
		log.Fatalf("failed to complete multipart upload %v", err)
	}

	// Print the versionId of the completed multipart upload.
	log.Printf("complete multipart upload result versionId:%#v\n", result)
}

Referências

Para o código de exemplo completo de multipart upload, consulte o exemplo no GitHub.
Um multipart upload completo envolve três operações de API:
- Para informações sobre a operação de API para inicializar um evento de multipart upload, consulte InitiateMultipartUpload.
- Para informações sobre a operação de API para enviar uma parte, consulte UploadPart.
- Para informações sobre a operação de API para concluir um multipart upload, consulte CompleteMultipartUpload.
Para informações sobre a operação de API para cancele um evento de multipart upload, consulte AbortMultipartUpload.
Para informações sobre a operação de API para listar partes enviadas, consulte NewListPartsPaginator.
Para informações sobre a operação de API para listar todos os eventos de multipart upload em andamento (eventos iniciados, mas ainda não concluídos ou cancelados), consulte NewListMultipartUploadsPaginator.