消息队列

消息队列

目录概述重试一定次数（消息丢失）加入到死讯队列（消息不丢失）总结

概述

在kafka中，或者是说在任何消息队列中都有个消费顺序的问题。为了保证一个队列顺序消费，当当中一个消息消费异常时，必将影响后续队列消息的消费，这样业务岂不是卡住了。比如笔者举个最简单的例子：我发送1-100的消息，在我的处理逻辑当中 msg%5==0我就进行 int i=1/0操作，这必将抛异常，一直阻塞在msg=5上，后面6-100无法消费。下面笔者给出解决方案。

重试一定次数（消息丢失）

@KafkaHandler

@KafkaListener(topics = {"quickstart-events"},groupId = "test-consumer-group-2", concurrency = "1")

public void test6(String msg){

businessProcess(msg);

}

private void businessProcess(String msg){

System.out.println("接收到消息:" + msg + "--" + System.currentTimeMillis() + "---" + Thread.currentThread().hashCode());

if (Integer.valueOf(msg) % 5 == 0) {

int i = 1 / 0;

}

}

说明：如果读者使用的是java客户端，也就是spring进行实现，那么在不做任何处理的情况下，会自动重试10次，然后消息会被直接处理掉。也就是说如果你的业务允许消息丢失，那么你不需要额外的编码处理

加入到死讯队列（消息不丢失）

消费端代码：

//1.启用手动提交offset

//2.配置errorHandler，用来加入到死讯队列

//3.不管业务处理是否处理异常还是正常都提交offset

@KafkaHandler

@KafkaListener(topics = {"quickstart-events"},groupId = "test-consumer-group-2",

errorHandler ="kafkaListenerErrorHandler", concurrency = "1")

public void test6(String msg,Acknowledgment ack){

try {

businessProcess(msg);

}finally {

//手动提交

ack.acknowledge();

}

}

//1.专门处理死讯队列消息，都是topicName+.DLT的主题

//2.死讯队列里，只有消费成功的才提交offset,否则等待bug修复完上线，继续处理

@KafkaHandlerKuNqqaI

@KafkaListener(topics = {"quickstart-events.DLT"},groupId = "test-consumer-group-2", concurrency = "1")

public void test7(String msg,Acknowledgment ack){

try {

businessProcess(msg);

ack.acknowledge();

KuNqqaI }catch (Exception e){

e.printStackTrace();

}

}

//业务代码

private void businessProcess(String msg){

System.out.println("接收到消息:" + msg + "--" + System.currentTimeMillis() + "---" + Thread.currentThread().hashCode());

if (Integer.valueOf(msg) % 5 == 0) {

int i = 1 / 0;

}

}

异常处理器

//1.向容器注册一个KafkaListenerErrorHandler类型的bean

//2.该bean就是当处理消息异常的时候，将消息加入到.DLT主题中

@Component("kafkaListenerErrorHandler")

public class KafkaListenerErrorHandlerTest implements KafkaListenerErrorHandler {

@Autowired

private KafkaTemplate kafkaTemplate;

private static final String TOPIC_DLT=".DLT";

@Override

public Object handleError(Message> message, ListenerExecutionFailedException exception) {

System.out.println("消费失败消息:"+message.toString());

//获取消息处理异常主题

MessageHeaders headers = message.getHeaders();

String topic=headers.get("kafka_receivedTopic")+TOPIC_DLT;

//放入死讯队列

kafkaTemplate.send(topic,message.getPayload());

return message;

}

}

效果图：

说明：以上基本上就是使用死讯队列的方案，也许读者会觉得这样编码复杂度很高，但其实不用担心，其实上面这些代码基本上是使用死讯队列的模板代码，在成熟一点的公司，一般会使用上述代码进行简单封装，这里笔者给个思路，有兴趣同学可以实现一下。我们其实可以使用aop思想，进行自定义一个@EnableDLT这样的注解去实现，这样上面这个方案使用起来是不是就简单优雅了。之前笔者在开发过程中使用过亚马逊的消息队列服务，也不过是这样实现罢了。

总结

本篇文章就到这里了，希望可以给你带来一些帮助，也希望您能够多多关注我们的更多内容！

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。