ApsaraMQ for RocketMQ：事務訊息

import java.time.Duration;
import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.apache.rocketmq.client.apis.producer.Transaction;
import org.apache.rocketmq.client.apis.producer.TransactionResolution;
import org.apache.rocketmq.client.java.message.MessageBuilderImpl;
import org.apache.rocketmq.client.apis.message.MessageBuilder;
import org.apache.rocketmq.shaded.com.google.common.base.Strings;

public class ProducerTransactionMessageExample {
    /**
     * 示範demo，類比訂單表查詢服務，用來確認訂單事務是否提交成功。
     */
    private static boolean checkOrderById(String orderId) {
        return true;
    }

    /**
     * 示範demo，類比本地事務的執行結果。
     */
    private static boolean doLocalTransaction() {
        return true;
    }

    public static void main(String[] args) throws ClientException {
        /**
         * 執行個體存取點，從控制台執行個體詳情頁的存取點頁簽中擷取。
         * 如果是在阿里雲ECS內網訪問，建議填寫VPC存取點。
         * 如果是在本地公網訪問，或者是線下IDC環境訪問，可以使用公網存取點。使用公網存取點訪問，必須開啟執行個體的公網訪問功能。
         */
        String endpoints = "xxx-hangzhou.rmq.aliyuncs.com:8080";
        //訊息發送的目標Topic名稱，需要提前在控制台建立，如果不建立直接使用會返回報錯。
        String topic = "topic1";
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        /**
         * 如果是使用公網存取點訪問，configuration還需要設定執行個體的使用者名稱和密碼。執行個體使用者名稱和密碼在控制台存取控制的智能身份識別頁簽中擷取。
         * 如果是在阿里雲ECS內網訪問，無需填寫該配置，服務端會根據內網VPC資訊智能擷取。
         * 如果執行個體類型為Serverlesss執行個體，則不管是公網訪問還是內網訪問都必須設定執行個體的使用者名稱密碼。
         */
        builder.setCredentialProvider(new StaticSessionCredentialsProvider("Instance UserName", "Instance Password"));
        builder.setRequestTimeout(Duration.ofMillis(5000));
        ClientConfiguration configuration = builder.build();
        
        //構造事務生產者：事務訊息需要生產者構建一個事務檢查器，用於檢查確認異常半事務的中間狀態。
        Producer producer = provider.newProducerBuilder()
            .setTransactionChecker(messageView -> {
                /**
                 * 事務檢查器一般是根據業務的ID去檢查本地事務是否正確提交還是復原，此處以訂單ID屬性為例。
                 * 在訂單表找到了這個訂單，說明本地事務插入訂單的操作已經正確提交；如果訂單表沒有訂單，說明本地事務已經復原。
                 */
                final String orderId = messageView.getProperties().get("OrderId");
                if (Strings.isNullOrEmpty(orderId)) {
                    // 錯誤的訊息，直接返回Rollback。
                    return TransactionResolution.ROLLBACK;
                }
                return checkOrderById(orderId) ? TransactionResolution.COMMIT : TransactionResolution.ROLLBACK;
            }).setTopics(topic)
            .setClientConfiguration(configuration)
            .build();
        //開啟事務分支。
        final Transaction transaction;
        try {
            transaction = producer.beginTransaction();
        } catch (ClientException e) {
            e.printStackTrace();
            //事務分支開啟失敗，直接退出。
            return;
        }
        Message message = provider.newMessageBuilder()
            .setTopic(topic)
            //設定訊息索引鍵，可根據關鍵字精確尋找某條訊息。
            .setKeys("messageKey1")
            //設定訊息Tag，用於消費端根據指定Tag過濾訊息。
            .setTag("messageTag")
            //一般事務訊息都會設定一個本地事務關聯的唯一ID，用來做本地事務回查的校正。
            .addProperty("OrderId", "xxx")
            //訊息體。
            .setBody("messageBody".getBytes())
            .build();
        //發送半事務訊息
        final SendReceipt sendReceipt;
        try {
            sendReceipt = producer.send(message, transaction);
        } catch (ClientException e) {
            //半事務訊息發送失敗，事務可以直接退出並復原。
            return;
        }
        /**
         * 執行本地事務，並確定本地事務結果。
         * 1. 如果本地事務提交成功，則提交訊息事務。
         * 2. 如果本地事務提交失敗，則復原訊息事務。
         * 3. 如果本地事務未知異常，則不處理，等待事務訊息回查。
         *
         */
        boolean localTransactionOk = doLocalTransaction();
        if (localTransactionOk) {
            try {
                transaction.commit();
            } catch (ClientException e) {
                // 業務可以自身對即時性的要求選擇是否重試，如果放棄重試，可以依賴事務訊息回查機制進行事務狀態的提交。
                e.printStackTrace();
            }
        } else {
            try {
                transaction.rollback();
            } catch (ClientException e) {
                // 建議記錄異常資訊，復原異常時可以無需重試，依賴事務訊息回查機制進行事務狀態的提交。
                e.printStackTrace();
            }
        }
    }
}