当前位置：首页 > news >正文

记录一下学习kafka的使用以及思路

news 来源：原创 2025/5/10 6:21:32

下面这是kafka的依赖

        <dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-kafka</artifactId></dependency>

我在学习的时候直接导入是没有导入成功的，我猜测大概的原因是我本地的仓库没有kafka的依赖，其实也就是没有下载过，所以没有指定版本号的话是在本地是找不到的所以报错，指定版本号下载了以后就可以了，后面就可以不用指定版本号了。

我们主要使用kafka里面的注解进行使用，第一个就是配置kafka，第二步就是使用kafkaTemplate进行sent方法进行发数据放到队列里面即可。

下面是配置

spring:kafka:bootstrap-servers: 156.65.20.76:9092,156.65.20.77:9092,156.65.20.78:9092 #指定Kafka集群的地址，这里有三个地址，用逗号分隔。listener:ack-mode: manual_immediate #设置消费者的确认模式为manual_immediate，表示消费者在接收到消息后立即手动确认。concurrency: 3  #设置消费者的并发数为5missing-topics-fatal: false  #设置为false，表示如果消费者订阅的主题不存在，不会抛出异常。producer:key-serializer: org.apache.kafka.common.serialization.StringSerializer  # 设置消息键的序列化器value-serializer: org.apache.kafka.common.serialization.StringSerializer #设置消息值的序列化器acks: 1  #一般就是选择1，兼顾可靠性和吞吐量 ，如果想要更高的吞吐量设置为0，如果要求更高的可靠性就设置为-1consumer:auto-offset-reset: earliest #设置为"earliest"表示将从最早的可用消息开始消费，即从分区的起始位置开始读取消息。enable-auto-commit: false #禁用了自动提交偏移量的功能，为了避免出现重复数据和数据丢失，一般都是手动提交key-deserializer: org.apache.kafka.common.serialization.StringDeserializer  # 设置消息键的反序列化器value-deserializer: org.apache.kafka.common.serialization.StringDeserializer #设置消息值的反序列化器

然后是生产者使用sent方法

executorService.submit(() -> {try {//获取内网流出带宽，并将结果封装到消息集合中dealMetricDataToMessage(ALiYunConstant.ECS_INTRANET_OUT_RATE, ALiYunConstant.INTRANET_OUT_RATE_NAME, ALiYunConstant.LW_INTRANET_OUT_RATE_CODE,startTime, endTime, instance, messages);} catch (Exception e) {log.error("获取ECS的指标数据-多线程处理任务异常！", e);} finally {latch.countDown();}});}//等待任务执行完毕latch.await();//将最终的消息集合发送到kafkaif (CollectionUtils.isNotEmpty(messages)) {for (int i = 0; i < messages.size(); i++) {if (StringUtils.isNotBlank(messages.get(i).getValue())&& "noSuchInstance".equals(messages.get(i).getValue())) {continue;}kafkaTemplate.send(topicName,  messages.get(i));}}

这里是一个线程池的发送信息的方法，但是我们只需要记住一个即可，就是kafkaTmeplate.sent(topicname,message);

然后消费者里面是这样写的

// 消费监听@KafkaListener(topics = "zhuoye",groupId ="zhuoye-aliyunmetric")public void consumeExtractorChangeMessage(ConsumerRecord<String, String> record, Acknowledgment ack){try {String value = record.value();//处理数据，存入openTsDb.................................ack.acknowledge();//手动提交}catch (Exception e){log.error("kafa-topic【zhuoye】消费阿里云指标源消息【失败】");log.error(e.getMessage());}}

主要就是@kafkalistener注解，然后就是使用数据就可以了。

然后其他的底层原理我们后面再去学习