SpringBoot整合Kafka

本文介绍: 先启动 zookeeper,在启动 kafka（这两个在bin下的windows 目录下），都需要指定配置文件（配置文件在config 目录下）。解压安装，记得放在D盘根目录下，要不然启动时会报命令行太长的错误。下载2的版本，3.的版本会报错。

Spr in gBo o t 整合Kaf k a

文章 目录

SpringBoot整合Kafka

下载与安装

下载地址：
https://kafka.apache.org/downloads
下载2的版本，3.的版本会报错
 解压安装，记得放在D盘根目录下，要不然启动时会报命令行太长的错误在这里插入图片描述

zookeeper-server-start.bat ../../config/zookeeper.properties

kafka-server-start.bat ../../config/server.properties

先启动 zookeeper,在启动k afka（这两个在bin下的windows 目录下），都需要指定配置文件（配置文件在conf i g 目录下）。

创建 topic，测试 生产消费程序

在这里插入图片描述

kafka-topics.bat --zookeeper 127.0.0.1:2181 --list

kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic itheima

kafka-topics.bat --delete --zookeeper localhost:2181 --topic itheima

在这里插入图片描述

kafka-console-producer.bat --broker-list localhost:9092 --topic itheima2022

kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic itheima2022 --from-beginning

Spr ingBoot整合Kafka

导坐标

在这里插入图片描述

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

做配置

在这里插入图片描述

spring: 
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: order

做客户端

在这里插入图片描述

@Service
public class MessageServiceKafkaImpl implements MessageService {

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;
    @Override
    public void sendMessage(String id) {
        System.out.println("待发送短信的订单已纳入处理队列 （kafka）,id:" + id);
        kafkaTemplate.send("itheima2022",id);
    }

    @Override
    public String doMessage() {
        return null;
    }
}

在这里插入图片描述

@Component
public class MessageListener {

    @KafkaListener(topics = "itheima2022")
    public void onMessage(ConsumerRecord<String,String> record){
        System.out.println("已完成短信发送业务(kafka)：id:"+record.value());

    }
}