ApsaraMQ for RocketMQ：Pesan terjadwal dan tertunda

Skenario

Pemicu peristiwa berbasis waktu yang akurat dan andal diperlukan dalam skenario seperti penjadwalan terdistribusi dan pemrosesan waktu habis tugas. ApsaraMQ for RocketMQ menyediakan pesan terjadwal untuk membantu menyederhanakan pengembangan tugas penjadwalan waktu dan mengimplementasikan pemicu waktu dengan performa tinggi, skalabilitas, serta keandalan.

Skenario 1: Penjadwalan waktu terdistribusi

Penjadwalan waktu terdistribusi melibatkan tugas-tugas dengan berbagai tingkat granularitas waktu. Contohnya: tugas untuk mengeksekusi pembersihan file setiap hari pukul 5 dan tugas untuk memicu pengiriman pesan setiap 2 menit. Solusi penjadwalan waktu tradisional berbasis dataset kompleks dan tidak efisien dalam skenario terdistribusi. Sebaliknya, pesan terjadwal dalam ApsaraMQ for RocketMQ memungkinkan Anda mengenkapsulasi berbagai jenis pemicu waktu.

Skenario 2: Pemrosesan waktu habis tugas

Contoh khas skenario yang melibatkan pemrosesan waktu habis tugas adalah pembayaran e-commerce, di mana pesanan yang belum dibayar dibatalkan setelah jangka waktu tertentu tanpa pembayaran, bukan langsung dibatalkan. Dalam kasus ini, Anda dapat menggunakan pesan terjadwal dalam ApsaraMQ for RocketMQ untuk memeriksa dan memicu tugas waktu habis.

Mekanisme kerja

Apa itu pesan terjadwal?

Pesan terjadwal adalah jenis pesan unggulan yang disediakan oleh ApsaraMQ for RocketMQ. Setelah pesan terjadwal dikirim ke broker, pesan tersebut hanya dapat dikonsumsi setelah periode waktu tertentu atau pada waktu tertentu. Anda dapat menggunakan pesan terjadwal untuk mengimplementasikan penjadwalan dan pemicuan tertunda dalam skenario terdistribusi.

Aturan pengaturan waktu

• Waktu terjadwal atau tertunda untuk pesan terjadwal dalam ApsaraMQ for RocketMQ direpresentasikan sebagai timestamp, bukan periode waktu.

• Waktu terjadwal adalah timestamp Unix dalam detik. Anda harus mengonversi waktu pengiriman pesan menjadi timestamp Unix dalam detik.

• Waktu terjadwal harus berada dalam rentang waktu yang diizinkan. Jika waktu terjadwal melebihi rentang, waktu terjadwal tidak berlaku dan pesan segera dikirim oleh broker.

• Item berikut menggambarkan waktu terjadwal maksimum untuk instance berbagai jenis:

  • Instance Edisi Standar berlangganan dan bayar sesuai pemakaian serta instance Edisi Standar serverless dan Edisi Profesional: 7 hari.

  • Instance Edisi Profesional berlangganan dan bayar sesuai pemakaian serta Edisi Platinum Perusahaan: 40 hari.

• Waktu terjadwal harus lebih lambat dari waktu saat ini. Jika waktu terjadwal diatur ke waktu yang lebih awal dari waktu saat ini, waktu terjadwal tidak berlaku dan pesan segera dikirim oleh broker.

Siklus hidup pesan terjadwal

• Inisialisasi

  Pesan dibuat dan diinisialisasi oleh produser dan siap dikirim ke broker.

• Dijadwalkan

  Pesan dikirim ke broker dan disimpan dalam sistem penyimpanan berbasis waktu sampai waktu pengiriman yang ditentukan tercapai. Indeks tidak langsung dibuat untuk pesan tersebut.

• Menunggu Konsumsi

  Pada waktu yang ditentukan, pesan ditulis ke mesin penyimpanan reguler, di mana pesan tersebut terlihat oleh konsumen dan menunggu konsumsi oleh konsumen.

• Dikonsumsi

  Pesan diperoleh oleh konsumen dan diproses berdasarkan logika bisnis lokal konsumen.

  Dalam proses ini, broker menunggu konsumen mengembalikan hasil konsumsi. Jika tidak ada respons yang diterima dari konsumen dalam periode waktu tertentu, ApsaraMQ for RocketMQ melakukan percobaan ulang pada pesan. Untuk informasi lebih lanjut, lihat Percobaan Ulang Konsumsi.

