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包下