Controller
/**
* 回调方法中监控消息是否发送成功 或 失败时做补偿处理
*/
@GetMapping("/callback/{message}")
public String sendMessageCallback(@PathVariable("message") String callbackMessage) {
kafkaProducer.sendMessageCallback(callbackMessage);
return "niu bi a";
}
Service :kafka producer
/**
* callback
*/
public void sendMessageCallback(String callbackMessage) {
kafkaTemplate.send("topic1", callbackMessage).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
@Override
public void onFailure(Throwable ex) {
System.out.println("发送消息失败:" + ex.getMessage());
}
@Override
public void onSuccess(SendResult<String, Object> result) {
System.out.println("发送消息成功:" + result.getRecordMetadata().topic() + "-"
+ result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset());
}
});
}
}
消费者:
@Service
public class KafkaConsumer3 {
@KafkaListener(topics = "topic1", groupId = KafkaConstants.KAFKA_GROUP_ID_THREE)
public void receivetopicmessage(String message) {
System.out.println("KafkaConsumer3 ---> receivetopicmessage:接收消息,内容为:" + message);
}
}
结果:
=====