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

SpringBoot配置多个kafka配置

news 2025/7/28 5:07:33

参考文章：SpringBoot整合kafka配置多个kafka配置_springboot配置多个kafka-CSDN博客

引入依赖

       <dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.7.14</version></dependency>

yml配置

有几个就配置几个 ，这里只配置两个

因为我在本地启动了两个kafka，所以bootstrap-servers这里分别指向两个不同端口的kafka地址

几个属性详细解释：

acks

 #acks = 0：设置成 表示 producer 完全不理睬 leader broker 端的处理结果。此时producer 发送消息后立即开启下 条消息的发送，根本不等待 leader broker 端返回结果
#acks= all 或者－1 ：表示当发送消息时， leader broker 不仅会将消息写入本地日志，同时还会等待所有其他副本都成功写入它们各自的本地日志后，才发送响应结果给，消息安全但是吞吐量会比较低。
#acks = 1：默认的参数值。 producer 发送消息后 leader broker 仅将该消息写入本地日志，然后便发送响应结果给producer ，而无须等待其他副本写入该消息。折中方案，只要leader一直活着消息就不会丢失，同时也保证了吞吐量

auto-offset-reset:

#当kafka中没有初始offset或offset超出范围时将自动重置offset
# earliest:重置为分区中最小的offset;
# latest:重置为分区中最新的offset(消费分区中新产生的数据);
# none:只要有一个分区不存在已提交的offset,就抛出异常;

spring:kafka:# 第一个kafka的配置first:bootstrap-servers: 127.0.0.1:9092 # 这个是kafka的地址,对应server.properties中配置的producer:                         # 生产者配置key-serializer: org.apache.kafka.common.serialization.StringSerializer  # Kafka提供的序列化和反序列化类value-serializer: org.apache.kafka.common.serialization.StringSerializerbatch-size: 16384        # 批量大小buffer-memory: 33554432  # 生产端缓冲区大小retries: 10              # 消息发送重试次数acks: -1                  # 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)properties:linger:ms: 2000             # 提交延迟consumer:  # 消费者配置key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializergroup-id: testGroup             # 默认的消费组IDenable-auto-commit: true        # 是否自动提交offset auto-commit-interval: 2000      # 提交offset延时(接收到消息后多久提交offset)max-poll-records: 500           # 单次拉取消息的最大条数,根据业务需要配置auto-offset-reset: latestlistener:missing-topics-fatal: false # 当kafka启动的时候如果未找到对应topic不报错properties:session:timeout:ms: 120000 # 消费会话超时时间（超过这个时间 consumer 没有发送心跳，就会触发 rebalance 操作）request:timeout:ms: 18000  # 消费请求的超时时间# 第二个kafka的配置second:bootstrap-servers: 127.0.0.1:9093  # 这个是kafka的地址,对应server.properties中配置的producer:                          # 生产者配置key-serializer: org.apache.kafka.common.serialization.StringSerializer  # Kafka提供的序列化和反序列化类value-serializer: org.apache.kafka.common.serialization.StringSerializerbatch-size: 16384         # 批量大小buffer-memory: 33554432   # 生产端缓冲区大小retries: 10               # 消息发送重试次数acks: -1                   # 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)properties:linger:ms: 2000              # 提交延迟consumer: # 消费者配置key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializergroup-id: testGroup             # 默认的消费组IDenable-auto-commit: true        # 是否自动提交offset auto-commit-interval: 2000      # 提交offset延时(接收到消息后多久提交offset)max-poll-records: 500           # 单次拉取消息的最大条数,根据业务需要配置auto-offset-reset: latestlistener:missing-topics-fatal: false # 当kafka启动的时候如果未找到对应topic不报错properties:session:timeout:ms: 120000 # 消费会话超时时间（超过这个时间 consumer 没有发送心跳，就会触发 rebalance 操作）request:timeout:ms: 18000  # 消费请求的超时时间

读取第一个kafka配置 

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;/*** 第一个kafka配置**/
@Configuration
public class FirstKafkaConfig {/*** 读取第一个kafka配置* Primary注解表示默认以这个为准** @return 第一个kafka配置*/@Primary@ConfigurationProperties(prefix = "spring.kafka.first")@Beanpublic KafkaProperties firstKafkaProperties() {return new KafkaProperties();}/*** 构建第一个kafka的生产者发送template** @param firstKafkaProperties 第一个kafka配置* @return 第一个kafka的生产者发送template*/@Primary@Beanpublic KafkaTemplate<String, String> firstKafkaTemplate(@Autowired @Qualifier("firstKafkaProperties") KafkaProperties firstKafkaProperties) {return new KafkaTemplate<>(firstProducerFactory(firstKafkaProperties));}/*** 新建第一个kafka的生产者工厂** @param firstKafkaProperties 第一个kafka配置* @return 第一个kafka的生产者工厂*/private DefaultKafkaProducerFactory<String, String> firstProducerFactory(KafkaProperties firstKafkaProperties) {return new DefaultKafkaProducerFactory<>(firstKafkaProperties.buildProducerProperties());}/*** 构建第一个kafka的消费者监听容器工厂** @param firstKafkaProperties 第一个kafka配置* @return 第一个kafka的消费者监听容器工厂*/@Bean("firstKafkaListenerContainerFactory")public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>firstKafkaListenerContainerFactory(@Autowired @Qualifier("firstKafkaProperties") KafkaProperties firstKafkaProperties) {ConcurrentKafkaListenerContainerFactory<Integer, String> factory =new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(firstConsumerFactory(firstKafkaProperties));return factory;}/*** 新建第一个kafka的消费者工厂** @param firstKafkaProperties 第一个kafka配置* @return 第一个kafka的消费者工厂*/private ConsumerFactory<? super Integer, ? super String> firstConsumerFactory(KafkaProperties firstKafkaProperties) {return new DefaultKafkaConsumerFactory<>(firstKafkaProperties.buildConsumerProperties());}}

