SpringBoot集成kafka
集成kafka
pom引入
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.8.11</version>
</dependency>
配置kafka
spring:
kafka:
bootstrap-servers: 127.0.0.1:9092
producer:
# 发生错误后，消息重发的次数。
retries: 0
#当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。
batch-size: 16384
# 设置生产者内存缓冲区的大小。
buffer-memory: 33554432
# 键的序列化方式
key-serializer: org.apache.kafka.common.serialization.StringSerializer
# 值的序列化方式
value-serializer: org.apache.kafka.common.serialization.StringSerializer
# acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。
# acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
# acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。
acks: 1
consumer:
# 自动提交的时间间隔 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5D
auto-commit-interval: 1S
# 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
# latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
# earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录
auto-offset-reset: latest
# 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量
enable-auto-commit: false
# 键的反序列化方式
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
# 值的反序列化方式
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
# 设置默认消费者分组id
group-id: dev
listener:
# 在侦听器容器中运行的线程数。
concurrency: 5
#listner负责ack，每调用一次，就立即commit
ack-mode: manual_immediate
missing-topics-fatal: false
kafka生产者示例
@Component
@Slf4j
public class KafkaProducer {
@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;
// 发送kafka消息
public void send(String topic, Object data) {
if (StringUtils.isEmpty(topic)) {
log.error("Kafka topic不能为空");
throw new IllegalArgumentException("Kafka topic不能为空");
} else if (ObjectUtil.isEmpty(data)) {
log.error("Kafka消息不能为空");
throw new IllegalArgumentException("Kafka消息不能为空");
} else {
// 将数据转换为JSON字符串
String jsonStr = JSONUtil.toJsonStr(data);
// 向kafka发送消息
ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, jsonStr);
future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
@Override
public void onFailure(Throwable throwable) {
//发送失败的处理
log.info(topic + " - 生产者 发送消息失败：" + throwable.getMessage());
}
@Override
public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
//成功的处理
log.info(topic + " - 生产者 发送消息成功：" + stringObjectSendResult.toString());
}
});
}
}
}
kafka消费者示例
@Component
@Slf4j
public class KafkaConsumer {
@KafkaListener(topics = "testTopic")
public void topicMessage(String message){
log.info("Received Message in topic: {}", message);
// 处理接收到的消息
}
@KafkaListener(topics = "testTopic2")
public void topicMessage2(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic){
try {
Optional message = Optional.ofNullable(record.value());
if (message.isPresent()) {
log.info("Received Message in topic: {}", message.get());
}
// 处理接收到的消息
} catch (Exception e) {
log.error("Error processing message", e);
} finally {
ack.acknowledge();
}
}
}
kafka配置批量并发消费
配置批量消费
spring:
kafka:
consumer:
# 一次 poll 最多返回的记录数
max-poll-records: 3100
max-partition-fetch-bytes: 15728640 #设置拉取数据的大小,15M
listener:
type: batch # 开启批量消费
配置并发消费
为了加快消费，我们可以提高并发数，比如下面配置我们将并发设置为 3。注意：并发量根据实际分区数决定，必须小于等于分区数，否则会有线程一直处于空闲状态
spring:
kafka:
consumer:
# 一次 poll 最多返回的记录数
max-poll-records: 3100
max-partition-fetch-bytes: 15728640 #设置拉取数据的大小,15M
listener:
type: batch # 开启批量消费
concurrency: 3 # 设置并发数
批量消费示例
@KafkaListener(topics = "testTopic")
public void batchListener(List<ConsumerRecord<?,?>> records, Acknowledgment ack){
try {
records.forEach(record -> {
// TODO 业务代码
});
} catch (Exception e) {
log.error("Kafka监听异常"+e.getMessage(),e);
} finally {
ack.acknowledge();//手动提交偏移量
}
}
除了通过配置文件配置外，还可以通过自定义配置类的方式实现批量消费，但是相对yml配置来说还是有点麻烦的：
/**
* 消费者配置
*/
@Configuration
public class KafkaConsumerConfig {
/**
* 消费者配置
* @return
*/
public Map<String,Object> consumerConfigs(){
Map<String,Object> props = new HashMap<>();
props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-group");
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 5);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
return props;
}
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, Object>> batchFactory() {
ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
//并发数量
factory.setConcurrency(3);
//开启批量监听
factory.setBatchListener(true);
return factory;
}
}
同时监听器通过@KafkaListener注解的containerFactory 配置指定批量消费的工厂即可，如下：
@KafkaListener(topics = {"testTopic"},containerFactory = "batchFactory")
public void batchListener(List<ConsumerRecord<?,?>> records, Acknowledgment ack){
try {
records.forEach(record -> {
// TODO 业务代码
});
} catch (Exception e) {
log.error("Kafka监听异常"+e.getMessage(),e);
} finally {
ack.acknowledge();//手动提交偏移量
}
}