SpringBoot项目集成kafka及常规配置_springboot集成kafka配置_spring boot kafka常用配置

### 1.kafka相关配置


#### 1.1 KafkaConfiguration 公共配置



1
2
3
4
5
6
7
8

@Data
@Configuration
public class KafkaConfiguration {

/**
 * 主机地址
 */
@Value("${kafka.server-host}")
private String bootstrapServers;

/**
 * sasl 认证账号
 */
@Value("${iot.kafka.sasl.username:admin}")
private String userName;

/**
 * sasl 密码
 */
@Value("${iot.kafka.sasl.password:iot@2021}")
private String password;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

}

#### 1.2 KafkaConsumerConfiguration 消费者配置



1
2
3
4
5

@Data
@Configuration
public class KafkaConsumerConfiguration {

/**
 * 默认组id
 */
@Value("${iot.kafka.consumer.properties.group-id:default-group}")
private String groupId;

@Value("${iot.kafka.consumer.properties.fetch-max-wait:5000}")
private Integer fetchMaxWait;

/**
 * 此设置限制每次调用poll返回的消息数，这样可以更容易的预测每次poll间隔要处理的最大值。通过调整此值，可以减少poll间隔，减少重新平衡分组的
 */
@Value("${iot.kafka.consumer.properties.max-poll-record:100}")
private Integer maxPollRecordsConfig;

/**
 * 增大poll的间隔，可以为消费者提供更多的时间去处理返回的消息（调用poll（long）返回的消息，通常返回的消息都是一批），缺点是此值越大将会延迟组重新平衡。
 */
@Value("${iot.kafka.consumer.properties.max-poll-interval-ms:100}")
private Integer maxPollIntervalConfig;

/**
 * 是否开启自动提交
 */
@Value("${iot.kafka.consumer.properties.enable-auto-commit:#{false}}")
private boolean enableAutoCommitConfig;

/**
 * 消费策略
 * earliest  当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
 * latest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
 * none topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
 */
@Value("${iot.kafka.consumer.properties.auto-offset-reset:earliest}")
private String autoOffsetResetConfig;

@Value("${iot.kafka.consumer.properties.auto-commit-interval:1000}")
private String autoCommitIntervalMsConfig;

@Value("${iot.kafka.consumer.properties.session-timeout:30000}")
private String sessionTimeoutMsConfig;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

}

#### 1.3 KafkaListenerConfiguration 监听配置



1
2
3
4
5

@Data
@Configuration
public class KafkaListenerConfiguration {

/**
 * 启用线程数（提高并发）
 */
@Value("${iot.kafka.listener.concurrency:3}")
private Integer concurrency;

/**
 * 手动提交的方式，当enable-auto-commit: false时起作用
 * manual:手动调用Acknowledgment.acknowledge()后立即提交
 * record:当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
 * batch:当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
 * time: 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
 * count:当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
 * count_time:当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
 */
@Value("${iot.kafka.listener.ack-mode:manual_immediate}")
private String ackMode;

/**
 * 消费超时时间
 */
@Value("${iot.kafka.listener.poll-timeout:3000}")
private Long pollTimeout;

/**
 * 是否开启批量处理
 */
@Value("${iot.kafka.listener.batch_listener:#{true}}")
private Boolean batchListener;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

}

#### 1.4 KafkaProducerConfiguration 生产者配置



1
2
3
4
5

@Data
@Configuration
public class KafkaProducerConfiguration {
/**
* 重试次数 默认值0
*/
@Value(“${iot.kafka.producer.retries:0}”)
private Integer retries;

/**
 * acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为 - 1。
 * acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
 * acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1 的设置。
 */
@Value("${iot.kafka.producer.acks:all}")
private String acks;

/**
 * 指定缓存的大小，生产者缓存每个分区未发送的消息。默认 16384
 */
@Value("${iot.kafka.producer.batch-size:16384}")
private Integer batchSize;

/**
 * 生产者发送请求之前等待一段时间，设置等待时间是希望更多地消息填补到未满的批中。 默认 30
 */
@Value("${iot.kafka.producer.properties.linger.ms:30}")
private Integer lingerMs;

/**
 * 通过KafkaProducer发送出去的消息都是先进入到客户端本地的内存缓冲里，然后把很多消息收集成一个一个的Batch，再发送到Broker上去的 默认32m
 */
@Value("${iot.kafka.producer.buffer-memory:33554432}")
private Integer bufferMemory;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

}

### 2.工厂配置


#### 2.1 ConsumerFactoryBuilder 消费者工厂



1
2
3
4
5
6
7
8

@Configuration
public class ConsumerFactoryBuilder {

@Autowired
private KafkaConfiguration kafkaConfiguration;

@Autowired
private KafkaConsumerConfiguration kafkaConsumerConfiguration;

@Autowired
private KafkaListenerConfiguration kafkaListenerConfiguration;

/**
 * 消费者配置
 *
 * @return properties
 */
@Bean
public Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new ConcurrentHashMap<>();
    //配置地址
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfiguration.getBootstrapServers());
    //消费者组 默认组id
    props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConsumerConfiguration.getGroupId());
    //是否开启自动提交
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaConsumerConfiguration.isEnableAutoCommitConfig());
    /* 消费策略
     * earliest  当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
     * latest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
     * none topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
     */
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaConsumerConfiguration.getAutoOffsetResetConfig());
    //消费者默认等待服务响应时间(毫秒)
    props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, kafkaConsumerConfiguration.getFetchMaxWait());
    props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, kafkaConsumerConfiguration.getAutoCommitIntervalMsConfig());
    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, kafkaConsumerConfiguration.getSessionTimeoutMsConfig());
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaConsumerConfiguration.getMaxPollRecordsConfig());
    props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaConsumerConfiguration.getMaxPollIntervalConfig());
    //key序列化器选择
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    //value序列化器选择
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    //设置sasl认证
    props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
    props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
    props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username='" + kafkaConfiguration.getUserName() + "' password='" + kafkaConfiguration.getPassword() + "';");
    return props;
}

/**
 * kafka消费者工厂
 */
@Bean
public ConsumerFactory<Object, Object> consumerFactory() {
    return new DefaultKafkaConsumerFactory(consumerConfigs());
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。

深知大多数大数据工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《2024年大数据全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友。

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上大数据开发知识点，真正体系化！

由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新

如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）

一个人可以走的很快，但一群人才能走的更远。不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎扫码加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！

只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新**

如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）
[外链图片转存中…(img-UKftCNQ1-1712990397116)]

一个人可以走的很快，但一群人才能走的更远。不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎扫码加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！