依赖管理
在pom.xml文件中导入依赖
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.5.5.RELEASE</version>
</dependency>配置文件修改
需要自己配置AckMode时候的配置
spring:
application:
name: base.kafka
kafka:
bootstrap-servers: kafka服务地址1:端口,kafka服务地址2:端口,kafka服务地址3:端口
producer:
# 写入失败时，重试次数。当leader节点失效，一个repli节点会替代成为leader节点，此时可能出现写入失败，
# 当retris为0时，produce不会重复。retirs重发，此时repli节点完全成为leader节点，不会产生消息丢失。
retries: 0
#procedure要求leader在考虑完成请求之前收到的确认数，用于控制发送记录在服务端的持久化，其值可以为如下：
#acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为-1。
#acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
#acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1的设置。
#可以设置的值为：all, -1, 0, 1
acks: 1
consumer:
group-id: testGroup
# smallest和largest才有效，如果smallest重新0开始读取，如果是largest从logfile的offset读取。一般情况下我们都是设置smallest
auto-offset-reset: earliest
# 设置自动提交offset
enable-auto-commit: true
max-poll-records: 2
server:
port: 8060
消费kafka消息
kafka支持的消费模式，设置在AbstractMessageListenerContainer.AckMode的枚举中，下面就介绍下各个模式的区别
/**
* The offset commit behavior enumeration.
*/
public enum AckMode {
/**
* Commit after each record is processed by the listener.
*/
RECORD,
/**
* Commit whatever has already been processed before the next poll.
*/
BATCH,
/**
* Commit pending updates after
* {@link ContainerProperties#setAckTime(long) ackTime} has elapsed.
*/
TIME,
/**
* Commit pending updates after
* {@link ContainerProperties#setAckCount(int) ackCount} has been
* exceeded.
*/
COUNT,
/**
* Commit pending updates after
* {@link ContainerProperties#setAckCount(int) ackCount} has been
* exceeded or after {@link ContainerProperties#setAckTime(long)
* ackTime} has elapsed.
*/
COUNT_TIME,
/**
* User takes responsibility for acks using an
* {@link AcknowledgingMessageListener}.
*/
MANUAL,
/**
* User takes responsibility for acks using an
* {@link AcknowledgingMessageListener}. The consumer
* immediately processes the commit.
*/
MANUAL_IMMEDIATE,
}
AckMode模式
| AckMode模式 | 作用 |
|---|---|
| MANUAL | 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交 |
| MANUAL_IMMEDIATE | 手动调用Acknowledgment.acknowledge()后立即提交 |
| RECORD | 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交 |
| BATCH | 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交 |
| TIME | 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交 |
| COUNT | 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交 |
| COUNT_TIME | TIME或COUNT 有一个条件满足时提交 |
监听器工厂的配置类：
/**
* kafka消费者配置
*/
@Configuration
@EnableKafka
public class KafkaConsumerConfig {
@Value("${spring.kafka.bootstrap-servers}")
private String servers;
//会话过期时长,consumer通过ConsumerCoordinator间歇性发送心跳
//超期后,会被认为consumer失效,服务迁移到其他consumer节点.(group)
//需要注意,Coordinator与kafkaConsumer共享底层通道,也是基于poll获取协调事件,但是会在单独的线程中
@Value("${spring.kafka.consumer.session.timeout}")
private String sessionTimeout;
@Value("${spring.kafka.consumer.concurrency}")
private int concurrency;
//单次最多允许poll的消息条数.
//此值不建议过大,应该考虑你的业务处理效率.
@Value("${spring.kafka.consumer.maxpoll.records}")
private int maxPollRecords;
//两次poll之间的时间隔间最大值,如果超过此值将会被认为此consumer失效,触发consumer重新平衡.
//此值必须大于,一个batch所有消息处理时间总和.
//最大500000
//2分钟
@Value("${spring.kafka.consumer.maxpoll.interval}")
private int maxPollIntervalMS;
@Value("${spring.kafka.consumer.group-id}")
private String groupId;
@Bean
public StringJsonMessageConverter converter() {
return new StringJsonMessageConverter();
}
@Bean
public KafkaListenerContainerFactory<?> batchDataFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
Map<String, Object> consumerConfig = consumerConfigs();
consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
ConsumerFactory<String, String> consumerFactory = new DefaultKafkaConsumerFactory<>(consumerConfig);
factory.setConsumerFactory(consumerFactory);
factory.setConcurrency(concurrency);
//设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
factory.setBatchListener(true);
factory.setMessageConverter(new BatchMessagingMessageConverter());
//设置提交偏移量的方式
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
factory.getContainerProperties().setPollTimeout(3000);
return factory;
}
public Map<String, Object> consumerConfigs() {
Map<String, Object> propsMap = new HashMap<>();
propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//每一批数量
propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, this.maxPollRecords);
propsMap.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG,this.maxPollIntervalMS);
return propsMap;
}
@Bean
public TestMessages listener() {
return new TestMessages();
}
}监听器使用的配置
@Component
public class TestMessages {
/**
* MANUAL 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
* @param message
* @param ack
*/
@KafkaListener(containerFactory = "batchDataFactory" , topics = "kafka(topic名称)")
public void onMessageManual(List<Object> message, Acknowledgment ack){
log.info("batchDataFactory处理数据量：{}",message.size());
message.forEach(item -> log.info("batchDataFactory处理数据内容：{}",item));
ack.acknowledge();//直接提交offset
}
}
MANUAL_IMMEDIATE
当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
MANUAL和MANUAL_IMMEDIATE两者的相同和区别
相同之处
这两种模式都是需要进行手动确认ack.acknowledge();才能完成消息的消费，否则在重启消费端实例的时候数据会再次被消费端接收到。
两者的区别
MANUAL: 在处理完最后一次轮询的所有结果后，将队列排队，并在一次操作中提交偏移量。可以认为是在批处理结束时提交偏移量
MANUAL_IMMEDIATE：只要在侦听器线程上执行确认，就立即提交偏移。会在批量执行的时候逐一提交它们。
其他模式大家都可以在批量处理工厂类中进行修改设置：文章来源：https://www.toymoban.com/news/detail-564495.html
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);上面的内容可能存在没有描述清楚或者错误的地方，假如开发同学发现了，请及时告知，我会第一时间修改相关内容。假如我的这篇内容对你有任何帮助的话，麻烦给我点一个赞。你的点赞就是我前进的动力。文章来源地址https://www.toymoban.com/news/detail-564495.html
到了这里，关于Spring Boot 整合kafka消费模式AckMode以及手动消费 依赖管理的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
