kafka:spring集成 kafka(springboot集成、客户端集成)
摘要
对于java的kafka集成,一般选用springboot集成kafka,但可能由于对接方kafka老旧、kafka不安全等问题导致kafak版本与spring版本不兼容,这个时候就得自己根据kafka客户端api集成了。
一、springboot集成kafka
具体官方文档地址:https://docs.spring.io/spring-boot/docs/current/reference/htmlsingle/
![](https://img-blog.csdnimg.cn/img_convert/14c63230085e91d15ecfe9c9e9d16c45.png)
1、加入依赖,spring-boot-starter-web和spring-kafka 的版本号可以看它们依赖的spring版本是否一致,这里pom依赖如下:
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<version>2.7.9</version>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.9.6</version>
</dependency>
2、添加application.yml配置,具体如下:
server:
port: 8087
spring:
mvc:
pathmatch:
matching-strategy: ant_path_matcher
kafka:
bootstrap-servers: 192.168.189.128:9092,92.168.189.128:9093,192.168.189.128:9094
consumer:
properties:
group:
id: boot-kafka
3、发送消息,由于KafkaTemplate是自动装配的,所以只要在spring的bean里注入KafkaTemplate发送消息即可,具体如下:
package com.longqi.bootkafka.controller;
import com.longqi.bootkafka.entity.MessageParam;
import com.longqi.bootkafka.entity.Wrapper;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.validation.Valid;
/**
* <p>
* 测试 前端控制器
* </p>
* @author LongQi
* @since 2021-06-23
*/
@Slf4j
@RestController
@RequestMapping("/test")
@Api(value = "TestController", tags = {"测试 API"})
public class TestController {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
private Boolean isSend = true;
@PostMapping("/kafka/sendMessage")
@ApiOperation(httpMethod = "POST", value = "发送kafka告警消息", response = Wrapper.class)
public Wrapper sendKafkaMessage(@Valid @ApiParam("参数") @RequestBody MessageParam param) {
kafkaTemplate.send(param.getTopic(), param.getMessage());
return Wrapper.ok(true);
}
}
这里用参数{"message": "asd54a6d46a4ds","topic": "device-alarm-test"}进行测试,会报如下日志:
![](https://img-blog.csdnimg.cn/img_convert/a59cf3c12d3ab255594cb731adc1c744.png)
发现会报警告:[Producer clientId=producer-1] Error while fetching metadata with correlation id 34 : {device-alarm-test=LEADER_NOT_AVAILABLE},获取主题元数据错误,这个可以忽略,查找元数据失败,spring-kafka会自动创建主题的,后续再次发送消息,是不会报这个错误的。
查看可视化工具EFAK,发现主题device-alarm-test是自动创建成功,分区数是kafka的集群配置service.properties里配置的分区9,具体如下:
![](https://img-blog.csdnimg.cn/img_convert/18ecfb57615562ac7b0a488de35216fa.png)
![](https://img-blog.csdnimg.cn/img_convert/c8e914d2622681718fdd256372f8a74b.png)
可以看到,其中一个分区保存了这个消息,logsize变成了1,说明这个消息是发送成功的。另外也可以看到主题的各分区主备消息所在的节点是不一样的。
4、接收消息,接收消息也很简单,只要在spring的bean里使用KafkaListener注解即可,具体如下:
![](https://img-blog.csdnimg.cn/img_convert/b6a77f3a663df8c7794a6cb538a0214a.png)
可视化工具也能看到该主题该消费者9个分区的消费情况,具体如下:
![](https://img-blog.csdnimg.cn/img_convert/81d3aa89215db5341688137efce9297d.png)
logSize为存入分区parttion消息数量,Offset为消费的偏移量(已消费的数量),Lag为未消费的数量(积压的数量),Owner为消费者,目前可以看到消费者为同一个,即只有1个线程在消费这9个分区的消息。
二、客户端集成kafka
直接使用kafka客户端,建议使用最新版的客户端,毕竟没有其他框架版本限制,能用最新的就用最新的,毕竟新的一般性能强也修复了bug。好比23年2月份出现的kafka安全漏洞:远程代码执行漏洞CVE-2023-25194,对现在最新版3.4.0无效,对以前大部分版本就有效。
1、添加依赖,具体如下:
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>3.4.0</version>
</dependency>
2、发送和消费消息,具体代码如下:
package com.longqi.bootkafka.config;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
/**
* @author LongQi
* @projectName boot-integration
* @description: kafka配置
* @date 2023/3/13 14:42
*/
public class KafkaConfig {
public static void main(String[] args) {
// 声明主题
String topic = "device-alarm-test";
// 创建消费者
Properties consumerConfig = new Properties();
consumerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.189.128:9092,92.168.189.128:9093,192.168.189.128:9094");
consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG,"boot-kafka");
consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer kafkaConsumer = new KafkaConsumer(consumerConfig);
// 订阅主题并循环拉取消息
kafkaConsumer.subscribe(Arrays.asList(topic));
new Thread(new Runnable() {
@Override
public void run() {
while (true){
ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(10000));
for(ConsumerRecord<String, String> record:records){
System.out.println(record.value());
}
}
}
}).start();
// 创建生产者
Properties producerConfig = new Properties();
producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.189.128:9092,92.168.189.128:9093,192.168.189.128:9094");
producerConfig.put(ProducerConfig.CLIENT_ID_CONFIG,"boot-kafka-client");
producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer producer = new KafkaProducer<>(producerConfig);
// 给主题发送消息
producer.send(new ProducerRecord<>(topic, "hello,"+System.currentTimeMillis()));
}
}
最后可以看到打印消息如下:
![](https://img-blog.csdnimg.cn/img_convert/fd2b1beb275122d57bf30866ce6b4922.png)
成功接收到消息并打印
相关文章
- 亚马逊云科技携手宝马集团交付全新的云端车辆数据平台
- IT领导者对2023年人工智能的七个预测
- 如何为您的企业选择正确的数据布线
- ChatGPT是一项技术突破吗?
- 当今网络犯罪的兴起 如何应对网络安全威胁
- 如何定义现代化裸金属云
- 今年很火的AI绘画怎么玩
- 小编和ChatGPT聊了下智能运维,大家看看能不能把专家替了?
- 腾讯云发布三款云原生新品 持续加码云原生布局
- 刚插上网线,电脑怎么知道自己的IP是什么?
- 企业准备进行云迁移时需要问的四个重要问题
- 亚马逊云科技:引领业界践行可持续发展
- 马斯克认为ChatGPT的一些回应令人担忧
- 人工智能如何重塑制造业的未来?
- Chrome DevTools 的 Network 还能这么用?
- 从盛业看如何打造供应链金融现代化
- 为什么6G是下一代无线通信
- 亚马逊云科技专家视点:现代化应用的安全可证明性——构建最高标准云安全的必由之路
- 论控制面板在云数据仓库架构中的重要性
- 认真聊一次什么是光纤终端盒?