其他分享
首页 > 其他分享> > 4.死信队列

4.死信队列

作者:互联网

目录

死信队列

当消息未被确认消费,在满足指定规则情况下,会进入死信队列,运维人员会对一直不能消费的消息进行处理

生产者

/**
     * 同步-生产消息
     * @param msg
     * @throws PulsarClientException
     */
    @GetMapping("/deal/sendMsgBySync")
    public MessageId sendMsgBySync(String msg) throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();

        Producer<byte[]> producer = pulsarFactory.newProducer()
                .topic("my-deal-topic")
                .create();
        // 然后你就可以发送消息到指定的broker 和topic上:
        return producer.send(msg.getBytes());
    }

消费者

@Bean
    public void comsumerByDealListener() throws Exception {
        MessageListener myMessageListener = (consumer, msg) -> {
            String anObject = new String(msg.getData());
            System.out.println("收到消息了");
           // int i=1/0;//故意报错
            try {
                System.out.println("Message received: " + anObject);
//                consumer.acknowledge(msg);//故意不确认消费
            } catch (Exception e) {
//                consumer.negativeAcknowledge(msg);//故意不确认消费
            }
        };
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        pulsarFactory.newConsumer()
                .topic("my-deal-topic")
                .subscriptionName("my-subscriptionByListener")
                .ackTimeout(3, TimeUnit.SECONDS)//指定超时多久未确认消息后判断为异常
                .subscriptionType(SubscriptionType.Shared)
                .messageListener(myMessageListener)
                .deadLetterPolicy(DeadLetterPolicy.builder()
                        .maxRedeliverCount(4)//指定消费失败重试次数,超过次数进入死信队列
                        .deadLetterTopic("dealtopic")//死信队列名,不指定的话默认:{TopicName}-{Subscription}-DLQ
                        .build())
                .subscribe();
    }

死信队列

死信队列可使用监听方案

@GetMapping("/deal/cunsumerDeadLetterTopic")
    public void cunsumerDeadLetterTopic() throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        Consumer<byte[]> consumer = pulsarFactory.newConsumer()
                .topic("dealtopic")
                .subscriptionName("mysubscriptiondlq")
                .subscribe();
        Message<byte[]> receive = consumer.receive();
        System.out.println(new String(receive.getData()));
        consumer.acknowledge(receive);//确认消息被消费
        consumer.close();
    }

验证流程

1.打开死信队列监听/deal/cunsumerDeadLetterTopic(这步无所谓)

2.调用生产消息接口deal/sendMsgBySync

3.消费者comsumerByDealListener中有“故意报错”和注释掉的“acknowledge”,导致必然无法消费

代码下载

代码见此dealTopic包下

标签:deal,队列,pulsarFactory,死信,msg,consumer
来源: https://www.cnblogs.com/rb2010/p/15314691.html