kafkaTemplate.send(
invoiceEtcTradeTopic,
jsonMsg
).addCallback(new ListenableFutureCallback>(){
@Override
public void onFailure(Throwable ex) {
log.error("发送消息失败:{}",ex.getMessage());
}
@Override
public void onSuccess(SendResult result) {
String topicName = result.getRecordMetadata().topic();
int num = result.getRecordMetadata().partition();
long offset = result.getRecordMetadata().offset();
log.info("发送消息成功:topic={},partition={},offset={}",topicName,num,offset);
if(numOrder[0] == 1){
log.info("");
}else{
log.info("");
}
}
});
生产问题解决
生产者防止消息丢失,设置表。标识flag来标识是否发送成功。
利用spring的ListenableFutureCallback监听,onsuccess时修改表标识为1成功,onFailure时修改标识为0发送失败。
为0的定时重发消息,重发的消息成功了进onsuccess中,然后改标识
消费者接收之前存表。消息来了先从表里查询,没查询到数据,则重新消费。消费成功,手动提交偏移量。
查到数据了,直接手动提交偏移量。
多次消费不成功,设置阈值。
偏移量1 手动提交 当 auto.commit.enable 设置为false时,表示kafak的offset由消费者手动维护,spring-kafka提供了通过ackMode的值表示不同的手动提交方式;
ackMode有以下7种值:
// 当每一条记录被消费者监听器(ListenerConsumer)处理之后提交
RECORD,
// 当每一批poll()的数据被消费者监听器(ListenerConsumer)处理之后提交
BATCH,
// 当每一批poll()的数据被消费者监听器(ListenerConsumer)处理之后,距离上次提交时间大于TIME时提交
TIME,
// 当每一批poll()的数据被消费者监听器(ListenerConsumer)处理之后,被处理record数量大于等于COUNT时提交
COUNT,
// TIME | COUNT 有一个条件满足时提交
COUNT_TIME,
// 当每一批poll()的数据被消费者监听器(ListenerConsumer)处理之后, 手动调用Acknowledgment.acknowledge()后提交
MANUAL,
// 手动调用Acknowledgment.acknowledge()后立即提交
MANUAL_IMMEDIATE,
如果设置 AckMode 模式为 MANUAL 或者 MANUAL_IMMEDIATE,则需要对监听消息的方法中,引入 Acknowledgment 对象参数,并调用 acknowledge() 方法进行手动提交;如果没有提交程序重启后会重复消费;其他模式满足条件就会提交(不是严格意义上的手动),不需要调用acknowledge() 。
2 回滚offet 处理数据报错时记下offset,重置offset就会重复消费。
TopicPartition topicPartition = new TopicPartition(topic, partition);
ack设置ack 应答机制 对于某些不太重要的数据,对数据的可靠性要求不是很高,能够容忍数据的少量丢失, 所以没必要等 ISR 中的 follower 全部接收成功。 所以 Kafka 为用户提供了三种可靠性级别,用户根据对可靠性和延迟的要求进行权衡, 选择以下的配置。 acks 参数配置: acks: 0:producer 不等待 broker 的 ack,这一 *** 作提供了一个最低的延迟,broker 一接收到还 没有写入磁盘就已经返回,当 broker 故障时有可能丢失数据; 1:producer 等待 broker 的 ack,partition 的 leader 落盘成功后返回 ack,如果在 follower 同步成功之前 leader 故障,那么将会丢失数据 -1(all):producer 等待 broker 的 ack,partition 的 leader 和 follower (ISRL里的follower,不是全部的follower)全部落盘成功后才 返回 ack。但是如果在 follower 同步完成后,broker 发送 ack 之前,leader 发生故障,那么会造成数据重复
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)