• Komitmen Konsumsi

  Konsumen menyelesaikan konsumsi dan mengirimkan hasil konsumsi ke broker. Broker menandai apakah pesan saat ini telah dikonsumsi.

  Secara default, ApsaraMQ for RocketMQ menyimpan semua pesan. Saat hasil konsumsi dikirimkan, pesan ditandai sebagai telah dikonsumsi alih-alih langsung dihapus. Pesan hanya dihapus jika masa retensi berakhir atau sistem kehabisan ruang penyimpanan. Sebelum pesan dihapus, konsumen dapat mengonsumsi ulang pesan tersebut.

• Penghapusan Pesan

  Jika masa retensi pesan berakhir atau ruang penyimpanan tidak mencukupi, ApsaraMQ for RocketMQ menghapus pesan yang tersimpan paling lama dari file fisik secara bergulir. Untuk informasi lebih lanjut, lihat Penyimpanan dan Pembersihan Pesan.

Batasan

Konsistensi tipe pesan

Pesan terjadwal hanya dapat dikirim ke topik yang MessageType-nya diatur ke Delay.

Granularitas waktu

Waktu terjadwal untuk pesan terjadwal dalam ApsaraMQ for RocketMQ akurat hingga milidetik. Nilai granularitas waktu default adalah 1.000 ms.

Status pesan terjadwal dalam ApsaraMQ for RocketMQ dapat disimpan secara persisten. Jika sistem pesan mengalami kegagalan dan di-restart, pesan masih dikirimkan berdasarkan waktu pengiriman yang ditentukan. Namun, jika sistem penyimpanan mengalami pengecualian atau di-restart, latensi mungkin terjadi dalam pengiriman pesan terjadwal.

Skenario

Berbeda dengan pesan normal, pesan terjadwal harus memiliki timestamp pengiriman yang ditentukan untuk mereka.

Kode berikut memberikan contoh tentang cara mengirim dan menerima pesan terjadwal dalam Java.

Untuk informasi tentang kode sampel lengkap untuk perpesanan, lihat Apache RocketMQ 5.x SDK (direkomendasikan).

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;


public class ProducerExample {
    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 = "rmq-cn-xxx.{regionId}.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 = "Topik Anda";
        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 Otentikasi Cerdas halaman Kontrol Akses 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 mendapatkan nama pengguna dan kata sandi berdasarkan informasi VPC. 
         * Jika instance adalah instance serverless ApsaraMQ for RocketMQ, Anda harus menentukan nama pengguna dan kata sandi untuk mengakses instance melalui Internet. Jika Anda mengaktifkan fitur bebas otentikasi di VPC untuk instance dan mengakses instance di VPC, Anda tidak perlu menentukan nama pengguna atau kata sandi. 
         */
        //builder.setCredentialProvider(new StaticSessionCredentialsProvider("Nama Pengguna Instance", "Kata Sandi Instance"));
        ClientConfiguration configuration = builder.build();
        Producer producer = provider.newProducerBuilder()
                .setTopics(topic)
                .setClientConfiguration(configuration)
                .build();
        // Kirim pesan terjadwal dan tertunda.
        // Tentukan timestamp Unix dalam milidetik. Dalam contoh ini, timestamp yang ditentukan menunjukkan bahwa pesan dikirim dalam 10 menit dari waktu saat ini. 
        long deliverTimeStamp = System.currentTimeMillis() + 10L * 60 * 1000;
        Message message = provider.newMessageBuilder()
                .setTopic("topic")
                // Kunci pesan. Anda dapat menggunakan kata kunci untuk menemukan pesan secara akurat. 
                .setKeys("messageKey")
                // Tentukan tag pesan. Konsumen dapat menggunakan tag untuk memfilter pesan. 
                .setTag("messageTag")
                .setDeliveryTimestamp(deliverTimeStamp)
                // Tubuh pesan.
                .setBody("messageBody".getBytes())
                .build();
        try {
            // Kirim pesan. Perhatikan hasilnya dan tangkap pengecualian seperti kegagalan. 
            SendReceipt sendReceipt = producer.send(message);
            System.out.println(sendReceipt.getMessageId());
        } catch (ClientException e) {
            e.printStackTrace();
        }
    }
}

import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Collections;

public class PushConsumerExample {
    private static final Logger LOGGER = LoggerFactory.getLogger(PushConsumerExample.class);

    private PushConsumerExample() {
    }