读取第二个kafka配置

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;/*** 第二个kafka配置**/
@Configuration
public class SecondKafkaConfig {/*** 读取第二个kafka配置** @return 第二个kafka配置*/@ConfigurationProperties(prefix = "spring.kafka.second")@Bean("secondKafkaProperties")public KafkaProperties secondKafkaProperties() {return new KafkaProperties();}/*** 构建第二个kafka的生产者发送template** @param secondKafkaProperties 第二个kafka配置* @return 第二个kafka的生产者发送template*/@Beanpublic KafkaTemplate<String, String> secondKafkaTemplate(@Autowired @Qualifier("secondKafkaProperties") KafkaProperties secondKafkaProperties) {return new KafkaTemplate<>(secondProducerFactory(secondKafkaProperties));}/*** 新建第二个kafka的生产者工厂** @param secondKafkaProperties 第二个kafka配置* @return 第二个kafka的生产者工厂*/private DefaultKafkaProducerFactory<String, String> secondProducerFactory(KafkaProperties secondKafkaProperties) {return new DefaultKafkaProducerFactory<>(secondKafkaProperties.buildProducerProperties());}/*** 构建第二个kafka的消费者监听容器工厂** @param secondKafkaProperties 第二个kafka配置* @return 第二个kafka的消费者监听容器工厂*/@Bean("secondKafkaListenerContainerFactory")public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>secondKafkaListenerContainerFactory(@Autowired @Qualifier("secondKafkaProperties") KafkaProperties secondKafkaProperties) {ConcurrentKafkaListenerContainerFactory<Integer, String> factory =new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(secondConsumerFactory(secondKafkaProperties));return factory;}/*** 新建第二个kafka的消费者工厂** @param secondKafkaProperties 第二个kafka配置* @return 第二个kafka的消费者工厂*/private ConsumerFactory<? super Integer, ? super String> secondConsumerFactory(KafkaProperties secondKafkaProperties) {return new DefaultKafkaConsumerFactory<>(secondKafkaProperties.buildConsumerProperties());}
}

创建两个生产者

@RestController
@RequestMapping("/producer1")
@Api(tags = "kafka生产者测试1")
public class ProducerDemoController1 {@Resourceprivate KafkaTemplate<String, String> kafkaTemplate;/*** 发送消息** @param msg* @return*/@PostMapping("/sendMsg")public String sendMsg(@RequestParam(name = "msg", defaultValue = "hello kafka") String msg) {kafkaTemplate.send("YWZT-TOPIC-TEST", msg);return "消息已发送";}
}
@RestController
@RequestMapping("/producer2")
@Api(tags = "kafka生产者测试2")
public class ProducerDemoController2 {/*** 第二个kafka配置需要指定下名字 */@Resource(name = "secondKafkaTemplate")private KafkaTemplate<String, String> secondKafkaTemplate;/*** 发送消息** @param msg* @return*/@PostMapping("/sendMsg")public String sendMsg(@RequestParam(name = "msg", defaultValue = "hello kafka") String msg) {secondKafkaTemplate.send("YWZT-TOPIC-TEST", msg);return "消息已发送";}
}

创建两个消费者

@Slf4j
@Component
@RequiredArgsConstructor
public class YwztExConsumer1 {@KafkaListener(containerFactory = "firstKafkaListenerContainerFactory",topics = {"YWZT-TOPIC-TEST"},groupId = "testGroup")public void consumer(String value) throws JsonProcessingException {//获取报文log.info("------------------------报文信息----------------：{} ......", StrUtil.sub(value, 0, 400));}}
@Slf4j
@Component
@RequiredArgsConstructor
public class YwztExConsumer2 {@KafkaListener(containerFactory = "secondKafkaListenerContainerFactory",topics = {"YWZT-TOPIC-TEST"},groupId = "testGroup")public void consumer(String value) throws JsonProcessingException {//获取报文log.info("------------------------报文信息----------------：{} ......", StrUtil.sub(value, 0, 400));}}

http://www.mrgr.cn/news/14108.html

相关文章：

  • 神经网络——CIFAR10小实战
  • RabbitMQ 常见问题与故障排查
  • 奇门WMS-A和金蝶云星空单据接口对接
  • 程序设计训练3.16最大报销额
  • 主机休眠之后 Ubuntu 虚拟机无网络
  • 【有道云-注册安全分析报告】
  • 小程序面试题一
  • 彩色相机拍照，图片时亮时暗
  • 悬浮翻译软件有哪些？试试这些利器
  • 独立站除了Shopify还有什么？
  • 数据库：头歌实验二数据库安全性技术
  • 【网络安全】服务基础第一阶段——第六节：Windows系统管理基础---- DNS部署与安全
  • 从算法到硬件实现：《基于FPGA的数字信号处理》（可下载）
  • 【IoT】将各类遥控器（红外，频射，蓝牙，wifi，Zigbee）等设备接入米家，实现家庭物联网设备控制（以极米Z7X投影仪为例）
  • 骁龙CPU简介
  • gitlab使用
  • 国内外主流的车辆管理系统盘点，你用过哪一款？
  • php邮件系统怎么安装和集成到现有系统中？
  • 【java计算机毕设】游泳馆管理系统小程序MySQL ssm vue uniapp maven项目设计源代码带文档PPT 小组作业
  • python从入门到精通：文件操作