第二步,服务器上启动kafkaorg.springframework.kafka spring-kafka
如果不知道怎么安装,启动请查看博主的文章
https://blog.csdn.net/csdnerM/article/details/121851493
配置文件连接kafka
spring.kafka.bootstrap-servers=ip:端口 spring.kafka.consumer.group-id=consumer-group点对点消费
编写生产类
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RestController; @RestController public class KafkaProducer { @Autowired private KafkaTemplatekafkaTemplate; @GetMapping("/kafka/normal/{topic}/{message}") public void sendMessage1(@PathVariable("topic") String topic, @PathVariable("message") String normalMessage) { kafkaTemplate.send(topic, normalMessage); } }
编写消费者
import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Service; @Service public class KafkaConsumer { @KafkaListener(topics = {"topictest1"}) public void message1(ConsumerRecord, ?> record){ // 消费的哪个topic、partition的消息,打印出消息内容 System.out.println("点对点消费1:"+record.topic()+"-"+record.partition()+"-"+record.value()); } }
测试
如果有两个方法
@KafkaListener(topics = {"topictest1"}) public void message1(ConsumerRecord, ?> record){ // 消费的哪个topic、partition的消息,打印出消息内容 System.out.println("点对点消费1:"+record.topic()+"-"+record.partition()+"-"+record.value()); } @KafkaListener(topics = {"topictest1"}) public void message(ConsumerRecord, ?> record){ // 消费的哪个topic、partition的消息,打印出消息内容 System.out.println("点对点消费2:"+record.topic()+"-"+record.partition()+"-"+record.value()); }
则只会消费一个
生产者是同一个,消费者如下
@KafkaListener(topics = {"topictest2"},groupId = "1") public void message2(ConsumerRecord, ?> record){ // 消费的哪个topic、partition的消息,打印出消息内容 System.out.println("发布订阅模式1:"+record.topic()+"-"+record.partition()+"-"+record.value()); } @KafkaListener(topics = {"topictest2"},groupId = "2") public void message3(ConsumerRecord, ?> record){ // 消费的哪个topic、partition的消息,打印出消息内容 System.out.println("发布订阅模式2:"+record.topic()+"-"+record.partition()+"-"+record.value()); }
测试
生产者
@GetMapping("/kafka/callbackOne/{message}") public void sendMessage2(@PathVariable("message") String callbackMessage) { kafkaTemplate.send("topictest3", callbackMessage).addCallback(success -> { // 消息发送到的topic String topic = success.getRecordmetadata().topic(); // 消息发送到的分区 int partition = success.getRecordmetadata().partition(); // 消息在分区内的offset long offset = success.getRecordmetadata().offset(); System.out.println("发送消息成功:" + topic + "-" + partition + "-" + offset); }, failure -> { System.out.println("发送消息失败:" + failure.getMessage()); }); } @GetMapping("/kafka/callbackTwo/{message}") public void sendMessage3(@PathVariable("message") String callbackMessage) { kafkaTemplate.send("topictest3", callbackMessage).addCallback(new ListenableFutureCallback>() { @Override public void onFailure(Throwable ex) { System.out.println("发送消息失败:"+ex.getMessage()); } @Override public void onSuccess(SendResult result) { System.out.println("发送消息成功:" + result.getRecordmetadata().topic() + "-" + result.getRecordmetadata().partition() + "-" + result.getRecordmetadata().offset()); } }); }
消费者
@KafkaListener(topics = {"topictest3"}) public void message4(ConsumerRecord, ?> record){ // 消费的哪个topic、partition的消息,打印出消息内容 System.out.println("回调方法:"+record.topic()+"-"+record.partition()+"-"+record.value()); }
测试
@GetMapping("/kafka/transaction1") public void sendMessage4(){ // 声明事务:后面报错消息不会发出去 kafkaTemplate.executeInTransaction(operations -> { operations.send("topictest4","test executeInTransaction"); throw new RuntimeException("fail"); }); }
接收者
@KafkaListener(topics = {"topictest4"}) public void message5(ConsumerRecord, ?> record){ // 消费的哪个topic、partition的消息,打印出消息内容 System.out.println("回调方法:"+record.topic()+"-"+record.partition()+"-"+record.value()); }
测试
没有发送
@GetMapping("/kafka/transaction2") public void sendMessage5(){ // 不声明事务:后面报错但前面消息已经发送成功了 kafkaTemplate.send("topictest4","test executeInTransaction"); System.out.println("发送消息"); throw new RuntimeException("fail"); }
测试
已发送
好了,就是这么的简单,完整代码请移至SpringBoot+Kafka 查看
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)