# SpringBoot+Kafka
**Repository Path**: wangfugui-ma/spring-boot-kafka
## Basic Information
- **Project Name**: SpringBoot+Kafka
- **Description**: SpringBoot集成kafka实现点对点消费，发布订阅模式，方法回调，消息处理异常等等的功能
- **Primary Language**: Java
- **License**: GPL-3.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No
## Statistics
- **Stars**: 6
- **Forks**: 9
- **Created**: 2021-12-10
- **Last Updated**: 2023-12-27
## Categories & Tags
**Categories**: Uncategorized
**Tags**: Kafka, SpringBoot, 消息对列, zookeeper
## README
## 啦啦啦啦啦，富贵同学又开始开坑了，出了个免费的专栏，主要给大家从0基础开始用springBoot集成第三方的插件或者功能，如果这篇专栏能帮到你，一定不要忘了点一个赞哦！！欢迎大家收藏分享
## 第一步，导入jar包
```java
org.springframework.kafka
spring-kafka
```
## 第二步，服务器上启动kafka
如果不知道怎么安装，启动请查看博主的文章
[https://blog.csdn.net/csdnerM/article/details/121851493](https://blog.csdn.net/csdnerM/article/details/121851493)
配置文件连接kafka
```java
spring.kafka.bootstrap-servers=ip:端口
spring.kafka.consumer.group-id=consumer-group
```
## 点对点消费
编写生产类
```java
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;
/**
* @author MaSiyi
* @version 1.0.0 2021/12/10
* @since JDK 1.8.0
*/
@RestController
public class KafkaProducer {
@Autowired
private KafkaTemplate kafkaTemplate;
/** 发送消息
* @Param:
* @return:
* @Author: MaSiyi
* @Date: 2021/12/10
*/
@GetMapping("/kafka/normal/{topic}/{message}")
public void sendMessage1(@PathVariable("topic") String topic, @PathVariable("message") String normalMessage) {
kafkaTemplate.send(topic, normalMessage);
}
}
```
编写消费者
```java
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
/**
* @author MaSiyi
* @version 1.0.0 2021/12/10
* @since JDK 1.8.0
*/
@Service
public class KafkaConsumer {
/** 消费监听
* @Param: [record]
* @return: void
* @Author: MaSiyi
* @Date: 2021/12/10
*/
@KafkaListener(topics = {"topictest1"})
public void message1(ConsumerRecord record){
// 消费的哪个topic、partition的消息,打印出消息内容
System.out.println("点对点消费1："+record.topic()+"-"+record.partition()+"-"+record.value());
}
}
```
测试
如果有两个方法
```java
/** 点对点消费
* @Param: [record]
* @return: void
* @Author: MaSiyi
* @Date: 2021/12/10
*/
@KafkaListener(topics = {"topictest1"})
public void message1(ConsumerRecord record){
// 消费的哪个topic、partition的消息,打印出消息内容
System.out.println("点对点消费1："+record.topic()+"-"+record.partition()+"-"+record.value());
}
/** 点对点消费
* @Param: [record]
* @return: void
* @Author: MaSiyi
* @Date: 2021/12/10
*/
@KafkaListener(topics = {"topictest1"})
public void message(ConsumerRecord record){
// 消费的哪个topic、partition的消息,打印出消息内容
System.out.println("点对点消费2："+record.topic()+"-"+record.partition()+"-"+record.value());
}
```
则只会消费一个
## 发布订阅模式
生产者是同一个，消费者如下
```java
/** 发布订阅模式
* @Param: [record]
* @return: void
* @Author: MaSiyi
* @Date: 2021/12/10
*/
@KafkaListener(topics = {"topictest2"},groupId = "1")
public void message2(ConsumerRecord record){
// 消费的哪个topic、partition的消息,打印出消息内容
System.out.println("发布订阅模式1："+record.topic()+"-"+record.partition()+"-"+record.value());
}
/** 发布订阅模式
* @Param: [record]
* @return: void
* @Author: MaSiyi
* @Date: 2021/12/10
*/
@KafkaListener(topics = {"topictest2"},groupId = "2")
public void message3(ConsumerRecord record){
// 消费的哪个topic、partition的消息,打印出消息内容
System.out.println("发布订阅模式2："+record.topic()+"-"+record.partition()+"-"+record.value());
}
```
测试
## 方法回调
生产者
```java
@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());
}
});
}
```
消费者
```java
/** 消息回调
* @Param: [record]
* @return: void
* @Author: MaSiyi
* @Date: 2021/12/10
*/
@KafkaListener(topics = {"topictest3"})
public void message4(ConsumerRecord record){
// 消费的哪个topic、partition的消息,打印出消息内容
System.out.println("回调方法："+record.topic()+"-"+record.partition()+"-"+record.value());
}
```
测试
## 事物提交
### 有异常不发送
```java
@GetMapping("/kafka/transaction1")
public void sendMessage4(){
// 声明事务：后面报错消息不会发出去
kafkaTemplate.executeInTransaction(operations -> {
operations.send("topictest4","test executeInTransaction");
throw new RuntimeException("fail");
});
}
```
接收者
```java
/** 事物
* @Param: [record]
* @return: void
* @Author: MaSiyi
* @Date: 2021/12/10
*/
@KafkaListener(topics = {"topictest4"})
public void message5(ConsumerRecord record){
// 消费的哪个topic、partition的消息,打印出消息内容
System.out.println("回调方法："+record.topic()+"-"+record.partition()+"-"+record.value());
}
```
测试
没有发送
### 有异常发送
```java
@GetMapping("/kafka/transaction2")
public void sendMessage5(){
// 不声明事务：后面报错但前面消息已经发送成功了
kafkaTemplate.send("topictest4","test executeInTransaction");
System.out.println("发送消息");
throw new RuntimeException("fail");
}
```
测试
已发送
好了，就是这么的简单，完整代码请移至[SpringBoot+Kafka ](https://gitee.com/WangFuGui-Ma/spring-boot-kafka)查看
