Spring for Apache Kafka实战
背景介绍
Kafka是一个分布式的、可分区的、可复制的消息系统，在现在的互联网公司，应用广泛，在我们公司在主要运用在定时推送业务，批量数据处理，日志上传等方面，我发现网上大部分博客，在使用上还只是对Apache 官方提供的client，进行运用开发，在这里推荐使用 Spring for Apache Kafka（简称spring-kafka） ，更新维护稳定，方法众多，并且强大，现已加入Spring豪华大礼包。
大坑预警
spring-kafka实际上也是对apache的kafka-client进行了包装和开发，所以使用的时候一定注意，你引入的spring-kafka里封装的kafka-client的版本要和服务器上的kafka服务器版本要对应，不然就会产生问题，比如消费失败。官网上在首页就贴出了SpringKafka和kafka-client版本（它的版本号要和kafka服务器的版本保持一致）的对应关系Spring for Apache Kafka
生产者的使用
spring-kafka对于生产者提供了两个模板类，分别是
- KafkaTemplate ：包装了一个生产者，并提供方便的方法将数据发送到kafka的topics。
- ReplyingKafkaTemplate ：KafkaTemplate的子类，增加了一个请求/应答功能，在发送数据后会返回一个future ，里面封装了消费者的应答信息。
这里主要介绍KafkaTemplate。对于使用，官方提供了这样一个案例
To use the template, configure a producer factory and provide it in the template’s constructor:
@Bean
public ProducerFactory<Integer, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(producerConfigs());
@Bean
public Map<String, Object> producerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
// See https://kafka.apache.org/documentation/#producerconfigs for more properties
return props;
}
@Bean
public KafkaTemplate<Integer, String> kafkaTemplate() {
return new KafkaTemplate<Integer, String>(producerFactory());
}官方的文档只是给出了，如何配置和通过工厂构造出一个模板。而我们需要的是一个完整的能在项目中使用的案例。在这里我给出一个与Spring boot集成，完整的生产者案例。
首先构造一个模板类
@Configuration
public class TestTemplateConfig{
public Map<String, Object> producerConfigs() {
Map<String, Object> props = new HashMap<>();
// kafka.metadata.broker.list=10.16.0.214:9092,10.16.0.215:9092,10.16.0.216:9092
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
props.put(ProducerConfig.RETRIES_CONFIG, 0);
props.put(ProducerConfig.ACKS_CONFIG,"-1");
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 4096);
props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 40960);
props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG,5048576);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return props;
}
/** 获取工厂 */
public ProducerFactory<String, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(producerConfigs());
}
/** 注册实例 */
@Bean(name="testTemplate")
public KafkaTemplate<String, String> testTemplate() {
return new KafkaTemplate<>(producerFactory());
}
}
运用@Configuration注解定义配置类，在内部通过spring-kafka提供的KafkaProducerFactory，来构造出KafkaTemplate，并在方法上通过@Bean注解将其注册，这样在初始化Spring容器时，就会将KafkaTemplate注入到容器中进行实例化。
发送消息
@Resource
private KafkaTemplate testTemplate;
//同步发送
public void syncSend(){
testTemplate.send("topic",result.toString()).get(10, TimeUnit.SECONDS);
}
//异步发送
public void asyncSend() {
ListenableFuture<SendResult<Integer, String>> future = testTemplate.send("topic",result.toString());
future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
@Override
public void onSuccess(SendResult<Integer, String> result) {
System.out.println("success");
}
@Override
public void onFailure(Throwable ex) {
System.out.println("failure");
}
});
}
消费者
对于消费者的使用，spring-kafka提供了，两种方式
Messages can be received by configuring a MessageListenerContainer and providing a Message Listener, or by using the @KafkaListener annotation.。 —— Spring for Apache Kafka
也就是通过配置MessageListenerContainer，可以提供消息侦听器或使用@KafkaListener注释来接收消息。
而这个官方的消息侦听器，其实就是接口，官方提供了8个接口来满足不同的应用场景，这里我们使用最常用的MessageListener接口
public class KafkaConsumerSerivceImpl implements MessageListener<String, String> {
@Override
public void onMessage(ConsumerRecord<String, String> data) {
//根据不同的主题进行消费
if("主题1".equals(data.topic())){
}else if("主题2".equals(data.topic())){
}
}
}
另一种通过@KafkaListener注解极为方便
@Component
public class KafkaConsumer {
@KafkaListener(topics = {"testTopic"})
public void receive(String message){
System.out.println("消费消息:" + message);
}
}
无论是哪种方法，都需要创建一个MessageListenerContainer，对于这个容器的构造官方给的案例比较详细。
@Configuration
@EnableKafka
public class KafkaConfig {
@Bean
KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setConcurrency(3);
factory.getContainerProperties().setPollTimeout(3000);
return factory;
}
@Bean
public ConsumerFactory<Integer, String> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
...
return props;
}
}
构造过程
Notice that to set container properties, you must use the getContainerProperties() method on the factory. It is used as a template for the actual properties injected into the container. —— Spring for Apache Kafka
说明，要设置监听容器的属性，比如轮询时间，只能是通过getContainerProperties().set()方法，来设置。
关键配置
生产者配置
bootstrap.servers：brokers集群地址
acks：
- -1 ：producer在所有follower副本确认接收到数据后才算一次发送完成
- 0 ：producer不等待来自broker同步完成的确认继续发送下一条
- 1 ：producer在leader已成功收到的数据并得到确认后发送下一条
retries：发送失败重试次数
batch.size：当多个记录被发送到同一个分区时，生产者会尝试将记录合并到更少的请求中
linger.ms：批处理延迟时间上限
buffer.memory：批处理缓冲区
max.request.size：发送数据最大大小
消费者配置
concurrency：消费监听器容器并发数
session.timeout.ms：检测消费者故障的超时
auto.offset.reset：
- earliest ：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
- latest ：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
- none ：topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
enable.auto.commit：
- true ： 自动提交位移
- false ：关闭自动提交位移, 在消息被完整处理之后再手动提交位移
auto.commit.interval.ms:若enable.auto.commit=true，这里设置自动提交周期
备注：
对数据可靠性较高的场景建议 offset 手动提交并将acks 设置为 "all" 即所有副本都同步到数据时send方法才返回, 以此来完全判断数据是否发送成功, 理论上来讲数据不会丢失。
