Spring纯Java配置集成kafka代码实例
这篇文章主要介绍了Spring纯Java配置集成kafka代码实例,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
KafkaConfig.java
packagecom.niugang.config;
importjava.util.HashMap;
importjava.util.Map;
importorg.apache.kafka.clients.consumer.ConsumerConfig;
importorg.apache.kafka.clients.consumer.OffsetAndMetadata;
importorg.apache.kafka.clients.consumer.OffsetCommitCallback;
importorg.apache.kafka.clients.producer.ProducerConfig;
importorg.apache.kafka.common.TopicPartition;
importorg.apache.kafka.common.serialization.StringDeserializer;
importorg.apache.kafka.common.serialization.StringSerializer;
importorg.slf4j.Logger;
importorg.slf4j.LoggerFactory;
importorg.springframework.context.annotation.Bean;
importorg.springframework.context.annotation.Configuration;
importorg.springframework.kafka.annotation.EnableKafka;
importorg.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
importorg.springframework.kafka.core.ConsumerFactory;
importorg.springframework.kafka.core.DefaultKafkaConsumerFactory;
importorg.springframework.kafka.core.DefaultKafkaProducerFactory;
importorg.springframework.kafka.core.KafkaTemplate;
importorg.springframework.kafka.core.ProducerFactory;
importorg.springframework.kafka.listener.AbstractMessageListenerContainer;
importcom.niugang.controller.SenderConttoller;
/**
*
*@ClassName:KafkaConfig
*@Description:kafka配置类,基于springjava纯配置的
*@author:niugang
*@date:2018年10月20日下午8:04:26
*@Copyright:863263957@qq.com.Allrightsreserved.
*
*/
@Configuration
@EnableKafka
publicclassKafkaConfig{
privateLoggerlogger=LoggerFactory.getLogger(KafkaConfig.class);
@Bean
publicConcurrentKafkaListenerContainerFactorykafkaListenerContainerFactory(){
ConcurrentKafkaListenerContainerFactoryfactory=newConcurrentKafkaListenerContainerFactory<>();
//偏移量提交方式
//factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.COUNT);
//异步提交偏移量(默认就是true)
//factory.getContainerProperties().setSyncCommits(true);
//回调函数经常用于记录提交错误
/*factory.getContainerProperties().setCommitCallback(newOffsetCommitCallback(){
@Override
publicvoidonComplete(Mapoffsets,Exceptionexception){
if(exception!=null){
logger.error("Commitfailedforeffsets{}",offsets,exception);
}
}
});*/
factory.setConsumerFactory(consumerFactory());
returnfactory;
}
/**
*消费者工厂配置
*
*@return
*/
@Bean
publicConsumerFactoryconsumerFactory(){
returnnewDefaultKafkaConsumerFactory<>(consumerProps());
}
/**
*生产者工厂配置
*
*@return
*/
@Bean
publicProducerFactoryproducerFactory(){
returnnewDefaultKafkaProducerFactory<>(senderProps());
}
/**
*kafka发送消息模板
*
*@return
*/
@Bean
publicKafkaTemplatekafkaTemplate(){
returnnewKafkaTemplate(producerFactory());
}
/**
*消费者监听
*
*@return
*/
@Bean
publicConsumerListenerlistener(){
returnnewConsumerListener();
}
/**
*消费配置方法
*
*@return
*/
privateMapconsumerProps(){
Mapprops=newHashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG,"kafka_group_1");
/**
*enable.auto.commit默认5秒自动提交偏移量
*/
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"100");
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,"15000");
/**
*kafka是基于key-value键值对的，以下配置key和value的反序列化放
*/
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);
returnprops;
}
/**
*生产者配置方法
*
*生产者有三个必选属性
*
*1.bootstrap.serversbroker地址清单，清单不要包含所有的broker地址，
*生产者会从给定的broker里查找到其他broker的信息。不过建议至少提供两个broker信息，一旦其中一个宕机，生产者仍能能够连接到集群上。
*
*
*2.key.serializerbroker希望接收到的消息的键和值都是字节数组。生产者用对应的类把键对象序列化成字节数组。
*
*
*3.value.serializer值得序列化方式
*
*
*
*@return
*/
privateMapsenderProps(){
Mapprops=newHashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
/**
*当从broker接收到的是临时可恢复的异常时，生产者会向broker重发消息，但是不能无限
*制重发，如果重发次数达到限制值，生产者将不会重试并返回错误。
*通过retries属性设置。默认情况下生产者会在重试后等待100ms，可以通过retries.backoff.ms属性进行修改
*/
props.put(ProducerConfig.RETRIES_CONFIG,0);
/**
*在考虑完成请求之前，生产者要求leader收到的确认数量。这可以控制发送记录的持久性。允许以下设置：
*
*-
*
acks=0如果设置为零，则生产者将不会等待来自服务器的任何确认。该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且
*retries配置将不会生效（因为客户端通常不会知道任何故障）。为每条记录返回的偏移量始终设置为-1。
* acks=1
*这意味着leader会将记录写入其本地日志，但无需等待所有follower的完全确认即可做出回应。在这种情况下，
*如果leader在确认记录后立即失败但在关注者复制之前，则记录将丢失。
*acks=all
*这意味着leader将等待完整的同步副本集以确认记录。这保证了只要至少一个同步副本仍然存活，记录就不会丢失。这是最强有力的保证。
*这相当于acks=-1设置
*/
props.put(ProducerConfig.ACKS_CONFIG,"1");
/**
*当有多条消息要被发送到统一分区是，生产者会把他们放到统一批里。kafka通过批次的概念来提高吞吐量，但是也会在增加延迟。
*/
//以下配置当缓存数量达到16kb，就会触发网络请求，发送消息
props.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);
//每条消息在缓存中的最长时间，如果超过这个时间就会忽略batch.size的限制，由客户端立即将消息发送出去
props.put(ProducerConfig.LINGER_MS_CONFIG,1);
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);
//key的序列化方式
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class);
//value序列化方式
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class);
returnprops;
}
}
ConsumerListener.java
packagecom.niugang.config;
importorg.springframework.kafka.annotation.KafkaListener;
/**
*
*@ClassName:ConsumerListener
*@Description:消费者监听
*@author:niugang
*@date:2018年10月21日下午2:05:21
*@Copyright:863263957@qq.com.Allrightsreserved.
*
*/
publicclassConsumerListener{
/**
*topicPattern:支持正则表达式
*@paramfoo
*/
@KafkaListener(id="foo",topics="annotated1")
publicvoidlisten1(Stringfoo){
System.out.println("接收消息为："+foo);
}
}
源码：https://gitee.com/niugangxy/kafka/tree/master/kafka-spring-boot
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。
声明：本文内容来源于网络，版权归原作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎发送邮件至：czq8825#qq.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。