SpringBoot（八） SpringBoot整合Kafka

1、springboot（八） springboot整合kafkaspringboot（八） springboot整合kafka一、预备工作提前解释：假如你运行出问题，请检查kafka的版本与springboot的版本是否与我文中的全都，本文中的环境已经经过测试。kafka服务版本为 kafka_2.11-1.1.0 (scala)， 也就是1.1.0springboot版本：1.5.10.release提前启动zk，kafka，并且创建一个topicrootbasic kafka_2.11-1.1.0 bin/kafka-topics.sh -create -zookeeper localhos

2、t:2181 -replication-factor 1 -partitions 1 -topic test_topic确保你的kafka能够拜访，假如拜访不了，需要打开外网拜访。config/pertiesadvertised.listeners=plaintext:/28:9092maven 依靠org.springframework.kafkaspring-kafkaorg.apache.kafkakafka-clients二、项目结构为了越发体现实际开发需求，普通生产者都是在调用某些接口的服务处理完规律之后然后往kafka里

3、面扔数据，然后有一个消费者不停的监控这个topic，然后处理数据，所以这里把生产者作为一个接口，消费者放到kafka这个名目下，注重component注解，不然扫描不到kafkalistener 三、详细实现代码springboot配置文件application.ymlspring:kafka:bootstrap-servers: 28:9092producer:key-serializer: mon.serialization.stringserializervalue-serializer: mon.serialization.stringserializerco

4、nsumer:group-id: testenable-auto-commit: trueauto-commit-interval: 1000key-deserializer: mon.serialization.stringdeserializervalue-deserializer: mon.serialization.stringdeserializer生产者package cn.saytime.web;import org.springframework.beans.factory.annotation.autowired;import org.springframework.kafk

5、a.core.kafkatemplate;import org.springframework.web.bind.annotation.requestmapping;import org.springframework.web.bind.annotation.restcontroller;/* 测试kafka生产者*/restcontrollerrequestmapping("kafka")public class testkafkaproducercontroller autowiredprivate kafkatemplate kafkatemplate

6、;requestmapping("send")public string send(string msg)kafkatemplate.send("test_topic", msg);return "success"消费者这里的消费者会监听这个主题，有消息就会执行，不需要举行while(true)package cn.saytime.kafka;import org.apache.kafka.clients.consumer.consumerrecord;import org.spring

7、framework.kafka.annotation.kafkalistener;import org.springframework.stereotype.component;/* kafka消费者测试*/componentpublic class testconsumer kafkalistener(topics = "test_topic")public void listen (consumerrecord record) throws exception system.out.printf("topic = %s, offset

8、= %d, value = %s n", record.topic(), record.offset(), record.value();项目启动类package cn.saytime;import org.springframework.boot.springapplication;import org.springframework.boot.autoconfigure.springbootapplication;springbootapplicationpublic class testapplicationpublic static void main(string args) springapplication.run(testapplication.class, args);四、测试运行项目，执行：http:/localhost:8080/kafka/send?msg=hello控制台输出：topic = test_topic, offset = 19, value = hello1为了体现消费者不止执行一次就结束，再调用一