Ringkasan messaging transaksional dan two-phase commit - ApsaraMQ for RocketMQ

Pesan transaksional adalah jenis pesan fitur yang disediakan oleh ApsaraMQ for RocketMQ untuk menjamin bahwa transaksi lokal dan pengiriman pesan sama-sama berhasil atau sama-sama gagal. Mekanisme commit dua fase ini menjaga sinkronisasi antara layanan inti dan konsumen downstream-nya tanpa overhead penguncian resource seperti pada transaksi terdistribusi eXtended Architecture (XA).

Distributed transaction requirements

Gunakan pesan transaksional ketika:

Sistem pesanan harus memperbarui database-nya dan memberi notifikasi ke layanan logistik, poin, serta keranjang secara atomik.
Layanan pembayaran harus mencatat debit dan menerbitkan event ke konsumen ledger downstream.
Mengirimkan pesan tanpa menyelesaikan transaksi lokal (atau sebaliknya) akan menyebabkan sistem berada dalam keadaan tidak konsisten.

Cara kerja pesan transaksional

Mengapa pesan normal tidak cukup

Menggabungkan transaksi database lokal dengan pengiriman pesan normal menciptakan celah di mana salah satu operasi dapat berhasil tanpa yang lainnya:

Pesan dikirim, tetapi transaksi lokal gagal. Konsumen downstream bertindak berdasarkan perubahan yang belum dikomit.
Transaksi lokal dikomit, tetapi pengiriman pesan gagal. Konsumen downstream tidak pernah mengetahui perubahan tersebut.
Terjadi timeout, dan baik produsen maupun broker tidak dapat menentukan apakah harus melakukan commit atau rollback.

Normal message solution

Mengapa transaksi XA terlalu mahal

Protokol XA dapat mengoordinasikan transaksi terdistribusi lintas sistem, tetapi mengunci resource selama durasi transaksi berlangsung. Seiring bertambahnya jumlah sistem yang berpartisipasi, kontensi penguncian meningkat dan throughput menurun.

Commit dua fase dengan half message

ApsaraMQ for RocketMQ menggunakan protokol commit dua fase untuk pesan transaksional yang menghindari kedua masalah tersebut:

Transactional message solution

Kirim half message. Produsen mengirim pesan ke broker. Broker menyimpannya secara persisten dan mengembalikan acknowledgment (ACK) ke produsen. Pesan tersebut ditandai sebagai *belum siap dikirim* — pesan dalam keadaan ini disebut half message. Konsumen downstream belum dapat melihatnya.
Jalankan transaksi lokal. Produsen menjalankan operasi database lokalnya (misalnya, memperbarui status pesanan dari *belum dibayar* menjadi *sudah dibayar*).
Commit atau rollback. Produsen melaporkan hasil transaksi lokal ke broker:
- Commit: Broker menandai half message sebagai *siap dikirim* dan mengirimkannya ke konsumen.
- Rollback: Broker membuang half message tersebut. Konsumen tidak pernah menerimanya.
Pemeriksaan status transaksi (pemulihan). Jika broker tidak menerima hasil commit atau rollback — karena kegagalan jaringan atau restart produsen — broker mengirim kueri status ke instans produsen dalam kluster. Produsen memeriksa hasil transaksi lokal dan melaporkannya kembali ke broker.

Transaction status check workflow

Catatan

Untuk interval kueri dan jumlah maksimum retry, lihat Batas parameter.

Siklus hidup pesan

Pesan transaksional melewati status berikut:

Transactional message lifecycle

Status	Deskripsi
Inisialisasi	Produsen menyusun half message dan bersiap mengirimkannya ke broker.
Transaksi menunggu commit	Broker menyimpan half message dalam sistem penyimpanan transaksi. Berbeda dengan pesan normal, half message tidak disimpan oleh broker dengan cara standar. Pesan ini tidak terlihat oleh konsumen.
Committed for consumption	Transaksi lokal berhasil. Broker menyimpan half message dalam sistem penyimpanan, sehingga pesan tersebut terlihat oleh konsumen.
Rollback pesan	Transaksi lokal gagal. Broker membuang half message tersebut. Alur kerja berakhir.
Sedang dikonsumsi	Konsumen mengambil pesan dan mulai memprosesnya. Jika konsumen tidak mengembalikan hasil dalam batas waktu yang dikonfigurasi, ApsaraMQ for RocketMQ mencoba mengirim ulang. Untuk detailnya, lihat Retry konsumsi.
Commit hasil konsumsi	Konsumen melakukan commit terhadap hasil konsumsi. Pesan ditandai sebagai telah dikonsumsi tetapi tidak langsung dihapus.
Penghapusan pesan	Periode retensi pesan berakhir atau storage space hampir penuh. ApsaraMQ for RocketMQ menghapus pesan tertua secara bergilir. Lihat Penyimpanan dan pembersihan pesan.

