ApsaraMQ for RocketMQ：Pesan transaksional

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 ini digunakan untuk mensimulasikan layanan query tabel pesanan untuk memeriksa apakah transaksi pesanan telah dikirim. 
     */
    private static boolean checkOrderById(String orderId) {
        return true;
    }

    /**
     // Demo ini digunakan untuk mensimulasikan hasil eksekusi transaksi lokal. 
     */
    private static boolean doLocalTransaction() {
        return true;
    }

    public static void main(String[] args) throws ClientException {
        /**
         * Titik akhir dari instance. Anda dapat melihat titik akhir di tab Endpoints halaman Detail Instance di konsol ApsaraMQ for RocketMQ. 
         * Jika klien instance ApsaraMQ for RocketMQ ditempatkan di instance Elastic Compute Service (ECS) dan Anda ingin mengakses instance ApsaraMQ for RocketMQ di jaringan internal, kami sarankan Anda menentukan titik akhir virtual private cloud (VPC). 
         * Jika Anda mengakses instance melalui Internet atau dari pusat data, Anda dapat menentukan titik akhir publik. Jika Anda mengakses instance melalui Internet, Anda harus mengaktifkan fitur akses Internet untuk instance. 
         */
        String endpoints = "xxx-hangzhou.rmq.aliyuncs.com:8080";
        // Nama topik ke mana pesan dikirim. Sebelum Anda menggunakan topik untuk menerima pesan, Anda harus membuat topik di konsol ApsaraMQ for RocketMQ. Jika tidak, kesalahan akan dikembalikan. 
        String topic = "topic1";
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        /**
         * Jika Anda mengakses instance menggunakan titik akhir publik, Anda harus menentukan nama pengguna dan kata sandi instance. Anda dapat memperoleh nama pengguna dan kata sandi di tab Intelligent Authentication halaman Access Control yang sesuai dengan instance di konsol ApsaraMQ for RocketMQ. 
         * Jika klien instance ApsaraMQ for RocketMQ ditempatkan di instance ECS dan Anda ingin mengakses instance ApsaraMQ for RocketMQ di jaringan internal, Anda tidak perlu menentukan nama pengguna atau kata sandi. Broker secara otomatis memperoleh nama pengguna dan kata sandi berdasarkan informasi VPC. 
         * Jika instance adalah instance serverless, Anda harus menentukan nama pengguna dan kata sandi instance, terlepas dari apakah Anda mengakses instance melalui Internet atau di VPC. 
         */
        builder.setCredentialProvider(new StaticSessionCredentialsProvider("Instance UserName", "Instance Password"));
        builder.setRequestTimeout(Duration.ofMillis(5000));
        ClientConfiguration configuration = builder.build();
        
        MessageBuilder messageBuilder = new MessageBuilderImpl();

        // Bangun produsen transaksi. Produsen harus membangun pemeriksa transaksi untuk memeriksa status perantara dari pesan setengah yang luar biasa. 
        Producer producer = provider.newProducerBuilder()
            .setTransactionChecker(messageView -> {
                /**
                 * Pemeriksa transaksi memeriksa apakah transaksi lokal dikomit atau dibatalkan dengan benar berdasarkan ID bisnis. Dalam contoh ini, ID pesanan digunakan. 
                 * Jika pesanan ini ditemukan di tabel pesanan, tindakan penyisipan pesanan dikomit dengan benar oleh transaksi lokal. Jika tidak ada pesanan yang ditemukan di tabel pesanan, transaksi lokal dibatalkan. 
                 */
                final String orderId = messageView.getProperties().get("OrderId");
                if (Strings.isNullOrEmpty(orderId)) {
                    // Terjadi kesalahan pada pesan. Rollback dikembalikan. 
                    return TransactionResolution.ROLLBACK;
                }
                return checkOrderById(orderId) ? TransactionResolution.COMMIT : TransactionResolution.ROLLBACK;
            }).setTopics(topic)
            .setClientConfiguration(configuration)
            .build();
        // Buat cabang transaksi. 
        final Transaction transaction;
        try {
            transaction = producer.beginTransaction();
        } catch (ClientException e) {
            e.printStackTrace();
            // Jika cabang transaksi gagal dibuat, transaksi dihentikan. 
            return;
        }
        Message message = messageBuilder.setTopic(topic)
            // Kunci pesan. Anda dapat menggunakan kata kunci untuk menemukan pesan secara akurat. 
            .setKeys("messageKey1")
            // Tag pesan. Konsumen dapat menggunakan tag untuk memfilter pesan. 
            .setTag("messageTag")
            // ID unik yang terkait dengan transaksi lokal. ID ini digunakan untuk memverifikasi query status transaksi lokal. 
            .addProperty("OrderId", "xxx")
            // Tubuh pesan. 
            .setBody("messageBody".getBytes())
            .build();
        // Kirim pesan setengah.
        final SendReceipt sendReceipt;
        try {
            sendReceipt = producer.send(message, transaction);
        } catch (ClientException e) {
            // Jika pesan setengah gagal dikirim, transaksi dihentikan dan pesan dibatalkan. 
            return;
        }
        /**
         * Eksekusi transaksi lokal dan tentukan hasil eksekusi. 
         * 1. Jika hasilnya adalah Commit, kirimkan pesan. 
         * 2. Jika hasilnya adalah Rollback, batalkan pesan. 
         * 3. Jika pengecualian tidak diketahui terjadi, tidak ada tindakan yang dilakukan sampai respons diperoleh dari query status transaksi lokal. 
         *
         */
        boolean localTransactionOk = doLocalTransaction();
        if (localTransactionOk) {
            try {
                transaction.commit();
            } catch (ClientException e) {
                // Anda dapat menentukan apakah akan mencoba ulang pesan berdasarkan persyaratan bisnis Anda. Jika Anda tidak ingin mencoba ulang pesan, Anda dapat memeriksa status transaksi lokal untuk mengirim status transaksi. 
                e.printStackTrace();
            }
        } else {
            try {
                transaction.rollback();
            } catch (ClientException e) {
                // Kami sarankan Anda mencatat informasi pengecualian. Dengan cara ini, jika pengecualian terjadi selama pembatalan pesan, Anda dapat memeriksa status transaksi lokal untuk mengirim status transaksi tanpa mencoba ulang pesan. 
                e.printStackTrace();
            }
        }
    }
}