个人网站优秀,app和小程序的区别,国外有名的设计网站,怎么制作网页步骤一、背景
写这边篇文章的目的#xff0c;是记录我在集成kafka客户端遇到的一些问题#xff0c;文章会记录整个接入的过程#xff0c;其中会遇到几个坑#xff0c;如果需要最终版本#xff0c;直接看最后一节就行了#xff0c;感觉Spring-Kafka的文档太少了#xff0c;如…一、背景
写这边篇文章的目的是记录我在集成kafka客户端遇到的一些问题文章会记录整个接入的过程其中会遇到几个坑如果需要最终版本直接看最后一节就行了感觉Spring-Kafka的文档太少了如果采用SpringBoot集成的方式接入一不小可能就会踩坑
二、操作步骤
1 添加依赖 dependencygroupIdorg.springframework.kafka/groupIdartifactIdspring-kafka/artifactId/dependency
2 添加配置文件
spring:profiles:active: devapplication:name: goods-centerkafka:bootstrap-servers: 192.168.31.114:9092producer:acks: alltimeout.ms: 5000# 值序列化使用Jsonvalue-serializer: org.springframework.kafka.support.serializer.JsonSerializerkey-serializer: org.apache.kafka.common.serialization.LongSerializerenable:idempotence: true # 默认为Trueconsumer:group-id: goods-centervalue-deserializer: org.springframework.kafka.support.serializer.JsonDeserializerkey-deserializer: org.apache.kafka.common.serialization.LongDeserializerenable-auto-commit: false # 取消自动提交
3 生产者代码
package com.ychen.goodscenter.fafka;import com.ychen.goodscenter.vo.req.SubmitOrderReq;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;Component
public class MessageProducer {private final Logger logger LoggerFactory.getLogger(this.getClass());Autowiredprivate KafkaTemplateLong, SubmitOrderReq kafkaTemplate;public void sendOrderMessage(SubmitOrderReq msg) {kafkaTemplate.send(TopicConstants.ORDER_MESSAGE_TOPIC, msg.getOrderId(), msg);logger.info(order-message-topic message sent, orderId: {}, msg.getOrderId());}
}4 消费者代码
package com.ychen.goodscenter.fafka;import com.ychen.goodscenter.service.OrderService;
import com.ychen.goodscenter.vo.req.SubmitOrderReq;
import org.apache.kafka.clients.consumer.Consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;Component
public class MessageListener {private final Logger logger LoggerFactory.getLogger(this.getClass());Autowiredprivate OrderService orderService;KafkaListener(topics order-message-topic)public void processMessage(ConsumerLong, SubmitOrderReq consumer, SubmitOrderReq submitOrderReq) {try {logger.info(order-message-topic message received, orderId: {}, submitOrderReq.getOrderId());orderService.submitOrder(submitOrderReq);// 同步提交consumer.commitSync();logger.info(order-message-topic message acked: orderId: {}, submitOrderReq.getOrderId());} catch (DuplicateKeyException dupe) {// 处理异常情况logger.error(order-message-topic message processMessage data DuplicateKeyException, dupe);// 重复数据忽略掉同步提交consumer.commitSync();} catch (Exception e) {// 处理异常情况logger.error(order-message-topic message processMessage error, e);}}
}
三、开始踩坑了
1 添加信任自己包
Caused by: java.lang.IllegalArgumentException: The class com.ychen.goodscenter.vo.req.SubmitOrderReq is not in the trusted packages: [java.util, java.lang]. If you believe this class is safe to deserialize, please provide its name. If the serialization is only done by a trusted source, you can also enable trust all (*).at org.springframework.kafka.support.mapping.DefaultJackson2JavaTypeMapper.getClassIdType(DefaultJackson2JavaTypeMapper.java:129)at org.springframework.kafka.support.mapping.DefaultJackson2JavaTypeMapper.toJavaType(DefaultJackson2JavaTypeMapper.java:103)at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize(JsonDeserializer.java:572)at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:1429)... 15 common frames omitted
原因 因为我在消费消息时用了json序列化需要给这个序列化添加信任自己包不加json序列号会报错
解决方法添加配置
spring:kafka:consumer:properties:spring.json.trusted.packages: com.ychen.**
解决途径百度
2 consumer.commitSync(); 无效
问题发现当我正在批量消费消息时强制重启应用进程发现有部分消息丢失了没有处理
我发了5000个样本请求最后只生成了4912 个订单中途强制重启了2次 问题分析有2中可能
第一种之前配置的enable-auto-commit: false 是无效的。
第二种 consumer.commitSync(); 一次将批量拉取的offset提交了
问题排查
通过在 consumer.commitSync(); 代码之前和之后分别打一个断点然后发送一批数据
consumer.commitSync(); 之前 consumer.commitSync(); 之后 结果发生了突变说明是consumer.commitSync();执行之后引发的offset突变
翻阅源码 总体而言通过官方文档和源代码我们可以确定 commitSync() 提交的是已经成功拉取到的消息的最大 offset而不是当前正在处理的消息的 offset。
3 缺少AckMode 配置
既然consumer.commitSync();无法在批量处理消息的环境保证消息不丢失那么需要寻找新的解决方案
在org.springframework.kafka.annotation.KafkaListener 类的注释上面有写到可以使用org.springframework.kafka.support.Acknowledgment
然后我们消费者的代码改造后为
package com.ychen.goodscenter.fafka;import com.ychen.goodscenter.service.OrderService;
import com.ychen.goodscenter.vo.req.SubmitOrderReq;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;Component
public class MessageListener {private final Logger logger LoggerFactory.getLogger(this.getClass());Autowiredprivate OrderService orderService;KafkaListener(topics order-message-topic)public void processMessage(ConsumerRecordLong, SubmitOrderReq record, Acknowledgment acknowledgment) {try {logger.info(order-message-topic message received, orderId: {}, record.value().getOrderId());orderService.submitOrder(record.value());// 同步提交acknowledgment.acknowledge();logger.info(order-message-topic message acked: orderId: {}, record.value().getOrderId());} catch (DuplicateKeyException dupe) {// 处理异常情况logger.error(order-message-topic message processMessage data DuplicateKeyException, dupe);// 重复数据忽略掉同步提交acknowledgment.acknowledge();} catch (Exception e) {// 处理异常情况logger.error(order-message-topic message processMessage error, e);}}
}
Caused by: java.lang.IllegalStateException: No Acknowledgment available as an argument, the listener container must have a MANUAL AckMode to populate the Acknowledgment.
这里我也有点不懂了明明已经配置自动提交了还需要配置 ack-mode: MANUAL既然他说要那就配置吧
在application.yml 增加配置
spring:kafka:listener:ack-mode: MANUAL
现在准备2000个样本然后让消费者实例强制重启2次看看数据库的订单数量是否为2000条 现在正确了支持系统宕机仍然不丢失消息了
四、最终的配置文件和消费者代码
1 配置文件
spring:profiles:active: devapplication:name: goods-centerkafka:bootstrap-servers: 192.168.31.114:9092listener:ack-mode: MANUALproducer:acks: alltimeout.ms: 5000# 值序列化使用Jsonvalue-serializer: org.springframework.kafka.support.serializer.JsonSerializerkey-serializer: org.apache.kafka.common.serialization.LongSerializerenable:idempotence: true # 默认为Trueconsumer:properties:spring.json.trusted.packages: com.ychen.** # 信任自己包不加json序列号会报错group-id: goods-centervalue-deserializer: org.springframework.kafka.support.serializer.JsonDeserializerkey-deserializer: org.apache.kafka.common.serialization.LongDeserializerenable-auto-commit: false # 取消自动提交
2 消费者代码
package com.ychen.goodscenter.fafka;import com.ychen.goodscenter.service.OrderService;
import com.ychen.goodscenter.vo.req.SubmitOrderReq;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;Component
public class MessageListener {private final Logger logger LoggerFactory.getLogger(this.getClass());Autowiredprivate OrderService orderService;KafkaListener(topics order-message-topic)public void processMessage(ConsumerRecordLong, SubmitOrderReq record, Acknowledgment acknowledgment) {try {logger.info(order-message-topic message received, orderId: {}, record.value().getOrderId());orderService.submitOrder(record.value());// 同步提交acknowledgment.acknowledge();logger.info(order-message-topic message acked: orderId: {}, record.value().getOrderId());} catch (DuplicateKeyException dupe) {// 处理异常情况logger.error(order-message-topic message error DuplicateKeyException, dupe);// 重复数据忽略掉同步提交acknowledgment.acknowledge();} catch (Exception e) {// 处理异常情况logger.error(order-message-topic message error unknown , e);}}
}