Secara default, ApsaraMQ for RocketMQ menyimpan semua pesan. Pesan yang telah dikonsumsi tidak langsung dihapus — konsumen masih dapat mengonsumsinya ulang hingga periode retensi berakhir atau ruang penyimpanan direklamasi.

Kirim pesan transaksional (Java)

Prasyarat

Sebelum memulai, pastikan Anda telah memiliki:

Topik dengan MessageType diatur ke Transaction di Konsol ApsaraMQ for RocketMQ.
Titik akhir instans (dari tab Endpoints pada halaman Instance Details)
(Jika berlaku) username dan password instans (dari tab Intelligent Authentication pada halaman Access Control)

Perbedaan dari pesan normal

Mengirim pesan transaksional berbeda dari mengirim pesan normal dalam dua hal:

Pemeriksa transaksi wajib. Daftarkan pemeriksa transaksi saat membuat produsen. Pemeriksa ini dijalankan secara otomatis jika broker melakukan kueri status transaksi setelah terjadi kegagalan.
Binding topik wajib. Bind topik target ke produsen saat pembuatan agar pemeriksa bawaan dapat memulihkan status transaksi.

Kode contoh

Buat produsen dengan pemeriksa transaksi, mulai transaksi, kirim half message, jalankan transaksi lokal, lalu lakukan commit atau rollback.

Contoh kode

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 {

    // Mensimulasikan pemeriksaan apakah pesanan ada di database.
    private static boolean checkOrderById(String orderId) {
        return true;
    }

    // Mensimulasikan transaksi lokal (misalnya, menyisipkan catatan pesanan).
    private static boolean doLocalTransaction() {
        return true;
    }

    public static void main(String[] args) throws ClientException {
        // Ganti dengan titik akhir instans Anda.
        // Temukan ini di tab Endpoints pada halaman Instance Details
        // di Konsol ApsaraMQ for RocketMQ.
        String endpoints = "<your-instance-endpoint>";

        // Topik harus memiliki MessageType diatur ke Transaction.
        String topic = "<your-transaction-topic>";

        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder()
            .setEndpoints(endpoints);

        // Autentikasi:
        // - Titik akhir publik: tentukan username dan password (temukan di tab
        //   Intelligent Authentication pada halaman Access Control).
        // - Titik akhir VPC di ECS: tidak perlu kredensial; broker menyelesaikannya
        //   dari VPC.
        // - Instans serverless: selalu tentukan kredensial, terlepas dari
        //   metode akses.
        builder.setCredentialProvider(
            new StaticSessionCredentialsProvider("<your-username>", "<your-password>"));
        builder.setRequestTimeout(Duration.ofMillis(5000));
        ClientConfiguration configuration = builder.build();

        MessageBuilder messageBuilder = new MessageBuilderImpl();

        // Buat produsen dengan pemeriksa transaksi.
        // Pemeriksa ini dijalankan ketika broker melakukan kueri terhadap half message
        // yang hasil commit/rollback-nya tidak diterima.
        Producer producer = provider.newProducerBuilder()
            .setTransactionChecker(messageView -> {
                // Cari ID pesanan yang dilampirkan pada half message.
                // Jika pesanan ada di database, transaksi lokal
                // berhasil dikomit. Jika tidak, lakukan rollback.
                final String orderId = messageView.getProperties().get("OrderId");
                if (Strings.isNullOrEmpty(orderId)) {
                    return TransactionResolution.ROLLBACK;
                }
                return checkOrderById(orderId)
                    ? TransactionResolution.COMMIT
                    : TransactionResolution.ROLLBACK;
            })
            .setTopics(topic)
            .setClientConfiguration(configuration)
            .build();

        // Langkah 1: Mulai transaksi.
        final Transaction transaction;
        try {
            transaction = producer.beginTransaction();
        } catch (ClientException e) {
            e.printStackTrace();
            return;
        }

        // Langkah 2: Susun dan kirim half message.
        Message message = messageBuilder.setTopic(topic)
            .setKeys("messageKey1")
            .setTag("messageTag")
            // Lampirkan ID bisnis agar pemeriksa transaksi dapat mengkuerinya nanti.
            .addProperty("OrderId", "xxx")
            .setBody("messageBody".getBytes())
            .build();

        final SendReceipt sendReceipt;
        try {
            sendReceipt = producer.send(message, transaction);
        } catch (ClientException e) {
            // Pengiriman half message gagal. Transaksi berakhir di sini.
            return;
        }

        // Langkah 3: Jalankan transaksi lokal.
        boolean localTransactionOk = doLocalTransaction();

        // Langkah 4: Lakukan commit atau rollback berdasarkan hasil transaksi lokal.
        if (localTransactionOk) {
            try {
                transaction.commit();
            } catch (ClientException e) {
                // Jika panggilan commit gagal, broker akan memanggil
                // pemeriksa transaksi untuk menyelesaikan statusnya.
                e.printStackTrace();
            }
        } else {
            try {
                transaction.rollback();
            } catch (ClientException e) {
                // Catat error tersebut. Broker akan memanggil pemeriksa transaksi
                // untuk menyelesaikan statusnya.
                e.printStackTrace();
            }
        }
    }
}

