Azure Cosmos DB 中的交易批處理操作

適用於: ✅ NoSQL

交易批次是指一組點操作，這些操作在具有相同分割鍵的容器中需要共同成功或失敗。定義操作，將其加入批次，然後執行該批次。如果所有操作依照交易批次操作中描述的順序成功，交易即被提交。然而，若有任何操作失敗，整個交易會被回滾。

Azure Cosmos DB 中的交易是什麼？

典型資料庫中的交易可定義為作為單一邏輯工作單元執行的一系列操作。每筆交易都提供 ACID（原子性、一致性、隔離性與耐久性）特性保證。

原子性 保證交易中所有操作都被視為單一單位，且要麼全部提交，要麼都不提交。
一致性 確保資料在交易間始終處於有效狀態。
隔離確保沒有兩筆交易互相干擾。許多商用系統提供多種隔離等級，可依應用需求使用。
持久性 確保資料庫中任何已提交的變更都始終存在。

Azure Cosmos DB 支援完全符合 ACID 規範的交易，並能在同一邏輯分割鍵內實現快照隔離。

交易批次操作與儲存程序

Azure Cosmos DB 目前支援儲存程序，這些程序同時提供操作的交易範圍。然而，交易批次操作提供以下優點：

語言選項 – 你已經使用的 SDK 和語言支援交易批次，而儲存程序則需要用 JavaScript 撰寫。
程式碼版本管理 ——將應用程式程式碼版本化並導入 CI/CD 管線，比起協調更新儲存程序並確保在適當時間切換更為自然。這也讓變更回復變得更容易。
效能 – 與儲存程序執行相比，等效操作的延遲可減少最多 30%。
內容序列化 – 交易批次中的每個操作都可以為其有效載荷使用自訂序列化選項。

如何建立交易批次操作

從一個容器實例開始，呼叫 CreateTransactionalBatch：

PartitionKey partitionKey = new PartitionKey("road-bikes");

TransactionalBatch batch = container.CreateTransactionalBatch(partitionKey);

接著，在批次中加入多個操作：

Product bike = new (
    id: "68719520766",
    category: "road-bikes",
    name: "Chropen Road Bike"
);

batch.CreateItem<Product>(bike);

Part part = new (
    id: "68719519885",
    category: "road-bikes",
    name: "Tronosuros Tire",
    productId: bike.id
);

batch.CreateItem<Part>(part);

最後，在批次中呼叫 ExecuteAsync ：

using TransactionalBatchResponse response = await batch.ExecuteAsync();

收到回應後，檢視回應是否成功。若回應顯示成功，請擷取結果：

if (response.IsSuccessStatusCode)
{
    TransactionalBatchOperationResult<Product> productResponse;
    productResponse = response.GetOperationResultAtIndex<Product>(0);
    Product productResult = productResponse.Resource;

    TransactionalBatchOperationResult<Part> partResponse;
    partResponse = response.GetOperationResultAtIndex<Part>(1);
    Part partResult = partResponse.Resource;
}

這很重要

如果發生失敗，失敗的操作會顯示對應錯誤的狀態碼。其他操作的 HTTP 狀態碼統一為 424（依賴失敗）。若操作失敗，因為嘗試建立已存在的項目，則回傳 409（衝突）狀態碼。狀態碼可以讓你找出交易失敗的原因。

呼叫 CosmosBatch.createCosmosBatch：

PartitionKey partitionKey = new PartitionKey("road-bikes");

CosmosBatch batch = CosmosBatch.createCosmosBatch(partitionKey);

接著，在批次中加入多個操作：

Product bike = new Product();
bike.setId("68719520766");
bike.setCategory("road-bikes");
bike.setName("Chropen Road Bike");

batch.createItemOperation(bike);

Part part = new Part();
part.setId("68719519885");
part.setCategory("road-bikes");
part.setName("Tronosuros Tire");
part.setProductId(bike.getId());

batch.createItemOperation(part);

最後，使用容器實例呼叫 executeCosmosBatch 並包含批次：

CosmosBatchResponse response = container.executeCosmosBatch(batch);

收到回應後，檢視回應是否成功。若回應顯示成功，請擷取結果：

if (response.isSuccessStatusCode())
{
    List<CosmosBatchOperationResult> results = response.getResults();
}

這很重要

如果發生失敗，失敗的操作會顯示對應錯誤的狀態碼。其他操作的狀態碼都是 424，表示相依失敗。若操作失敗，因為嘗試建立已存在的項目，則回傳 409（衝突）狀態碼。狀態碼可以讓你找出交易失敗的原因。

取得或建立容器實例：

container = database.create_container_if_not_exists(id="batch_container",
                                                        partition_key=PartitionKey(path='/category'))

在 Python 中，事務性批次操作看起來與單一操作的 API 相似，且是包含 operation_type_string、args_tuple 和 batch_operation_kwargs_dictionary 的元組。以下範例項目展示批次操作的功能：


create_demo_item = {
    "id": "68719520766",
    "category": "road-bikes",
    "name": "Chropen Road Bike"
}

# for demo, assume that this item already exists in the container. 
# the item id will be used for read operation in the batch
read_demo_item1 = {
    "id": "68719519884",
    "category": "road-bikes",
    "name": "Tronosuros Tire",
    "productId": "68719520766"
}

# for demo, assume that this item already exists in the container. 
# the item id will be used for read operation in the batch
read_demo_item2 = {
    "id": "68719519886",
    "category": "road-bikes",
    "name": "Tronosuros Tire",
    "productId": "68719520766"
}

