springboot集成kafka是比较简单的是事情,但是kafka发送消息的失败回调在日常工作中,如果不容忍消息丢失的话,发送失败需要再次发送或者放到数据库中用任务重推。
以下是演示用的发送类代码
kafkaTemplate.send(topic, JSON.toJSONString(jsonObject)).addCallback(new ListenableFutureCallback<SendResult>() { @Override public void onFailure(Throwable throwable) { if (throwable instanceof KafkaProducerException) { String value = (String) ((KafkaProducerException) throwable).getProducerRecord().value(); log.info("{} get throwable msg:{}", value, throwable.getMessage()); } else { log.info("get throwable msg:{}", throwable.getMessage()); } } @Override public void onSuccess(SendResulto) { log.info("{}, success", o.getProducerRecord().value()); } });