ApsaraMQ for RocketMQ：Kirim dan terima pesan terjadwal

Informasi latar belakang

Pesan terjadwal dikonsumsi setelah cap waktu yang telah ditentukan sebelumnya. Pesan ini dapat digunakan dalam skenario yang memerlukan jendela waktu antara produksi dan konsumsi pesan, atau untuk memicu tugas terjadwal melalui pesan.

Untuk informasi lebih lanjut tentang istilah yang digunakan dalam pesan terjadwal serta tindakan pencegahan yang perlu diperhatikan, lihat Pesan Terjadwal dan Pesan Tertunda.

Prasyarat

Sebelum memulai, pastikan langkah-langkah berikut telah dilakukan:

• SDK untuk Java telah diinstal. Untuk informasi lebih lanjut, lihat Persiapkan Lingkungan.

• Sumber daya yang ingin ditentukan dalam kode telah dibuat di konsol ApsaraMQ for RocketMQ, termasuk instance, topik, dan grup konsumen. Untuk informasi lebih lanjut, lihat Buat Sumber Daya.

• Pasangan AccessKey dari akun Alibaba Cloud Anda telah diperoleh. Untuk informasi lebih lanjut, lihat Buat Pasangan AccessKey.

• Opsional. Pengaturan logging telah dikonfigurasi. Untuk informasi lebih lanjut, lihat Pengaturan Logging.

Kirim pesan terjadwal

Untuk informasi lebih lanjut tentang contoh kode rinci, lihat ApsaraMQ for RocketMQ Repositori Kode.

Contoh kode berikut menunjukkan cara mengirim pesan terjadwal menggunakan SDK klien TCP untuk Java:

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.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Properties;

public class ProducerDelayTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // Pastikan variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_ID dan ALIBABA_CLOUD_ACCESS_KEY_SECRET telah dikonfigurasi. 
        // ID AccessKey yang digunakan untuk otentikasi. 
        properties.put(PropertyKeyConst.AccessKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
        // Rahasia AccessKey yang digunakan untuk otentikasi. 
        properties.put(PropertyKeyConst.SecretKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
        // Titik akhir TCP. Anda dapat memperoleh titik akhir di bagian titik akhir TCP halaman Detail Instance di konsol ApsaraMQ for RocketMQ. 
        properties.put(PropertyKeyConst.NAMESRV_ADDR,
          "XXX");
        Producer producer = ONSFactory.createProducer(properties);
        // Sebelum Anda mengirim pesan, panggil metode start() hanya sekali untuk memulai produser. 
        producer.start();
        Message msg = new Message(
                // Topik tempat pesan diproduksi. 
                "Topic",
                // Tag pesan. Tag pesan mirip dengan tag Gmail dan digunakan oleh konsumen untuk menyaring pesan di broker ApsaraMQ for RocketMQ. 
                "tag",
                // Tubuh pesan. Tubuh pesan adalah data dalam format biner. ApsaraMQ for RocketMQ tidak memproses tubuh pesan. Produser dan konsumen harus menyetujui metode untuk serialisasi dan deserialisasi tubuh pesan. 
                "Halo MQ".getBytes());
        // Kunci pesan. Kunci adalah atribut spesifik bisnis dari pesan dan harus unik secara global jika memungkinkan. 
        // Jika Anda tidak dapat menerima pesan seperti yang diharapkan, Anda dapat menggunakan kunci untuk menanyakan pesan di konsol ApsaraMQ for RocketMQ. 
        // Catatan: Anda dapat mengirim dan menerima pesan bahkan jika Anda tidak menentukan kunci. 
        msg.setKey("ORDERID_100");

        try {
            // Cap waktu yang menunjukkan waktu ketika broker mengantarkan pesan ke konsumen. Unit: milidetik. Misalnya, jika Anda menetapkan parameter ini ke 2016-03-07 16:21:00, broker akan mengantarkan pesan pada pukul 16:21:00 tanggal 7 Maret 2016. Nilainya harus lebih dari waktu saat ini. Jika Anda menetapkan parameter ini ke waktu yang lebih awal dari waktu saat ini, pesan akan langsung dikirimkan ke konsumen. 
            long timeStamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2016-03-07 16:21:00").getTime();

            msg.setStartDeliverTime(timeStamp);
            // Kirim pesan. Jika tidak ada pengecualian yang dilempar, pesan dikirim. 
            SendResult sendResult = producer.send(msg);
            System.out.println("Message Id:" + sendResult.getMessageId());
        }
        catch (Exception e) {
            // Tentukan logika yang ingin Anda gunakan untuk mengirim ulang atau mempertahankan pesan jika pesan gagal dikirim dan perlu dikirim lagi. 
            System.out.println(new Date() + " Kirim pesan mq gagal. Topik adalah:" + msg.getTopic());
            e.printStackTrace();
        }

        // Sebelum Anda keluar dari aplikasi, hancurkan produser. 
        // Catatan: Memori dapat dihemat jika Anda menghancurkan produser. Jika Anda perlu sering mengirim pesan, jangan hancurkan produser. 
        producer.shutdown();
    }
}       

Berlangganan pesan terjadwal

Contoh kode untuk berlangganan pesan terjadwal sama dengan berlangganan pesan normal. Untuk informasi lebih lanjut, lihat Berlangganan Pesan.