本文代码格式不好调整，可以参考本人在其他地方的同篇博文 https://blog.csdn.net/russle/article/details/80296006
Apache Kafka, 分布式消息系统，
非常流行。Spring是非常流行的Java快速开发框架。将两者无缝平滑结合起来可以快速实现很多功能。本文件简要介绍Spring
Kafka，如何使用 KafkaTemplate发送消息到kafka的broker上， 如何使用“listener
container“接收Kafka消息。
1，Spring Kafka的组成
这一节我们首先介绍Spring Kafka的各个组成部分。
1.1 发送消息
与 JmsTemplate 或者JdbcTemplate类似，Spring Kafka提供了 KafkaTemplate. 该模板封装了Kafka消息生产者并提供各种消息发送方法。
消息发送的各种方法。
```
ListenableFuture> send(Stringtopic, V data);ListenableFuture> send(Stringtopic,Kkey, V data);ListenableFuture> send(Stringtopic, int partition, V data);ListenableFuture> send(Stringtopic, int partition,Kkey, V data);ListenableFuture> send(Message message);
```
1.2 接收消息
要接收消息，我们需要配置MessageListenerContainer并提供一个Message Listener，或者使用 @KafkaListener注解。
MessageListenserContainer
MessageListenserContainer 有以下两个实现类：
KafkaMessageListenerContainer
ConcurrentMessageListenerContainer
KafkaMessageListenerContainer可以让我们使用单线程消费Kafka topic的消息，而ConcurrentMessageListenerContainer 可以让我们多线程消费消息。
@KafkaListener 注解
Spring Kafka提供的 @KafkaListener注解，可以让我们监听某个topic或者topicPattern的消息。
监听符合topicPattern = “topic.*”的所有topic的消息
```
@Component@Slf4jpublicclassCmdReceiver { @KafkaListener(topicPattern ="topic.*")publicvoidlisten(ConsumerRecord record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) { Optional kafkaMessage = Optional.ofNullable(record.value());if(kafkaMessage.isPresent()) { Object message = kafkaMessage.get(); log.info("----------------- record =topic："+ topic+", "+ record); log.info("------------------ message =topic："+ topic+", "+ message); } }}
```
监听某个topic的消息publicclassListener { @KafkaListener(id ="id01", topics ="Topic1")publicvoidlisten(String data) { }}
2， Spring Kafka 例子
下面我们介绍一个具体的例子， 这个例会发送和接收指定topic的消息。
准备工作：
kafka_2.11-1.1.0.tgz和zookeeper-3.4.10.tar.gz
JDK jdk-8u171-linux-x64.tar.gz
IDE (Eclipse or IntelliJ)
Build tool (Maven or Gradle)
本文不涉及安装Kafka的介绍，请自行搜索，或者看官方文档。
pom文件：
也就是我们的依赖包. 这是笔者使用的依赖版本，仅供参考。
4.0.0com.yqkafkademo1.0-SNAPSHOTorg.springframework.bootspring-boot-starter-parent1.5.12.RELEASEUTF-8UTF-81.8org.springframework.bootspring-boot-starter-weborg.projectlomboklomboktrueorg.springframework.bootspring-boot-starter-testtestorg.springframework.kafkaspring-kafka1.1.8.RELEASEcom.google.code.gsongson2.8.2org.apache.kafkakafka-clients0.10.1.1io.springfoxspringfox-swagger22.7.0io.springfoxspringfox-swagger-ui2.7.0io.springfoxspringfox-spring-web2.7.0com.alibabafastjson1.1.33org.springframework.bootspring-boot-maven-plugin
*KafkaDemoApplication*
我们使用springboot的框架，这是我们程序的入口点。
@SpringBootApplicationpublicclassKafkaDemoApplication{privatestaticfinalLogger logger = LoggerFactory.getLogger(KafkaDemoApplication.class);publicstaticvoidmain(String[] args) { ConfigurableApplicationContext context = SpringApplication.run(KafkaDemoApplication.class, args); logger.info("Done start Spring boot"); }}
ProducerConfig
其实我们可以可以不用编写KafkaProducerConfig，直接使用KafkaTemplate（当然前提是我们要设置好producer需要的配置项，例如spring.kafka.bootstrap-servers, spring.kafka.producer.key-serializer, spring.kafka.producer.retries等等）
@Configuration@EnableKafkapublicclassKafkaProducerConfig{@BeanpublicProducerFactoryproducerFactory() {returnnewDefaultKafkaProducerFactory<>(producerConfigs()); }@BeanpublicMapproducerConfigs() { Map props =newHashMap<>(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092(根据实际情况修改)"); props.put(ProducerConfig.RETRIES_CONFIG,0); props.put(ProducerConfig.BATCH_SIZE_CONFIG,16384); props.put(ProducerConfig.LINGER_MS_CONFIG,1); props.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);returnprops; }@BeanpublicKafkaTemplatekafkaTemplate() {returnnewKafkaTemplate(producerFactory()); }}
KafkaConsumerConfig
同理，其实我们可以可以不用编写KafkaConsumerConfig，直接使用 @KafkaListener（当然前提是我们要设置好consumer需要的配置项，例如spring.kafka.bootstrap-servers, spring.kafka.consumer.key-deserializer, spring.kafka.consumer.group-id、spring.kafka.consumer.auto-offset-reset等等）
@Configuration@EnableKafkapublicclassKafkaConsumerConfig{@BeanKafkaListenerContainerFactory> kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactory factory =newConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); factory.setConcurrency(3); factory.getContainerProperties().setPollTimeout(3000);returnfactory; }@BeanpublicConsumerFactoryconsumerFactory() {returnnewDefaultKafkaConsumerFactory<>(consumerConfigs()); }@BeanpublicMapconsumerConfigs() { Map propsMap =newHashMap<>(); propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092(根据实际情况修改)"); propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false); propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"100"); propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,"15000"); propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); propsMap.put(ConsumerConfig.GROUP_ID_CONFIG,"group1"); propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");returnpropsMap; }@BeanpublicMyListenerlistener() {returnnewMyListener(); }}
定义了ProducerConfig和ConsumerConfig后我们需要实现具体的生产者和消费者。
本文的KafkaListenerContainerFactory 中使用了ConcurrentKafkaListenerContainer， 我们将使用多线程消费消息。
注意消息代理的地址是localhost:9092，
需要根据实际情况修改。需要特别注意的是，我在windows运行程序，kafka在我的linux虚拟机，
我需要配置windows的hosts文件，配置虚拟机hostname和ip的映射，例如192.168.119.131
ubuntu01
开发Listener
我们来开发自己的Listener监听具体的topic， 这里例子中我们监听以topic开头的主题，不做其他业务，只是打印出来。
@Component@Slf4jpublicclassMyListener{ @KafkaListener(topicPattern ="topic.*")publicvoidlisten(ConsumerRecord record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) { Optional kafkaMessage = Optional.ofNullable(record.value());if(kafkaMessage.isPresent()) { Object message = kafkaMessage.get(); log.info("------------------ message =topic："+ topic+", "+ message); } }}
开发producer
我在程序中增加了controller，这样我们可以通过controller给topic发消息。consumer一直在监听，只要有消息发送过去，就会打印出来。controller中调用了ProducerServiceImpl ， 具体代码比较简单就不再罗列。
我们producerServiceImpl主要是有这句， 通过KafkaTemplate 发送消息。
@Autowired
private KafkaTemplate template;
@ServicepublicclassProducerServiceImplimplementsProducerService{privatestaticfinalLogger logger = LoggerFactory.getLogger(ProducerServiceImpl.class);privateGson gson =newGsonBuilder().create();@AutowiredprivateKafkaTemplate template;//发送消息方法publicvoidsendJson(String topic, String json) { JSONObject jsonObj = JSON.parseObject(json); jsonObj.put("topic", topic); jsonObj.put("ts", System.currentTimeMillis() +""); logger.info("json+++++++++++++++++++++ message = {}", jsonObj.toJSONString()); ListenableFuture> future = template.send(topic, jsonObj.toJSONString()); future.addCallback(newListenableFutureCallback>() {@OverridepublicvoidonSuccess(SendResult result) { System.out.println("msg OK."+ result.toString()); }@OverridepublicvoidonFailure(Throwable ex) { System.out.println("msg send failed: "); } }); }
运行程序
运行第一步，确保Kafka broker配置正确，笔者的程序在Windows10机器上，Kafka在虚拟机上，因为我的地址是192.168.119.129:9092, 而不是localhost:9092.
运行第二步骤，在IDEA中选中KafkaDemoApplication , 单击鼠标右键，选择 Run KafkaDemoApplication
效果图
kafka段命令行接收到的消息
3，总结
Spring Kafka提供了很好的集成，我们只需配置properties文件，就可以直接使用KafkaTemplate发送消息，使用@KafkaListener监听消息。
参考文档：
https://docs.spring.io/spring-kafka/docs/2.1.6.RELEASE/reference/html/_reference.html#kafka-template
