ApsaraMQ for RocketMQ:Kirim dan berlangganan pesan tertunda

Kirim pesan tertunda

Satu-satunya perbedaan dibandingkan dengan mengirim pesan biasa adalah satu pemanggilan metode: msg.setStartDeliverTime(delayTime). Ini menetapkan timestamp absolut (dalam milidetik) saat broker mengirimkan pesan.

Untuk contoh lainnya, lihat ApsaraMQ for RocketMQ code library.

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;
import java.util.Date;
import java.util.Properties;

public class ProducerDelayTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // Ambil pasangan AccessKey dari variabel lingkungan.
        // Pastikan ALIBABA_CLOUD_ACCESS_KEY_ID dan ALIBABA_CLOUD_ACCESS_KEY_SECRET telah disetel.
        properties.put(PropertyKeyConst.AccessKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
        properties.put(PropertyKeyConst.SecretKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));

        // Tentukan titik akhir TCP.
        // Temukan nilai ini di bagian TCP Endpoint pada halaman Detail Instans
        // di Konsol ApsaraMQ for RocketMQ.
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "<your-tcp-endpoint>");

        Producer producer = ONSFactory.createProducer(properties);
        // Panggil start() sekali sebelum mengirim pesan apa pun.
        producer.start();

        Message msg = new Message(
                "<your-topic>",       // Topik yang dibuat di Konsol
                "DelayMessageTag",    // Tag untuk pemfilteran di sisi konsumen, mirip dengan tag Gmail
                "Hello MQ".getBytes() // Isi pesan dalam format biner.
                                      // ApsaraMQ for RocketMQ tidak memproses isi pesan.
                                      // Produsen dan konsumen harus menyepakati serialisasi.
        );
        // Opsional: tetapkan kunci bisnis untuk pelacakan pesan.
        // Kunci sebaiknya unik secara global jika memungkinkan. Gunakan untuk mencari
        // atau mengirim ulang pesan di Konsol ApsaraMQ for RocketMQ.
        msg.setKey("ORDERID_100");

        try {
            // Tetapkan waktu pengiriman menjadi 3 detik dari sekarang.
            // Nilainya adalah timestamp absolut dalam milidetik.
            // Tundaan maksimum yang dapat Anda tentukan adalah 40 hari.
            long delayTime = System.currentTimeMillis() + 3000;
            msg.setStartDeliverTime(delayTime);

            // Kirim pesan dalam mode transmisi sinkron.
            // Jika tidak ada exception yang dilempar, pesan berhasil dikirim.
            SendResult sendResult = producer.send(msg);
            if (sendResult != null) {
                System.out.println(new Date() + " Send mq message success. Topic is:"
                        + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
            }
        } catch (Exception e) {
            // Tangani kegagalan pengiriman: catat log, coba ulang, atau simpan pesan.
            System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
            e.printStackTrace();
        }

        // Matikan produsen saat aplikasi keluar.
        // Untuk skenario throughput tinggi, biarkan produsen tetap berjalan dan gunakan kembali.
        producer.shutdown();
    }
}

Ganti placeholder berikut dengan nilai aktual Anda:

Berlangganan pesan tertunda

Berlangganan pesan tertunda bekerja sama seperti berlangganan pesan biasa. Tidak diperlukan konfigurasi khusus pada konsumen—logika penundaan sepenuhnya berada di sisi produsen.

Untuk contoh kode konsumen, lihat Subscribe to messages.

Langkah selanjutnya

• Pesan terjadwal dan pesan tertunda: Pahami konsep dan batasan di balik pengiriman pesan tertunda dan terjadwal.