一、搭建kafka集群
参考文档:http://kafka.apache.org/quickstart 官方文档讲的很详细,而且没坑,照着做很快就可以搭好
注意点 or 建议:
1、在Linux下,启动的kafka集群经常无故退出,看日志也没有报错,就是启动了关闭流程,正常关闭。
解决方案:用守护进程启动,参考:https://blog.csdn.net/xiaoyu_bd/article/details/52268659。
bin/kafka-server-start.sh -daemon config/server.properties > k0.log &
2、kafka各项配置以及默认值说明:http://kafka.apache.org/documentation/#configuration 还是官方文档
3、kafka是Java进程,因此可以用 jps 命令方便的查看对应的端口。
4、以下配置为 监听地址,默认是 localhost,这样的话无法远程连接,需要配置为 特定的IP地址,然后用配置的IP来连接。
listeners=PLAINTEXT://172.17.10.89:9092
如上配置,本地连接也需要 使用 172.17.10.89:9092这个地址,而不是 localhost。
二、kafka介绍
参考文档:http://kafka.apache.org/intro 官方文档
https://www.jianshu.com/p/d3e963ff8b70 网友的中文版本,很详细,但要注意有些配置在新版本中发生了变化,比如: auto.create.topics.auto 变成了 auto.create.topics.enable 并且默认值是 true topic不存在时,按照默认配置创建topic
https://www.jianshu.com/p/4e00dff97f39 关闭 offset自动提交,让 spring-kafka 来提交offset
https://blog.csdn.net/lishuangzhe7047/article/details/74530417 kafka auto.offset.reset值详解,offset缺失情况下的消费者消 费策略
网上文档很多,不要犹豫该看哪些,而是都看下,对照着看。
三、Spring Boot 集成 kafka
pom spring-kafka 的版本要稍微注意下 我用的 kafka_2.11-2.0.0 配合 spring-kafka-1.1.3.RELEASE 可以正常使用,但是用 spring-kafka-2.xxx的时候 项目无法启动,会报错:not found
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>1.1.3.RELEASE</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.0</version>
<scope>provided</scope>
</dependency>
yml配置:
spring:
kafka:
bootstrap-servers: 172.17.10.89:9092,172.17.10.89:9093,172.17.10.89:9094
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
batch-size: 65536
buffer-memory: 524288
consumer:
group-id: default-group #默认组id 后面会配置多个消费者组
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
auto-offset-reset: latest
enable-auto-commit: false #关闭自动提交 改由spring-kafka提交
auto-commit-interval: 100
max-poll-records: 20 #批量消费 一次接收的最大数量
bootstrap-servers 这边配置的地址 可以事先看下 是否可以访问到对应的端口
配置读取类:
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
/**
* @author fandong
* @create 2018/11/1
*/
@Configuration
@Data
public class KafkaConsumerProps {
@Value("${spring.kafka.bootstrap-servers}")
private String bootstrapServers;
@Value("${spring.kafka.consumer.group-id}")
private String defaultGroupId;
@Value("${spring.kafka.consumer.auto-offset-reset}")
private String autoOffsetReset;
@Value("${spring.kafka.consumer.enable-auto-commit}")
private String enableAutoCommit;
@Value("${spring.kafka.consumer.auto-commit-interval}")
private String autoCommitInterval;
@Value("${spring.kafka.consumer.max-poll-records}")
private String maxPollRecords;
public KafkaConsumerProps() {
}
}
消费者配置类,配置多个消费者组、批量消费、并发数
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
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.listener.ConcurrentMessageListenerContainer;
import java.util.HashMap;
import java.util.Map;
/**
* @author fandong
* @create 2018/11/1
*/
@Configuration
public class KafkaConsumerConfig {
@Autowired
private KafkaConsumerProps kafkaConsumerProps;
private static final String GROUP0_ID = "group0";
private static final String GROUP1_ID = "group1";
@Bean
KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory0() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory0());
//对应topic分区数
factory.setConcurrency(3);
//设置为批量消费,每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
factory.setBatchListener(true);
factory.getContainerProperties().setPollTimeout(3000);
return factory;
}
public ConsumerFactory<String, String> consumerFactory0() {
Map<String, Object> map = consumerConfigs();
map.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP0_ID);
return new DefaultKafkaConsumerFactory<>(map);
}
@Bean
KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory1() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory1());
//对应topic分区数
factory.setConcurrency(3);
//设置为批量消费,每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
factory.setBatchListener(true);
factory.getContainerProperties().setPollTimeout(3000);
return factory;
}
public ConsumerFactory<String, String> consumerFactory1() {
Map<String, Object> map = consumerConfigs();
map.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP1_ID);
return new DefaultKafkaConsumerFactory<>(map);
}
public Map<String, Object> consumerConfigs() {
Map<String, Object> propsMap = new HashMap<>(16);
propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConsumerProps.getBootstrapServers());
propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaConsumerProps.getEnableAutoCommit());
propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, kafkaConsumerProps.getAutoCommitInterval());
propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConsumerProps.getDefaultGroupId());
propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaConsumerProps.getAutoOffsetReset());
propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaConsumerProps.getMaxPollRecords());
return propsMap;
}
}
注意:
1、factory.setConcurrency(3); 此处设置的目的在于:假设 topic test 下有 0、1、2三个 partition,Spring Boot中只有一个 @KafkaListener() 消费者订阅此 topic,此处设置并发为3,启动后 会有三个不同的消费者分别订阅 p0、p1、p2,本地实际有三个消费者线程。而 factory.setConcurrency(1); 的话 本地只有一个消费者线程, p0、p1、p2被同一个消费者订阅。由于 一个partition只能被同一个消费者组下的一个消费者订阅,对于只有一个 partition的topic,即使设置 并发为3,也只会有一个消费者,多余的消费者没有 partition可以订阅。
2、factory.setBatchListener(true); 设置批量消费 ,每个批次数量在Kafka配置参数ConsumerConfig.MAX_POLL_RECORDS_CONFIG中配置,限制的是 一次批量接收的最大条数,而不是 等到达到最大条数才接收,这点容易被误解。实际测试时,接收是实时的,当生产者大量写入时,一次批量接收的消息数量为 配置的最大条数。
生产者我们借助自动配置,在yml文件中加入生产者配置之后,直接注入 KafkaTemplate 即可使用。
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import java.util.List;
/**
* @author fandong
* @create 2018/11/1
*/
@Service
public class KafkaServiceImpl implements KafkaService {
private final KafkaTemplate<String, String> kafkaTemplate;
private final Logger logger = LoggerFactory.getLogger(this.getClass().getName());
@Autowired
public KafkaServiceImpl(KafkaTemplate kafkaTemplate) {
this.kafkaTemplate = (KafkaTemplate<String, String>) kafkaTemplate;
}
@Override
public void send(String topic, String value) {
ListenableFuture<SendResult<String, String>> resultListenableFuture = kafkaTemplate.send(topic, value);
resultListenableFuture.addCallback(
successCallback -> logger.info("发送成功:topic= " + topic + " value= " + value),
failureCallback -> logger.info("发送失败:topic= " + topic + " value= " + value));
}
@Override
@KafkaListener(topics = {"test"}, containerFactory = "kafkaListenerContainerFactory0")
public void kafkaConsumerTest(String msg) {
logger.info("接收到消息--" + msg);
}
@Override
@KafkaListener(topics = {"3-test"}, containerFactory = "kafkaListenerContainerFactory0")
public void listenPartition0(List<ConsumerRecord<String, String>> records) {
System.out.println(records.size());
for (ConsumerRecord<String, String> consumerRecord : records){
String value = consumerRecord.value();
logger.info("a 消息:partition " + consumerRecord.partition() + " value " + consumerRecord.value() );
}
}
@Override
@KafkaListener(topics = {"3-test"}, containerFactory = "kafkaListenerContainerFactory1")
public void listenPartition2(List<ConsumerRecord<String, String>> records) {
System.out.println(records.size());
for (ConsumerRecord<String, String> consumerRecord : records){
String value = consumerRecord.value();
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
logger.info("c 消息:partition " + consumerRecord.partition() + " value " + consumerRecord.value() + " thread id " + Thread.currentThread().getName());
}
}
}
指定 containerFactory = "kafkaListenerContainerFactory1" 参数给消费者分组,值为 之前定义的 KafkaListenerContainerFactory的 Bean 名称,不指定的情况下 默认是 方法名称。
注:topic 3-test下有 3个partition,由于之前配置了 factory.setConcurrency(3); 项目启动之后,本地会有三个消费者线程。
使用如下命令可以查看各个消费者组的情况 以下为查看 group0消费组,可以看到每个partition由不同的消费者订阅。
bin/kafka-consumer-groups.sh --bootstrap-server 172.17.10.89:9092 --describe --group group0
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
test 0 12 12 0 consumer-1-9419e037-1501-4f33-85c9-25df75e3a5a9 /172.17.10.33 consumer-1
3-test 1 1400 1400 0 consumer-8-1a520c8e-c412-4178-a76f-77c68e7472b7 /172.17.10.33 consumer-8
3-test 0 1398 1398 0 consumer-7-ede99b63-ea10-48dd-be62-61139360e39c /172.17.10.33 consumer-7
3-test 2 1397 1397 0 consumer-9-4bacc023-7de7-4e98-aefd-bb39d0bf6547 /172.17.10.33 consumer-9
四、关于提高消费者消费能力的思考
kafka写具有很好的性能,而消费者在消费时往往会有相对耗时的操作,所以经常出现 消费者性能跟不上的情况。
思路:
1、在topic下适当创建多个 partition,然后使用多个消费者来消费多个partition
2、使用批量消费,一次接收 多条消息,相比一个一个接收,(猜)可以减少IO次数,提高速度
3、消费者再使用 线程池配合适当长度的阻塞队列,进一步提高处理能力(需要分析任务类型以及考虑处理器的能力)。