# for demo, assume that this item already exists in the container. 
# the item id will be used for read operation in the batch
read_demo_item3 = {
    "id": "68719519887",
    "category": "road-bikes",
    "name": "Tronosuros Tire",
    "productId": "68719520766"
}

# for demo, we'll upsert the item with id 68719519885
upsert_demo_item = {
    "id": "68719519885",
    "category": "road-bikes",
    "name": "Tronosuros Tire Upserted",
    "productId": "68719520768"
}

# for replace demo, we'll replace the read_demo_item2 with this item
replace_demo_item = {
    "id": "68719519886",
    "category": "road-bikes",
    "name": "Tronosuros Tire replaced",
    "productId": "68719520769"
}

# for replace with etag match demo, we'll replace the read_demo_item3 with this item
# The use of etags and if-match/if-none-match options allows users to run conditional replace operations
# based on the etag value passed. When using if-match, the request will only succeed if the item's latest etag
# matches the passed in value. For more on optimistic concurrency control, see the link below:
# /azure/cosmos-db/nosql/database-transactions-optimistic-concurrency
replace_demo_item_if_match_operation = {
    "id": "68719519887",
    "category": "road-bikes",
    "name": "Tronosuros Tireh",
    "wasReplaced": "Replaced based on etag match"
    "productId": "68719520769"
}

準備將要加入到批次操作中的步驟：

create_item_operation = ("create", (create_demo_item,), {})
read_item_operation = ("read", ("68719519884",), {})
delete_item_operation = ("delete", ("68719519885",), {})
upsert_item_operation = ("upsert", (upsert_demo_item,), {})
replace_item_operation = ("replace", ("68719519886", replace_demo_item), {})
replace_item_if_match_operation = ("replace",
                                       ("68719519887", replace_demo_item_if_match_operation),
                                       {"if_match_etag": container.client_connection.last_response_headers.get("etag")})

將操作加入批次：

batch_operations = [
        create_item_operation,
        read_item_operation,
        delete_item_operation,
        upsert_item_operation,
        replace_item_operation,
        replace_item_if_match_operation
    ]

最後，執行批次：

try:
    # Run that list of operations
    batch_results = container.execute_item_batch(batch_operations=batch_operations, partition_key="road_bikes")
    # Batch results are returned as a list of item operation results - or raise a CosmosBatchOperationError if
    # one of the operations failed within your batch request.
    print("\nResults for the batch operations: {}\n".format(batch_results))
except exceptions.CosmosBatchOperationError as e:
        error_operation_index = e.error_index
        error_operation_response = e.operation_responses[error_operation_index]
        error_operation = batch_operations[error_operation_index]
        print("\nError operation: {}, error operation response: {}\n".format(error_operation, error_operation_response))
    # [END handle_batch_error]

備註

如果你在批次中使用 patch 操作和 replace_if_match_etag 操作：批次操作 kwargs 字典有限制，且總共只接受三個不同的鍵值。若想在批次中使用條件補丁，則可使用 filter_predicate 鍵進行補丁操作;若想在任何操作中使用 etag，也可用 if_match_etag/if_none_match_etag 鍵。

batch_operations = [
       ("replace", (item_id, item_body), {"if_match_etag": etag}),
       ("patch", (item_id, operations), {"filter_predicate": filter_predicate, "if_none_match_etag": etag}),
   ]

這很重要

打電話給 NewTransactionalBatch。

pk := azcosmos.NewPartitionKeyString("road-bikes")

batch := container.NewTransactionalBatch(pk)

接著，在批次中加入多個操作：

type Bike struct {
	ID       string `json:"id"`
	Category string `json:"category"`
	Name     string `json:"name"`
}

bike := Bike{ID: "68719520766", Category: "road-bikes", Name: "Chropen Road Bike"}
bikeItem, _ := json.Marshal(bike)

batch.CreateItem(bikeItem, nil)

type BikePart struct {
	ID        string `json:"id"`
	Category  string `json:"category"`
	Name      string `json:"name"`
	ProductID string `json:"productID"`
}

part := BikePart{ID: "68719519885", Category: "road-bikes", Name: "Tronosuros Tire", ProductID: bike.ID}
bikePartItem, _ := json.Marshal(part)

batch.CreateItem(bikePartItem, nil)

最後，使用容器實例和批次來呼叫ExecuteTransactionalBatch。

resp, _ := container.ExecuteTransactionalBatch(context.Background(), batch, nil)

if resp.Success {
	// execute conditional logic
}

這很重要

交易批次操作的執行方式

當交易批次執行時，交易批次中的所有操作會被分組、序列化成單一有效載荷，並以單一請求發送給 Azure Cosmos 資料庫服務。

服務接收請求並執行交易範圍內的所有操作，並使用相同的序列化協定回傳回應。此回應可視為成功或失敗，並提供每個操作的個別操作回應。

SDK 會公開回應，讓你驗證結果，並可選擇性地擷取每個內部操作結果。

局限性

目前已知有兩個限制：

Azure Cosmos DB 的請求大小限制限制了交易批次有效載荷的大小不得超過 2 MB，且最大執行時間為 5 秒。
目前每個交易批次有 100 次操作的限制，以確保效能符合預期且符合 SLA。

後續步驟

TransactionalBatch 的使用

意見反應

此頁面對您有幫助嗎？

Last updated on 2025-12-19

共用方式為

Azure Cosmos DB 中的交易批處理操作

Azure Cosmos DB 中的交易是什麼？

交易批次操作與儲存程序

如何建立交易批次操作

交易批次操作的執行方式

局限性

後續步驟

意見反應

其他資源