前言
需要對Kafka有一定的了解
先拋出幾個問題:
spring boot kafka使用
- 如何配置多線程消費征堪？
- 如何手動管理Offset咱士？
- 如何命名多線程名稱循帐？
1. quick start
先快速開始體驗一下
springBootVersion=1.5.17.RELEASE
jdk=1.8.0_161
// spring-kafka版本 2.2.0.RELEASE
compile("org.springframework.kafka:spring-kafka:2.2.0.RELEASE")
配置一下KafkaTemplate和KafkaConsumer
package com.dobest.bfas.collection.consumer;
import com.dobest.analytics.utils.IPUtil;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import java.util.HashMap;
import java.util.Map;
/**
* @author: hujiansong
*/
@Configuration
public class KafkaTemplateConfig {
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
@Bean
public ProducerFactory<String, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(producerConfigs());
}
@Bean
public Map<String, Object> producerConfigs() {
Map<String, Object> props = new HashMap<>(16);
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.RETRIES_CONFIG, "3");
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "gzip");
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 200);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
return props;
}
@Bean
ConcurrentKafkaListenerContainerFactory<String, String>
kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setConcurrency(3);
// factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
return factory;
}
@Bean
public ConsumerFactory<String, String> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> props = new HashMap<>(16);
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "test.consumer");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
props.put(ConsumerConfig.CLIENT_ID_CONFIG, IPUtil.getHostName() + "-h5-consumer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
return props;
}
}
這樣就配置好了给赞，開袋即食用:
@Component
@Slf4j
public class SinkMessageListener implements CommandLineRunner {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
// 指定topics，可以指定多個蓬坡，最好每個topic對應(yīng)一個
@KafkaListener(id = "test-consumer", topics = {"realtime.analytics_h5_source"})
public void listen(ConsumerRecord<String, String> cr) throws Exception {
log.debug(Thread.currentThread().getName() + " {}", cr.value());
}
@Override
public void run(String... args) throws Exception {
log.info("start producer 10000 message ");
for (int i = 0; i < 100; i++) {
kafkaTemplate.send("realtime.analytics_h5_source", UUID.randomUUID().toString(), "i" + i);
}
}
}
回答問題
- 如何配置多線程消費猿棉？
可以看到消費工廠磅叛，可以指定
factory.setConcurrency(3)
一行代碼解決多線程消費問題
- 如何手動管理Offset？
首先將
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
將自動提交設(shè)置為false
然后消費工廠設(shè)置提交模式
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
AckMode 如下:
RECORD :當listener一讀到消息萨赁，就提交offset
BATCH : poll() 函數(shù)讀取到的所有消息,就提交offset
TIME : 當超過設(shè)置的ackTime 弊琴，即提交Offset
COUNT ：當超過設(shè)置的COUNT，即提交Offset
COUNT_TIME ：TIME和COUNT兩個條件都滿足杖爽，提交offset
MANUAL ： Acknowledgment.acknowledge()即提交Offset敲董，和Batch類似
MANUAL_IMMEDIATE： Acknowledgment.acknowledge()被調(diào)用即提交Offset
count 和 time都可以通過factory的配置進行設(shè)置
factory.getContainerProperties().setAckCount();
factory.getContainerProperties().setAckTime();
設(shè)置完之后，就可以在Listen中回調(diào)
@KafkaListener(id = "test-consumer", topics = {"realtime.analytics_h5_source"})
public void listen(ConsumerRecord<String, String> cr,Acknowledgment acknowledgment) throws Exception {
log.debug(Thread.currentThread().getName() + " {}", cr.value());
// 注意: 比 quickstart 讀了一個Acknowledgment參數(shù)
acknowledgment.acknowledge();
}
如果遇到Exception
java.lang.IllegalStateException: Consumer cannot be configured for auto commit for ackMode MANUAL_IMMEDIATE
修改配置慰安，將自動提交關(guān)閉
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
這樣就可以手動提交offset了
- 如何命名多線程?
大家都知道腋寨，見名知意的重要性，所以Spring-kafka可以對并發(fā)消費修改線程名字
@KafkaListener( topics = {"realtime.analytics_h5_source"})
public void listen(ConsumerRecord<String, String> cr) throws Exception {
// 打印當前線程
log.debug(Thread.currentThread().getName() + " {}", cr.value());
}
KafkaListener不加id得到的線程名字
link
KafkaListener 加id得到的線程名字:
@KafkaListener(id = "test-consumer", topics = {"realtime.analytics_h5_source"})
public void listen(ConsumerRecord<String, String> cr) throws Exception {
log.debug(Thread.currentThread().getName() + " {}", cr.value());
}
link
總結(jié)
使用KafkaTemplate生產(chǎn)者步驟:
- 配置producerConfigs
- 配置producerFactory
- 配置KafkaTemplate
使用KafkaLinstener消費者步驟
- 配置consumerConfigs
- 配置consumerFactory
- 配置kafkaListenerContainerFactory
其中kafkaListenerContainerFactory可以設(shè)置并發(fā)數(shù)化焕，offset提交模式精置。