    public static void main(String[] args) throws ClientException, IOException, InterruptedException {
        /**
         * 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 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 = "rmq-cn-xxx.{regionId}.rmq.aliyuncs.com:8080";
        // Topik yang ingin Anda langgani. Sebelum Anda menentukan topik, Anda harus membuat topik di konsol ApsaraMQ for RocketMQ terlebih dahulu. Jika tidak, kesalahan akan dikembalikan. 
        String topic = "Topik Anda";
        // Grup konsumen tempat konsumen berada. Sebelum Anda menentukan grup konsumen, Anda harus membuat grup konsumen di konsol ApsaraMQ for RocketMQ terlebih dahulu. Jika tidak, kesalahan akan dikembalikan. 
        String consumerGroup = "Grup Konsumen Anda";
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                .setEndpoints(endpoints)
                /**
                 * Jika instance ApsaraMQ for RocketMQ adalah instance serverless dan Anda menggunakan titik akhir publik untuk mengakses instance, Anda harus menentukan ID instance. 
                 */
                //.setNamespace("InstanceId")
                /**
                 * Jika Anda menggunakan titik akhir publik untuk mengakses instance ApsaraMQ for RocketMQ, Anda harus menentukan nama pengguna dan kata sandi instance. Anda dapat memperoleh nama pengguna dan kata sandi di tab Otentikasi Cerdas halaman Kontrol Akses 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 mendapatkan nama pengguna dan kata sandi berdasarkan informasi VPC. 
                 * Jika instance adalah instance serverless ApsaraMQ for RocketMQ, Anda harus menentukan nama pengguna dan kata sandi untuk mengakses instance melalui Internet. Jika Anda mengaktifkan fitur bebas otentikasi di VPC untuk instance serverless dan mengakses instance di VPC, Anda tidak perlu menentukan nama pengguna atau kata sandi. 
                 */
                //.setCredentialProvider(new StaticSessionCredentialsProvider("Nama Pengguna Instance", "Kata Sandi Instance"))
                .build();
        // Aturan yang digunakan untuk memfilter pesan. Dalam contoh berikut, semua pesan dalam topik dilanggan. 
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        // Inisialisasi konsumen dorong. Saat Anda menginisialisasi konsumen dorong, Anda harus menentukan grup konsumen, parameter komunikasi, dan langganan untuk konsumen. 
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                // Grup konsumen. 
                .setConsumerGroup(consumerGroup)
                // Langganan. 
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                // Pendengar pesan. 
                .setMessageListener(messageView -> {
                    // Konsumsi pesan dan kembalikan hasil konsumsi. 
                    // LOGGER.info("Konsumsi pesan={}", messageView);
                    System.out.println("Konsumsi Pesan: " + messageView);
                    return ConsumeResult.SUCCESS;
                })
                .build();
        Thread.sleep(Long.MAX_VALUE);
        // Jika Anda tidak lagi memerlukan konsumen dorong, matikan prosesnya. 
        //pushConsumer.close();
    }
}   

import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
import org.apache.rocketmq.client.apis.message.MessageId;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;

import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.List;

public class SimpleConsumerExample {
    private static final Logger LOGGER = LoggerFactory.getLogger(SimpleConsumerExample.class);

    private SimpleConsumerExample() {
    }

