开发者

springboot之配置双kafka全过程

开发者 https://www.devze.com 2023-04-15 10:43 出处：网络 作者： 窦再兴
目录springboot配置双kafka引入Maven kafka jar、准备两个kafka；配置yml配置文件配置KafkaConfig类发送工具类MyKafkaProducer测试类接收类测试结果总结springboot配置双kafka
相关专题：springboot kafkaspringboot配置kafkaspringboot配置双kafka
目录
  • springboot配置双kafka
  • 引入Maven kafka jar、准备两个kafka；
  • 配置yml配置文件
  • 配置KafkaConfig类
  • 发送工具类MyKafkaProducer
  • 测试类
  • 接收类
  • 测试结果
  • 总结

springboot配置双kafka

使用spring boot 2.0.8.RELEASE 版本

引入Maven kafka jar、准备两个kafka；

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

配置yml配置文件

spring:
 kafka:
  bootstrap-servers: 180.167.180.242:9092 #kafka的访问地址，多个用","隔开
  consumer:
   enable-auto-commit: true
   group-id: kafka #群组ID
 outkafka:
  bootstrap-servers: localhost:9092 #kafka的访问地址，多个用","隔开
  consumer:
   enable-auto-commit: true
   group-id: kafka_1 #群组ID

配置KafkaConfig类

import Java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.annotation.EnableKafka;
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.core.ProducerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

@Configuration
@EnableKafka
public class KafkaConfig {
  @Value(编程"${spring.kafka.bootstrap-servers}")
  private String innerServers;
  @Value("${spring.kafka.consumer.group-id}")
  private String innerGroupid;
  @Value("${spring.kafka.consumer.enable-auto-commit}")
  private String innerEnableAutoCommit;

  @Bean
  @Primary//理解为默认优先选择当前容器下的消费者工厂
  KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setConcurrency(3);
    factory.getContainerProperties().setPollTimeout(3000);
    return factory;
  }

  @Bean//第一个消费者工厂的bean
  public ConsumerFactory<Integer, String> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(consumerConfigs());
  }

  @Bean
  public Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, innerServers);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, innerGroupid);
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, innerEnableAutoCommit);
//    props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
//    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    return props;
  }
 
  @Bean //生产者工厂配置
  public ProducerFactory<String, String> producerFactory() {
    return new DefaultKafkaProducerFactory<>(senderProps());
  }
 
  @Bean //kafka发送消息模板
  public KafkaTemplate<String, String> kafkaTemplate() {
    return new KafkaTemplate<String, String>(producerFactory());
  }
 
  /**
  * 生产者配置方法
  *
  * 生产者有三个必选属性
  * <p>
  * 1.bootstrap.servers broker地址清单，清单不要包含所有的broker地址，
  * 生产者会从给定的broker里查找到其他broker的信息。不过建议至少提供两个broker信息，一旦 其中一个宕机，生产者仍能能够连接到集群上。
  * </p>
  * <p>
  * 2.key.serializer broker希望接收到的消息的键和值都是字节数组。 生产者用对应的类把键对象序列化成字节数组。
  * </p>
  * <p>
  * 3.value.serializer 值得序列化方式
  * </p>
  *
  *
  * @return
  */
  private Map<String, Object> senderProps() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 开发者_Python教程innerServers);
    /**
    * 当从broker接收到的是临时可恢复的异常时，生产者会向broker重发消息，但是不能无限
    * 制重发，如果重发次数达到限制值，生产者将不会重试并返回错误。
    * 通过retries属性设置。默认情况下生产者会在重试后等待100ms，可以通过 retries.backoff.ms属性进行修改
    */
    props.put(ProducerConfig.RETRIES_CONFIG, 0);
    /**
    * 在考虑完成请求之前，生产者要求leader收到的确认数量。这可以控制发送记录的持久性。允许以下设置：
    * <ul>
    * <li>
    * <code> acks = 0 </ code>如果设置为零，则生产者将不会等待来自服务器的任何确认。该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且
    * <code>retries </ code>配置将不会生效（因为客户端通常不会知道任何故障）。为每条记录返回的偏移量始终设置为-1。
    * <li> <code> acks = 1 </code>
    * 这意味着leader会将记录写入其本地日志，但无需等待所有follower的完全确认即可做出回应。在这种情况下，
    * 如果leader在确认记录后立即失败但在关注者复制之前，则记录将丢失。
    * <li><code> acks = all </code>
    * 这意味着leader将等待完整的同步副本集以确认记录。这保证了只要至少一个同步副本仍然存活，记录就不会丢失。这是最强有力的保证。
    * 这相当于acks = -1设置
    */
    props.put(ProducerConfig.ACKS_CONFIG, "1");
    /**
    * 当有多条消息要被发送到统一分区是，生产者会把他们放到统一批里。kafka通过批次的概念来 提高吞吐量，但是也会在增加延迟。
    */
    // 以下配置当缓存数量达到16kb，就会触发网络请求，发送消息
