delaymq高性能的任意延時消息隊列

import com.aliyun.openservices.ons.api.*;
import com.aliyun.openservices.shade.org.apache.commons.lang3.time.DateFormatUtils;

import java.util.Date;
import java.util.Properties;

public class ProducerDelayTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // AccessKey ID阿里云身份驗證，在阿里云RAM控制臺創(chuàng)建。
        properties.put(PropertyKeyConst.AccessKey, "XXX");
        // AccessKey Secret阿里云身份驗證，在阿里云RAM控制臺創(chuàng)建。
        properties.put(PropertyKeyConst.SecretKey, "XXX");
        // 設(shè)置TCP接入域名，進(jìn)入消息隊列RocketMQ版控制臺實例詳情頁面的接入點區(qū)域查看。
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "localhost:9876");
        

        Producer producer = ONSFactory.createProducer(properties);
        // 在發(fā)送消息前，必須調(diào)用start方法來啟動Producer，只需調(diào)用一次即可。
        producer.start();

         {
            Message msg = new Message(
                    // 您在消息隊列RocketMQ版控制臺創(chuàng)建的Topic。
                    "TopicTest",
                    // Message Tag,可理解為Gmail中的標(biāo)簽，對消息進(jìn)行再歸類，方便Consumer指定過濾條件在消息隊列RocketMQ版服務(wù)器過濾。
                    "TagA",
                    // Message Body可以是任何二進(jìn)制形式的數(shù)據(jù)，消息隊列RocketMQ版不做任何干預(yù)，需要Producer與Consumer協(xié)商好一致的序列化和反序列化方式。
                    "演示15秒鐘>>> ".getBytes());
            // 設(shè)置代表消息的業(yè)務(wù)關(guān)鍵屬性，請盡可能全局唯一。
            // 以方便您在無法正常收到消息情況下，可通過控制臺查詢消息并補發(fā)。
            // 注意：不設(shè)置也不會影響消息正常收發(fā)。
            msg.setKey("ORDERID_100e");
            try {
                // 延時消息，單位毫秒（ms），在指定延遲時間（當(dāng)前時間之后）進(jìn)行投遞，例如消息在15秒后投遞。
                long delayTime = System.currentTimeMillis() + 15000;
                System.out.println("發(fā)送時間>>" + DateFormatUtils.format(new Date(), "yyyy-MM-dd HH????ss"));

                // 設(shè)置消息需要被投遞的時間。
                msg.setStartDeliverTime(delayTime);

                SendResult sendResult = producer.send(msg);
                // 同步發(fā)送消息，只要不拋異常就是成功。
                if (sendResult != null) {
                    System.out.println(DateFormatUtils.format(new Date(), "yyyy-MM-dd HH????ss") + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
                }
            } catch (Exception e) {
                // 消息發(fā)送失敗，需要進(jìn)行重試處理，可重新發(fā)送這條消息或持久化這條數(shù)據(jù)進(jìn)行補償處理。
                System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                e.printStackTrace();
            }
        }
        // 在應(yīng)用退出前，銷毀Producer對象。
        // 注意：如果不銷毀也沒有問題。
        producer.shutdown();
    }
}