zoukankan      html  css  js  c++  java
  • rocketmq延时消息

    rocketmq提供一种延时消息的解决方案,就是在特定的时间到了,消息才会被投递出去供consumer消费。

    总体来是简单的场景是满足了,但是需要注意的是延时的时间是需要按照默认配置的延时级别去配置的,而不是随意设置消息的延时时间。

    如果想不受延时级别的约束 可以参考之前的一遍文章http://blog.seoui.com/2017/08/19/delayqueue/

    默认的延迟级别

    messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

    这个配置下标从1开始 比如级别2是延时5秒、级别5是延时1分钟。默认配置在不满足需求的情况下,可以在broker配置文件加入messageDelayLevel参数覆盖默认的延时级别配置。

    示例

    和普通的消息不同之处在于Producer在发送消息的时候 需要设置message.setDelayTimeLevel();延迟级别方法。其他参数和消费端的写法并与不同之处。

    Producer
    public class ScheduledMessageProducer {
        public static void main(String[] args) throws Exception {
            DefaultMQProducer producer = 
            new DefaultMQProducer("ExampleProducerGroup");
            producer.start();
            int totalMessagesToSend = 100;
            for (int i = 0; i < totalMessagesToSend; i++) {
                Message message = new Message("TestTopic",
                ("Hello scheduled message " + i).getBytes());
             //延时的级别为3 对应的时间为10s 就是发送后延时10S在把消息投递出去
                message.setDelayTimeLevel(3);
                producer.send(message);
            }
            producer.shutdown();
        }
    }
    Consumer
    public class ScheduledMessageConsumer {
        public static void main(String[] args) throws Exception {
            DefaultMQPushConsumer consumer = 
            new DefaultMQPushConsumer("ExampleConsumer");
            consumer.subscribe("TestTopic", "*");
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages,
             ConsumeConcurrentlyContext context) {
                    for (MessageExt message : messages) {
                        System.out.println("ok!");
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
        }
    }

    笑笑笑技术圈

  • 相关阅读:
    OGG for sqlserver engryption && insert/delete
    MySQL中SQL_CALC_FOUND_ROWS的用法
    VMWARE 虚拟机新增硬盘,格式化分区,并挂载
    Mycat实战之数据迁移(oracle -- mysql)
    MYCAT实战之分片迁移
    mycat实战之性能测试
    Mycat实战之主键数据库自增方式
    对象的创建与克隆
    c#调用c++的dll,错误篇
    java 模拟消息的发送功能
  • 原文地址:https://www.cnblogs.com/peachyy/p/9406546.html
Copyright © 2011-2022 走看看