Ganti placeholder berikut dengan nilai aktual Anda:

Placeholder	Deskripsi	Contoh
`<your-instance-endpoint>`	Titik akhir instans (dari tab Endpoints pada halaman Instance Details)	`xxx-hangzhou.rmq.aliyuncs.com:8080`
`<your-transaction-topic>`	Nama topik dengan `MessageType` diatur ke `Transaction`	`order-tx-topic`
`<your-username>`	Username instans (dari tab Intelligent Authentication pada halaman Access Control)	`MjoxODgwNzcwODY5MD****`
`<your-password>`	Kata sandi instance	`NEh6cm9FVUl****`

Untuk contoh SDK lengkap dalam semua bahasa yang didukung, lihat SDK Apache RocketMQ 5.x.

Praktik terbaik

Minimalkan hasil transaksi yang tidak diketahui

Pemeriksaan status transaksi berfungsi sebagai jaring pengaman untuk kegagalan selama commit atau rollback. Volume tinggi pemeriksaan status akan menurunkan performa sistem dan menunda pengiriman pesan. Rancang transaksi lokal agar segera mengembalikan hasil Commit atau Rollback yang pasti.

Tangani transaksi yang sedang berjalan dengan benar

Ketika broker melakukan kueri status half message dan transaksi lokal masih berjalan, kembalikan Unknown — bukan Commit atau Rollback. Mengembalikan hasil yang prematur dapat menyebabkan inkonsistensi data.

Jika kueri status datang terlalu awal karena transaksi lokal lambat, pertimbangkan pendekatan berikut:

Tingkatkan penundaan kueri pertama. Konfigurasikan interval yang lebih lama sebelum broker mengirim kueri status pertamanya. Pertimbangan: ini juga menunda pemulihan untuk transaksi yang benar-benar gagal.
Deteksi status sedang berjalan secara eksplisit. Rancang logika transaksi lokal agar dapat membedakan antara "masih berjalan" dan "gagal" sehingga pemeriksa mengembalikan status yang tepat.

Batasan

Batasan	Detail
Jenis topik	Pesan transaksional memerlukan topik dengan `MessageType` diatur ke `Transaction`.
Satu SendReceipt per transaksi	Setiap transaksi hanya mendukung satu `SendReceipt`.
Hanya konsistensi eventual	Pesan transaksional menjamin konsistensi antara transaksi lokal dan pengiriman pesan. Namun, tidak menjamin konsistensi real-time di seluruh konsumen downstream. Hingga pesan dikirim, status downstream mungkin tertinggal dari transaksi upstream. Gunakan pesan transaksional hanya jika pemrosesan downstream asinkron dapat diterima.
Tanggung jawab sisi konsumen	ApsaraMQ for RocketMQ menjamin bahwa pesan yang dikomit akan dikirim, tetapi setiap konsumen downstream harus menangani pemrosesan dengan benar. Implementasikan logika retry konsumsi untuk menangani kegagalan sementara. Lihat Retry konsumsi.
Timeout transaksi	Jika broker tidak dapat menentukan hasil transaksi setelah timeout dan jumlah maksimum retry yang dikonfigurasi tercapai, broker akan melakukan rollback half message secara default. Lihat Batas parameter.