    public static void main(String[] args) throws ClientException, IOException {
        /**
         * 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 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 = "rmq-cn-xxx.{regionId}.rmq.aliyuncs.com:8080";
        // Topik yang ingin Anda langgani. Sebelum Anda menentukan topik, Anda harus membuat topik di konsol ApsaraMQ for RocketMQ terlebih dahulu. Jika tidak, kesalahan akan dikembalikan. 
        String topic = "Topik Anda";
        // Grup konsumen tempat konsumen berada. Sebelum Anda menentukan grup konsumen, Anda harus membuat grup konsumen di konsol ApsaraMQ for RocketMQ terlebih dahulu. Jika tidak, kesalahan akan dikembalikan. 
        String consumerGroup = "Grup Konsumen Anda";
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                .setEndpoints(endpoints)
                /**
                 * Jika instance ApsaraMQ for RocketMQ adalah instance serverless dan Anda menggunakan titik akhir publik untuk mengakses instance, Anda harus menentukan ID instance. 
                 */
                //.setNamespace("InstanceId")
                /**
                 * Jika Anda menggunakan titik akhir publik untuk mengakses instance ApsaraMQ for RocketMQ, Anda harus menentukan nama pengguna dan kata sandi instance. Anda dapat memperoleh nama pengguna dan kata sandi di tab Otentikasi Cerdas halaman Kontrol Akses 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 mendapatkan nama pengguna dan kata sandi berdasarkan informasi VPC. 
                 * Jika instance adalah instance serverless ApsaraMQ for RocketMQ, Anda harus menentukan nama pengguna dan kata sandi untuk mengakses instance melalui Internet. Jika Anda mengaktifkan fitur bebas otentikasi di VPC untuk instance serverless dan mengakses instance di VPC, Anda tidak perlu menentukan nama pengguna atau kata sandi. 
                 */
                //.setCredentialProvider(new StaticSessionCredentialsProvider("Nama Pengguna Instance", "Kata Sandi Instance"))
                .build();
        Duration awaitDuration = Duration.ofSeconds(10);
        // Aturan yang digunakan untuk memfilter pesan. Dalam contoh berikut, semua pesan dalam topik dilanggan. 
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        // Inisialisasi konsumen sederhana. Saat Anda menginisialisasi konsumen sederhana, Anda harus menentukan grup konsumen, parameter komunikasi, dan langganan untuk konsumen. 
        SimpleConsumer consumer = provider.newSimpleConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                // Grup konsumen. 
                .setConsumerGroup(consumerGroup)
                // Durasi timeout untuk permintaan polling panjang. 
                .setAwaitDuration(awaitDuration)
                // Langganan. 
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .build();
        // Jumlah maksimum pesan yang akan ditarik. 
        int maxMessageNum = 16;
        // Waktu tak terlihat pesan. 
        Duration invisibleDuration = Duration.ofSeconds(10);
        // Jika Anda menggunakan konsumen sederhana untuk mengonsumsi pesan, klien harus mendapatkan dan mengonsumsi pesan dalam loop. 
        // Untuk mengonsumsi pesan secara real-time, kami sarankan Anda menggunakan beberapa thread untuk menarik pesan secara bersamaan. 
        while (true) {
            final List<MessageView> messages = consumer.receive(maxMessageNum, invisibleDuration);
            messages.forEach(messageView -> {
                // LOGGER.info("Pesan diterima: {}", messageView);
                System.out.println("Pesan diterima: " + messageView);
            });
            for (MessageView message : messages) {
                final MessageId messageId = message.getMessageId();
                try {
                    // Setelah konsumsi selesai, konsumen harus memanggil metode ACK untuk mengirimkan hasil konsumsi ke broker. 
                    consumer.ack(message);
                    System.out.println("Pesan diakui berhasil, messageId= " + messageId);
                    //LOGGER.info("Pesan diakui berhasil, messageId={}", messageId);
                } catch (Throwable t) {
                    t.printStackTrace();
                    //LOGGER.error("Pesan gagal diakui, messageId={}", messageId, t);
                }
            }
        }
        // Jika Anda tidak lagi memerlukan konsumen sederhana, matikan prosesnya. 
        // consumer.close();
    }
}   

Catatan penggunaan

Jangan jadwalkan waktu pengiriman yang sama untuk sejumlah besar pesan

Pesan terjadwal disimpan dalam sistem penyimpanan berbasis waktu sebelum dikirim ke konsumen pada waktu pengiriman yang ditentukan. Jika Anda menentukan waktu pengiriman yang sama untuk sejumlah besar pesan terjadwal, sistem harus memproses pesan tersebut secara bersamaan pada waktu pengiriman. Ini memberikan beban berat pada sistem dan mengakibatkan penundaan dalam pengiriman pesan.

Tanya Jawab Umum

Apakah saya bisa membatalkan atau mengubah waktu terjadwal untuk pesan terjadwal sebelum waktu terjadwal untuk mengirim pesan tercapai?

Tidak, Anda tidak dapat mengubah waktu terjadwal untuk pesan terjadwal sebelum waktu terjadwal untuk mengirim pesan tercapai.

Apa yang terjadi jika saya menentukan waktu terjadwal yang lebih awal dari waktu saat ini untuk pesan terjadwal?

Jika Anda menentukan waktu terjadwal yang lebih awal dari waktu saat ini untuk pesan terjadwal, waktu terjadwal tidak berlaku, dan pesan segera dikirim.

Mengapa saya tidak dapat menemukan pesan terjadwal yang telah dikirim di konsol ApsaraMQ for RocketMQ?

Pesan terjadwal terlihat oleh konsumen dan dapat ditemukan di konsol ApsaraMQ for RocketMQ hanya setelah waktu terjadwal tercapai.