//    props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
    // 每条消息在缓存中的最长时间，如果超过这个时间就会忽略batch.size的限制，由客户端立即将消息发送出去
//    props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
//    props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    return props;
  }
 
  @Value("${spring.outkafka.bootstrap-servers}")
  private String outServers;
  @Value("${spring.outkafka.consumer.group-id}")
  private String outGroupid;
  @Value("${spring.outkafka.consumer.enable-auto-commit}")
  private String outEnableAutoCommit;
 

  static {
   
  }
 
  /**
  * 连接第二个kafka集群的配置
  */
  @Bean
  KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaListenerContainerFactoryOutSchedule() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFphpactory(consumerFactoryOutSchedule());
    factory.setConcurrency(3);
    factory.getContainerProperties().setPollTimeout(3000);
    return factory;
  }

  @Bean
  public ConsumerFactory<Integer, String> consumerFactoryOutSchedule() {
    return new DefaultKafkaConsumerFactory<>(consumerConfigsOutSchedule());
  }

  /**
  * 连接第二个集群的消费者配置
  */
  @Bean
  public Map<St编程客栈ring, Object> consumerConfigsOutSchedule() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, outServers);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, outGroupid);
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, outEnableAutoCommit);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    return props;
  }
 
  @Bean //生产者工厂配置
  public ProducerFactory<String, String> producerOutFactory() {
    return new DefaultKafkaProducerFactory<>(senderOutProps());
  }
 
  @Bean //kafka发送消息模板
  public KafkaTemplate<String, String> kafkaOutTemplate() {
    return new KafkaTemplate<String, String>(producerOutFactory());
  }
 
  /**
  * 生产者配置方法
  *
  * 生产者有三个必选属性
  * <p>
  * 1.bootstrap.servers broker地址清单，清单不要包含所有的broker地址，
  * 生产者会从给定的broker里查找到其他broker的信息。不过建议至少提供两个broker信息，一旦 其中一个宕机，生产者仍能能够连接到集群上。
  * </p>
  * <p>
  * 2.key.serializer broker希望接收到的消息的键和值都是字节数组。 生产者用对应的类把键对象序列化成字节数组。
  * </p>
  * <p>
  * 3.value.serializer 值得序列化方式
  * </p>
  *
  *
  * @return
  */
  private Map<String, Object> senderOutProps() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, outServers);
    /**
    * 当从broker接收到的是临时可恢复的异常时，生产者会向broker重发消息，但是不能无限
    * 制重发，如果重发次数达到限制值，生产者将不会重试并返回错误。
    * 通过retries属性设置。默认情况下生产者会在重试后等待100ms，可以通过 retries.backoff.ms属性进行修改
    */
    props.put(ProducerConfig.RETRIES_CONFIG, 0);
    /**
    * 在考虑完成请求之前，生产者要求leader收到的确认数量。这可以控制发送记录的持久性。允许以下设置：
    * <ul>
    * <li>
    * <code> acks = 0 </ code>如果设置为零，则生产者将不会等待来自服务器的任何确认。该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且
    * <code>retries </ code>配置将不会生效（因为客户端通常不会知道任何故障）。为每条记录返回的偏移量始终设置为-1。
    * <li> <code> acks = 1 </code>
    * 这意味着leader会将记录写入其本地日志，但无需等待所有follower的完全确认即可做出回应。在这种情况下，
    * 如果leader在确认记录后立即失败但在关注者复制之前，则记录将丢失。
    * <li><code> acks = all </code>
    * 这意味着leader将等待完整的同步副本集以确认记录。这保证了只要至少一个同步副本仍然存活，记录就不会丢失。这是最强有力的保证。
    * 这相当于acks = -1设置
    */
    props.put(ProducerConfig.ACKS_CONFIG, "1");
    /**
    * 当有多条消息要被发送到统一分区是，生产者会把他们放到统一批里。kafka通过批次的概念来 提高吞吐量，但是也会在增加延迟。
    */
    // 以下配置当缓存数量达到16kb，就会触发网络请求，发送消息
//    props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
    // 每条消息在缓存中的最长时间，如果超过这个时间就会忽略batch.size的限制，由客户端立即将消息发送出去
//    props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
//    props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    return props;
  }
}

