本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。
# 交易寄件匣模式
## 意圖
交易式寄件匣模式可解決當單一作業涉及資料庫寫入作業和訊息或事件通知時，分散式系統中所發生的雙重寫入作業問題。當應用程式寫入兩個不同的系統時，就會發生雙重寫入作業；例如，當微服務需要保留資料庫中的資料並傳送訊息以通知其他系統時。其中一項操作失敗可能會導致資料不一致。
## 動機
當微服務在資料庫更新之後傳送事件通知時，這兩項作業應以原子方式執行，以確保資料一致性和可靠性。
+ 如果資料庫更新成功，但事件通知失敗，下游服務將無法察覺變更，而且系統可能會進入不一致的狀態。
+ 如果資料庫更新失敗，但事件通知已傳送，資料可能會損毀，這可能會影響系統的可靠性。
## 適用性
出現下列情況時，請使用交易式寄件匣模式：
+ 您正在構建一個事件驅動的應用程式，其中資料庫更新會啟動事件通知。
+ 您想要確保涉及兩項服務的作業中的原子性。
+ 您想要導入[事件來源模式](event-sourcing.md)。
## 問題和考量
+ **重複訊息**：事件處理服務可能會傳送重複的訊息或事件，因此建議您追蹤已處理的訊息，讓消費服務成為等冪性。
+ **通知順序**：以服務更新資料庫的相同順序傳送訊息或事件。這對於事件來源模式而言相當重要，此模式是您可以在其中使用事件存放區來進行資料存放區的時間點復原。如果順序不正確，可能會影響資料的品質。如果未保留通知順序，則最終一致性和資料庫回復可以解決此問題。
+ **交易回復**：如果交易遭回復，請勿傳送事件通知。
+ **服務層級交易處理**：如果交易橫跨需要資料存放區更新的服務，請使用 [系列事件協同運作模式](saga-orchestration.md)來維持整個資料存放區的資料完整性。
## 實作
### 高層級架構
下列序列圖會顯示雙重寫入作業期間發生的事件順序。
![\[雙重寫入作業期間的事件順序\]](http://docs.aws.amazon.com/zh_tw/prescriptive-guidance/latest/cloud-design-patterns/images/outbox-1.png)
1. 飛行服務會寫入資料庫，並向付款服務傳送事件通知。
1. 訊息代理程式會將訊息和事件傳送至付款服務。訊息代理程式中的任何失敗都會導致付款服務無法接收更新。
如果飛行資料庫更新失敗但已傳送通知，則付款服務會根據事件通知處理付款。這將導致下游資料不一致。
### 使用 AWS 服務實作
為了示範序列圖中的模式，我們將使用下列 AWS 服務，如下圖所示。
+ 微服務是透過使用 [AWS Lambda](https://aws.amazon.com/lambda/) 來實作。
+ 主要資料庫是由 [Amazon Relational Database Service (Amazon RDS)](https://aws.amazon.com/rds/) 所管理。
+ [Amazon Simple Queue Service (Amazon SQS)](https://aws.amazon.com/sqs/) 充當接收事件通知的訊息代理程式。
![\[使用 AWS Lambda、Amazon RDS 和 Amazon SQS 的交易寄件匣模式\]](http://docs.aws.amazon.com/zh_tw/prescriptive-guidance/latest/cloud-design-patterns/images/outbox-2.png)
如果飛行服務在遞交該筆交易後失敗，此情況可能會導致事件通知無法傳送。
![\[遞交操作後的交易失敗\]](http://docs.aws.amazon.com/zh_tw/prescriptive-guidance/latest/cloud-design-patterns/images/outbox-3.png)
不過，交易可能會失敗並回復，但可能仍會傳送事件通知，導致付款服務處理付款。
![\[具有回復的遞交操作後的交易失敗\]](http://docs.aws.amazon.com/zh_tw/prescriptive-guidance/latest/cloud-design-patterns/images/outbox-4.png)
若要解決這個問題，您可以使用寄件匣資料表或變更資料擷取 (CDC)。下列區段將探討這兩個選項，以及您可以如何使用 AWS 服務來實作這些選項。
#### 將寄件匣資料表與關聯式資料庫搭配使用
寄件匣資料表會儲存來自飛行服務的所有事件與時間戳記和序號。
當飛行資料表更新時，同一筆交易中的寄件匣資料表也會隨之更新。另一個服務 (例如，事件處理服務) 會從寄件匣資料表讀取，然後將事件傳送至 Amazon SQS。Amazon SQS 會將有關事件的訊息傳送至付款服務，以便進一步處理。[Amazon SQS 標準佇列](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/standard-queues.html)可保證訊息至少交付一次，而且不會遺失。但是，當您使用 Amazon SQS 標準佇列時，相同的訊息或事件可能會傳遞多次，因此您應確保事件通知服務是等冪的 (也就是說，多次處理相同訊息不會產生不良影響)。如果您需要只處理訊息一次，並使用訊息排序，則可以[先輸入、先出 (FIFO) 佇列來使用 Amazon SQS](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-fifo-queues.html)。
如果飛行資料表更新失敗或寄件匣資料表更新失敗，則整筆交易都會回復，因此不會有不一致的下游資料。
![\[沒有下游資料不一致的回復\]](http://docs.aws.amazon.com/zh_tw/prescriptive-guidance/latest/cloud-design-patterns/images/outbox-5.png)
在下列圖表中，交易寄件匣架構是使用 Amazon RDS 資料庫來實作。當事件處理服務讀取寄件匣資料表時，它只會辨識屬於已遞交 (成功) 交易一部分的資料列，然後將事件的訊息置於 SQS 佇列中，付款服務會讀取此佇列以供進一步處理。此設計可解決雙重寫入作業問題，並使用時間戳記和序號來保留訊息和事件的順序。
![\[解決雙寫操作問題的設計\]](http://docs.aws.amazon.com/zh_tw/prescriptive-guidance/latest/cloud-design-patterns/images/outbox-6.png)
#### 使用變更資料擷取 (CDC)
某些資料庫支援發布項目層級修改，以擷取變更的資料。您可以識別已變更的項目，並據以傳送事件通知。這樣可以節省建立另一個資料表來追蹤更新的開銷。由飛行服務發起的事件會儲存在相同項目的另一個屬性中。
[Amazon DynamoDB](https://aws.amazon.com/dynamodb/) 是一個支援 CDC 更新的鍵值 NoSQL 資料庫。在下列順序圖中，DynamoDB 會將項目層級修改發布至 Amazon DynamoDB Streams。事件處理服務會從串流讀取，並將事件通知發布至付款服務，以便進一步處理。
![\[具有 DynamoDB 和 DynamoDB 串流的交易寄件匣\]](http://docs.aws.amazon.com/zh_tw/prescriptive-guidance/latest/cloud-design-patterns/images/outbox-7.png)
DynamoDB Streams 流會使用時間排序序列擷取 DynamoDB 表中與項目層級變更相關的資訊流程。
您可以在 DynamoDB 資料表上啟用串流，藉此實作交易寄件匣模式。事件處理服務的 Lambda 函數與這些串流相關聯。
+ 更新航班表時，DynamoDB Streams 會擷取變更的資料，而事件處理服務會輪詢串流以取得新記錄。
+ 當新的串流記錄可供使用時，Lambda 函數會同步將事件的訊息放在 SQS 佇列中，以供進一步處理。您可以視需要將屬性新增至 DynamoDB 項目，以擷取時間戳記和序列號，以改善實作的穩健性。
![\[具有 CDC 的交易寄件匣\]](http://docs.aws.amazon.com/zh_tw/prescriptive-guidance/latest/cloud-design-patterns/images/outbox-8.png)
## 範本程式碼
### 使用寄件匣資料表
本節中的範例程式碼說明如何使用寄件匣資料表來實作交易寄件匣模式。若要檢視完整的程式碼，請參閱 [GitHub 儲存庫](https://github.com/aws-samples/transactional-outbox-pattern)以取得此範例。
下面的代碼片段將 `Flight` 實體和 `Flight` 事件保存在單一交易中的各自資料表內的資料庫中。
```
@PostMapping("/flights")
@Transactional
public Flight createFlight(@Valid @RequestBody Flight flight) {
Flight savedFlight = flightRepository.save(flight);
JsonNode flightPayload = objectMapper.convertValue(flight, JsonNode.class);
FlightOutbox outboxEvent = new FlightOutbox(flight.getId().toString(), FlightOutbox.EventType.FLIGHT_BOOKED,
flightPayload);
outboxRepository.save(outboxEvent);
return savedFlight;
}
```
另一項服務負責定期掃描寄件匣資料表中是否有新事件、將事件傳送至 Amazon SQS，並在 Amazon SQS 成功回應時將其從資料表中刪除。輪詢速率可在 `application.properties` 檔案中設定。
```
@Scheduled(fixedDelayString = "${sqs.polling_ms}")
public void forwardEventsToSQS() {
List entities = outboxRepository.findAllByOrderByIdAsc(Pageable.ofSize(batchSize)).toList();
if (!entities.isEmpty()) {
GetQueueUrlRequest getQueueRequest = GetQueueUrlRequest.builder()
.queueName(sqsQueueName)
.build();
String queueUrl = this.sqsClient.getQueueUrl(getQueueRequest).queueUrl();
List messageEntries = new ArrayList<>();
entities.forEach(entity -> messageEntries.add(SendMessageBatchRequestEntry.builder()
.id(entity.getId().toString())
.messageGroupId(entity.getAggregateId())
.messageDeduplicationId(entity.getId().toString())
.messageBody(entity.getPayload().toString())
.build())
);
SendMessageBatchRequest sendMessageBatchRequest = SendMessageBatchRequest.builder()
.queueUrl(queueUrl)
.entries(messageEntries)
.build();
sqsClient.sendMessageBatch(sendMessageBatchRequest);
outboxRepository.deleteAllInBatch(entities);
}
}
```
### 使用變更資料擷取 (CDC)
本節中的範例程式碼說明如何使用 DynamoDB 的變更資料擷取 (CDC) 功能來實作交易寄件匣模式。若要檢視完整的程式碼，請參閱 [GitHub 儲存庫](https://github.com/aws-samples/transactional-outbox-pattern)以取得此範例。
下列 AWS Cloud Development Kit (AWS CDK) 程式碼片段會建立 DynamoDB 航班資料表和 Amazon Kinesis 資料串流 (`cdcStream`)，並設定航班資料表將其所有更新傳送至串流。
```
Const cdcStream = new kinesis.Stream(this, 'flightsCDCStream', {
streamName: 'flightsCDCStream'
})
const flightTable = new dynamodb.Table(this, 'flight', {
tableName: 'flight',
kinesisStream: cdcStream,
partitionKey: {
name: 'id',
type: dynamodb.AttributeType.STRING,
}
});
```
下列程式碼片段和組態定義了彈簧雲端串流函數，可擷取 Kinesis 串流中的更新，並將這些事件轉送至 SQS 佇列以進行進一步處理。
```
applications.properties
spring.cloud.stream.bindings.sendToSQS-in-0.destination=${kinesisstreamname}
spring.cloud.stream.bindings.sendToSQS-in-0.content-type=application/ddb
QueueService.java
@Bean
public Consumer sendToSQS() {
return this::forwardEventsToSQS;
}
public void forwardEventsToSQS(Flight flight) {
GetQueueUrlRequest getQueueRequest = GetQueueUrlRequest.builder()
.queueName(sqsQueueName)
.build();
String queueUrl = this.sqsClient.getQueueUrl(getQueueRequest).queueUrl();
try {
SendMessageRequest send_msg_request = SendMessageRequest.builder()
.queueUrl(queueUrl)
.messageBody(objectMapper.writeValueAsString(flight))
.messageGroupId("1")
.messageDeduplicationId(flight.getId().toString())
.build();
sqsClient.sendMessage(send_msg_request);
} catch (IOException | AmazonServiceException e) {
logger.error("Error sending message to SQS", e);
}
}
```
## GitHub 儲存庫
如需此模式範例架構的完整實作，請參閱 GitHub 存放庫，網址為 [https://github.com/aws-samples/transactional-outbox-pattern](https://github.com/aws-samples/transactional-outbox-pattern)。