生產(chǎn)者
說明
KafkaTemplate封裝了一個生成器，并提供了方便的方法來發(fā)送數(shù)據(jù)到kafka主題。 提供了異步和同步方法，異步方法返回一個Future。
其構(gòu)造方法有：
ListenableFuture<SendResult<K, V>> sendDefault(V data);
ListenableFuture<SendResult<K, V>> sendDefault(K key, V data);
ListenableFuture<SendResult<K, V>> sendDefault(int partition, K key, V data);
ListenableFuture<SendResult<K, V>> send(String topic, V data);
ListenableFuture<SendResult<K, V>> send(String topic, K key, V data);
ListenableFuture<SendResult<K, V>> send(String topic, int partition, V data);
ListenableFuture<SendResult<K, V>> send(String topic, int partition, K key, V data);
ListenableFuture<SendResult<K, V>> send(Message<?> message);
前3個方法需要向Temple提供默認主題
配置
使用Producer配置類
@Configuration
@EnableKafka
public class ProducerConfig {
@Value("${kafka.producer.servers}")
private String servers;
@Value("${kafka.producer.retries}")
private int retries;
@Value("${kafka.producer.batch.size}")
private int batchSize;
@Value("${kafka.producer.linger}")
private int linger;
@Value("${kafka.producer.buffer.memory}")
private int bufferMemory;
public Map<String, Object> producerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(Config.BOOTSTRAP_SERVERS_CONFIG, servers);
props.put(Config.RETRIES_CONFIG, retries);
props.put(Config.BATCH_SIZE_CONFIG, batchSize);
props.put(Config.LINGER_MS_CONFIG, linger);
props.put(Config.BUFFER_MEMORY_CONFIG, bufferMemory);
props.put(Config.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(Config.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(Config.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return props;
}
public ProducerFactory<String, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(producerConfigs());
}
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<String, String>(producerFactory());
}
}
示例
@RestController
@RequestMapping("/kafka/producer")
public class ProducerController {
private static Logger logger = LoggerFactory.getLogger(ProducerController.class);
@Value("${topic.name}")
private String topicName;
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
@RequestMapping("/send")
public Object sendKafka(String message) {
try {
logger.info("send kafka message: {}", message);
kafkaTemplate.send(topicName, UUID.randomUUID().toString(), message);
return "success";
} catch (Exception e) {
logger.error("發(fā)送kafka失敗", e);
return "fail";
}
}
}
消費者
說明
可以通過配置MessageListenerContainer并提供MessageListener或通過使用@KafkaListener注釋來接收消息。
MessageListenerContainer有兩個實現(xiàn)：
- KafkaMessageListenerContainer：從單個線程上的所有主題/分區(qū)接收所有消息
- ConcurrentMessageListenerContainer：委托給1個或多個KafkaMessageListenerContainer以提供多線程消費。通過container.setConcurrency(3)，來設置多個線程
配置
使用Consumer配置類
@Configuration
@EnableKafka
public class ConsumerConfig {
@Value("${kafka.consumer.servers}")
private String servers;
@Value("${kafka.consumer.enable.auto.commit}")
private boolean enableAutoCommit;
@Value("${kafka.consumer.session.timeout}")
private String sessionTimeout;
@Value("${kafka.consumer.auto.commit.interval}")
private String autoCommitInterval;
@Value("${kafka.consumer.group.id}")
private String groupId;
@Value("${kafka.consumer.topic}")
private String topic;
@Value("${kafka.consumer.auto.offset.reset}")
private String autoOffsetReset;
@Value("${kafka.consumer.concurrency}")
private int concurrency;
/**
* KafkaMessageListenerContainer： 從單個線程上的所有主題/分區(qū)接收所有消息
@Bean(initMethod = "doStart")
public KafkaMessageListenerContainer<String, String> kafkaMessageListenerContainer() {
KafkaMessageListenerContainer<String, String> container = new KafkaMessageListenerContainer<>(consumerFactory(), containerProperties());
return container;
}
*/
/**
* ConcurrentMessageListenerContainer：
* 委托給1個或多個KafkaMessageListenerContainer以提供多線程消費。
* 通過container.setConcurrency(3)，來設置多個線程
*/
@Bean(initMethod = "doStart")
public ConcurrentMessageListenerContainer<String, String> concurrentMessageListenerContainer() {
ConcurrentMessageListenerContainer<String, String> container = new ConcurrentMessageListenerContainer<>(consumerFactory(), containerProperties());
container.setConcurrency(concurrency);
return container;
}
public ConsumerFactory<String, String> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
public ContainerProperties containerProperties() {
ContainerProperties containerProperties = new ContainerProperties(topic);
containerProperties.setMessageListener(messageListener());
return containerProperties;
}
public Map<String, Object> consumerConfigs() {
Map<String, Object> propsMap = new HashMap<>();
propsMap.put(Config.BOOTSTRAP_SERVERS_CONFIG, servers);
propsMap.put(Config.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
propsMap.put(Config.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);
propsMap.put(Config.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
propsMap.put(Config.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
propsMap.put(Config.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
propsMap.put(Config.GROUP_ID_CONFIG, groupId);
propsMap.put(Config.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
return propsMap;
}
public MessageListener<String, String> messageListener() {
return new CustomMessageListener();
}
}
消息接收
Java實現(xiàn)
直接使用kafka0.10 client去收發(fā)消息
@Test
public void receive(){
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, broker);
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
props.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
try{
consumer.subscribe(Arrays.asList(topic));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(10000);
records.forEach(record -> {
System.out.printf("client : %s , topic: %s , partition: %d , offset = %d, key = %s, value = %s%n", clientId, record.topic(),
record.partition(), record.offset(), record.key(), record.value());
});
}
}catch (Exception e){
e.printStackTrace();
}finally {
consumer.close();
}
}
使用MessageListener接口
繼承MessageListener接口
public class CustomMessageListener implements MessageListener<Integer, String> {
private static Logger logger = LoggerFactory.getLogger(CustomMessageListener.class);
@Override
public void onMessage(ConsumerRecord<Integer, String> data) {
logger.info("received key: {}, value: {}", data.key(), data.value());
}
//或包含消費者的onMessage方法，以手動提交ofset
}
使用@KafkaListener注解
@KafkaListener(id = "foo", topics = "myTopic")
public void listen(String data) {
...
}
@KafkaListener(id = "bar", topicPartitions =
{ @TopicPartition(topic = "topic1", partitions = { "0", "1" }),
@TopicPartition(topic = "topic2", partitions = "0",
partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
})
public void listen(ConsumerRecord<?, ?> record) {
...
}
總結(jié)
- 對于生產(chǎn)者來說，封裝KafkaProducer到KafkaTemplate相對簡單
- 對于消費者來說，由于spring是采用注解的形式去標注消息處理方法
- 先在KafkaListenerAnnotationBeanPostProcessor中掃描bean，然后注冊到KafkaListenerEndpointRegistrar
- 而KafkaListenerEndpointRegistrar在afterPropertiesSet的時候去創(chuàng)建MessageListenerContainer
- messageListener包含了原始endpoint攜帶的bean以及method轉(zhuǎn)換成的InvocableHandlerMethod
- ConcurrentMessageListenerContainer這個銜接上，根據(jù)配置的spring.kafka.listener.concurrency來生成多個并發(fā)的KafkaMessageListenerContainer實例
- 每個KafkaMessageListenerContainer都自己創(chuàng)建一個ListenerConsumer，然后自己創(chuàng)建一個獨立的kafka consumer，每個ListenerConsumer在線程池里頭運行，這樣來實現(xiàn)并發(fā)
- 每個ListenerConsumer里頭都有一個recordsToProcess隊列，從原始的kafka consumer poll出來的記錄會放到這個隊列里頭，
- 然后有一個ListenerInvoker線程循環(huán)超時等待從recordsToProcess取出記錄，然后調(diào)用messageListener的onMessage方法(即KafkaListener注解標準的方法)
項目源碼
https://github.com/scjqwe/spring-kafka-examples
