springboot集成kafka

系统管理员 2022-05-30 10:43 203阅读 0赞

我自己的一个小Demo，在自己的Windows上进行

一、安装zookeeper

先去zookeeper官网进行下载

http://zookeeper.apache.org/releases.html\#download

下载后解压到一个目录：

1.进入Zookeeper设置目录，笔者D:\\develop\\zookeeper\\zookeeper-3.4.9\\conf  
2. 将“zoo\_sample.cfg”重命名为“zoo.cfg”  
3. 在任意文本编辑器（如notepad）中打开zoo.cfg  
4. 找到并编辑dataDir=D:\\\\develop\\\\zookeeper\\\\tmp  
5. 与Java中的做法类似，我们在系统环境变量中添加：  
　　a. 在系统变量中添加ZOOKEEPER\_HOME = D:\\develop\\zookeeper\\zookeeper-3.4.9\\conf  
　　b. 编辑path系统变量，添加为路径%ZOOKEEPER\_HOME%\\bin;

6. 在zoo.cfg文件中修改默认的Zookeeper端口（默认端口2181）可以在启动zookeeper时看最后的启动日志

启动日志如下：

2018-03-10 18:25:08,183 [myid:] - INFO  [main:NIOServerCnxnFactory@89] - binding to port 0.0.0.0/0.0.
    0.0:2181

二、安装与运行Kafka

首先下载kafka

http://kafka.apache.org/downloads.html。注意要下载二进制版本的

![20180310183119833][]

下载后解压到任意一个目录，我这里的目录是D:\\develop\\kafka\\kafka\_2.12-1.0.1  
  
1. 进入Kafka配置目录，D:\\develop\\kafka\\kafka\_2.12-1.0.1  
2. 编辑文件“server.properties”  
3. 找到并编辑log.dirs=D:\\\\develop\\\\kafka\\\\kafka-log,这里的目录自己修改成自己喜欢的  
4. 找到并编辑zookeeper.connect=localhost:2181。表示本地运行  
5. Kafka会按照默认，在9092端口上运行，并连接zookeeper的默认端口：2181。  
  
运行：  
重要：请确保在启动Kafka服务器前，Zookeeper实例已经准备好并开始运行。  
1.进入Kafka安装目录D:\\develop\\kafka\\kafka\_2.12-1.0.1  
2.按下Shift+右键，选择“打开命令窗口”选项，打开命令行。

3.现在输入

.\bin\windows\kafka-server-start.bat .\config\server.properties

会报错： 找不到或无法加载主类

需要对kafka安装目录中找到bin\\windows目录中的kafka-run-class.bat的%CLASSPATH%加上双引号

如下：

set COMMAND=%JAVA% %KAFKA_HEAP_OPTS% %KAFKA_JVM_PERFORMANCE_OPTS% %KAFKA_JMX_OPTS% %KAFKA_LOG4J_OPTS% -cp "%CLASSPATH%" %KAFKA_OPTS% %*

再回到kafka根目录下运行启动命令：

.\bin\windows\kafka-server-start.bat .\config\server.properties

如果之前已经启动过了，需要重新启动的话，需要把之前的日志目录下所有文件全部清空后才能正常启动

kafka在windows平台就是有这个BUG，没办法。只能手动删除\\kafka-logs里的日志文件重启kafka

三、写代码Demo

1、配置文件配置

#kafka
    # 指定kafka 代理地址，可以多个
    spring.kafka.bootstrap-servers=localhost:9092
    # 指定默认消费者group id
    spring.kafka.consumer.group-id=myGroup
    # 指定默认topic id
    spring.kafka.template.default-topic= my-replicated-topic
    # 指定listener 容器中的线程数，用于提高并发量
    spring.kafka.listener.concurrency= 3
    # 每次批量发送消息的数量
    spring.kafka.producer.batch-size= 1000
    #key-value序列化反序列化
    #spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
    #spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
    #spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
    #spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
    spring.kafka.producer.buffer-memory=524288

pom.xml支持配置

<dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency

2、kafka生产者

/**
     * @Description: 生产者
     * @author WEISANGNG
     * @date 2018年3月10日
     */
    @Service
    public class KafkaProducer {
    	@Autowired
    	private KafkaTemplate kafkaTemplate;
    
    	/**
    	 * 发送消息到kafka,主题为test
    	 */
    	public void sendTest(){
            kafkaTemplate.send("test","hello,kafka  "  + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS")));
        }
    }

3、kafka消息者

@Service
    public class KafkaConsumer {
    	protected static Logger logger=LoggerFactory.getLogger(KafkaConsumer.class); 
    	/**
         * 监听test主题,有消息就读取
         * @param message
         */
        @KafkaListener(topics = {"test"})
        public void consumer(String message){
        	logger.info("test topic message : {}", message);
        }
    }

4、定时任务测试代码

@Service
    public class KafkaScheduled {
    	private static Logger logger = LoggerFactory.getLogger(KafkaScheduled.class);
    	@Autowired
        private KafkaProducer kafkaSender;
    	// 然后每隔1分钟执行一次
    	@Scheduled(fixedRate = 1000 * 20)
    	public void testKafka() throws Exception {
    		logger.info("KafkaScheduled...start");
    		kafkaSender.sendTest();
    	}
    }

5、启动类配置

/** 
     * @Description: 启动入口
     * @author WEISANGNG   
     * @date 2018年3月10日   
     */
    @SpringBootApplication
    @EnableScheduling
    public class KafkaApplication {
    
    	protected static Logger logger=LoggerFactory.getLogger(KafkaApplication.class); 
    
    	public static void main(String[] args) {
    		SpringApplication.run(KafkaApplication.class, args);
    		logger.info("----------------SpringBoot Start Success-------------------");
    	}
    }

6、结果

![20180310185649863][]

以上是只一个入门的Demo，如果有什么地方写得不对的地方，还望帮忙指出，在这里表示感谢！

[20180310183119833]: /images/20220530/2e7c5f5dd0024d3a811f2f297c9748de.png
[20180310185649863]: /images/20220530/4d77e56c2c1d43138279f123822639e3.png