发送工具类MyKafkaProducer

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import lombok.extern.slf4j.Slf4j;

/**
* <p>
* KafkaProducer Description: kafka生产者
* </p>
*
* @author douzaixingDATE 2019年7月8日 下午4:09:29
*/
@Component // 这个必须加入容器不然，不会执行
@EnableScheduling // 这里是为了测试加入定时调度
@Slf4j
public class MyKafkaProducer {

  @Autowired
  private KafkaTemplate<String, String> kafkaTemplate;

  @Autowired
  private KafkaTemplate<String, String> kafkaOutTemplate;

  public ListenableFuture<SendResult<String, String>> send(String topic, String key, String json) {
    ListenableFuture<SendResult<String, String>> result = kafkaTemplate.send(topic, key, json);
    log.info("inner kafka send #topic=" + topic + "#key=" + key + "#json=" + jso编程n + "#推送成功===========");
    return result;
  }

  public ListenableFuture<SendResult<String, String>> sendOut(String topic, String key, String json) {
    ListenableFuture<SendResult<String, String>> result = kafkaOutTemplate.send(topic, key, json);
    log.info("out kafka send #topic=" + topic + "#key=" + key + "#json=" + json + "#推送成功===========");
    return result;
  }

}

测试类

@Slf4j
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes={OesBcServiceApplication.class})
public class MoreKafkaTest {
 
  @Autowired
  private MyKafkaProducer kafkaProducer;
 
  @Test
  public void sendInner() {
    for (int i = 0; i < 1; i++) {
      kafkaProducer.send("inner_test", "douzi" + i, "liyuehua" + i);
      kafkaProducer.sendOut("out_test", "douziout" + i, "fanbingbing" + i);
    }
  }
}

接收类

@Component
@Slf4j
public class KafkaConsumer { 
  @KafkaListener(topics={"inner_test"}, containerFactory="kafkaListenerContainerFactory")
  public void innerlistener(ConsumerRecord<String, String> record) {
    log.info("inner kafka receive #key=" + record.key() + "#value=" + record.value());
  }
 
  @KafkaListener(topics={"out_test"}, containerFactory="kafkaListenerContainerFactoryOutSchedule")
  public void outListener(ConsumerRecord<String, String> record) {
    log.info("out kafka receive #key=" + record.key() + "#value=" + record.value());
  }
}

测试结果

07-11 12:41:27.811 INFO  [com.wondertek.oes.bc.service.send.MyKafkaProducer] - inner kafka send #topic=inner_test#key=douzi0#json=liyuehua0#推送成功===========

 

07-11 12:41:27.995 INFO  [com.wondertek.oes.bc.service.send.KafkaConsumer] - inner kafka receive #key=douzi0#value=liyuehua0

07-11 12:41:28.005 INFO  [com.wondertek.oes.bc.service.send.MyKaf编程kaProducer] - out kafka send #topic=out_test#key=douziout0#json=fanbingbing0#推送成功===========

07-11 12:41:28.013 INFO  [com.wondertek.oes.bc.service.send.KafkaConsumer] - out kafka receive #key=douziout0#value=fanbingbing0

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持我们。

0

上一篇:

:下一篇

更多 开发 相关资讯：

精彩评论

暂无评论...
登录 注册
Qedev.com
验证码 换一张
取 消

关注公众号

热门标签

图文推荐

Spring中存储Bean的常见注解方式

Spring中存储Bean的常见注解方式
如何解决异步任务上下文丢失问题

如何解决异步任务上下文丢失问题
JustAuth-第三方Oauth2登录方式

JustAuth-第三方Oauth2登录方式
springboot引入远程nacos配置文件错误的解决方案

springboot引入远程nacos配置文件错误的解决方案
Python中跳出循环的两种方法

Python中跳出循环的两种方法
Python复制Excel中的行、列和单元格的操作代码

Python复制Excel中的行、列和单元格的操作代码
使用python提取PowerPoint中的音频和视频

使用python提取PowerPoint中的音频和视频
golang中字符串和数字转换方法

golang中字符串和数字转换方法

开发者开发者网给大家分享系统运维,大数据运维,云计算,编程开发技巧,路由交换,运维和开发相关的资讯及技术文章，同时StackOverflow中文社区，知识经验交流分享。

法律声明：本站内容均为网友上传，网站举办方负责审核和监督，如存在版权或非法内容，欢迎举报，我们将尽快予以删除。邮箱：devze@qq.com

Copyright © 2018-2020 开发者. All rights reserved. Powered by 开发者ICP备案号: 京ICP备10032868号-9