通过消息的分区和位移来判断消息是否被消费。或者对性能要求不高的话，可以考虑接入事务来处理。
kafka消息幂等:
Producer 发送事务消息:
先设置幂等参数: enable.idempotence = true
然后在设置transaction.id
然后在发送消息的时候使用如下方法:
try {
producer.beginTransaction();
producer.send();
producer.commitTransaction()
} catch(Exception e) {
producer.abortTransaction();
}
在producer端发送的消息，即使是没有commit的消息，其实再broker端都是产生一个log，所以，对consumer端，如果没有做相应的处理的话，
还是会接收到这些消息。所以，对consumer端，我们需要做一些额外的处理:
consumer设置: isolation.level
- read_uncommitted: 默认参数，能够读取到producer端的提交和未提交的消息。
- read_committed: 只能读取到producer端提交的消息。
网友评论