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，模擬訂單表查詢服務(wù)，用來確認(rèn)訂單事務(wù)是否提交成功。
     */
    private static boolean checkOrderById(String orderId) {
        return true;
    }

    /**
     * 演示demo，模擬本地事務(wù)的執(zhí)行結(jié)果。
     */
    private static boolean doLocalTransaction() {
        return true;
    }

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