【Springboot】整合kafka

发布于:2023-09-16 ⋅ 阅读:(60) ⋅ 点赞:(0)

安装zookeeper

jdk安装

环境准备:CentOS7,jdk1.8
步骤如下:

  • 下载自己需要的版本
    这里使用的jdk1.8,获取链接如下
    链接:https://pan.baidu.com/s/1Bkvy0ChTVH0hrKkxiE4ZDg 提取码:1zws

  • 上传到安装目录并解压
    使用linux连接工具,将下载的JDK压缩包上传到要安装的目录下,这里使用的是windTerm,我是将压缩包上传到/usr/java目录下,解压命令如下

    tar -zxvf jdk-8u152-linux-x64.tar.gz
    

    解压后目录下面多个文件夹
    在这里插入图片描述

  • 配置环境变量

    vim /etc/profile
    

    在文件最后面添加如下配置,JAVA_HOME是自己安装的目录,需要改为自己的目录

    export JAVA_HOME=/usr/java/jdk1.8.0_152
    export CLASSPATH=$:CLASSPATH:$JAVA_HOME/lib/    
    export PATH=$PATH:$JAVA_HOME/bin
    

    重新加载系统环境变量

    source /etc/profile
    
  • 验证是否安装成功

    java -version
    

    出现如下内容说明安装成功
    在这里插入图片描述

zookeeper安装

  • 下载压缩包:
    官网地址:https://zookeeper.apache.org/releases.html

  • 上传到安装目录并解压
    我的是/usr/zookeeper,解压压缩包
    在这里插入图片描述

  • 修改配置文件
    进入解压后的文件夹里面的conf目录下,该目录下有一个名zoo_sample.cfg文件,将该文件重新名为zoo.cfg,或者重新复制一份命名为zoo.cfg。使用vim命令打开zoo.cfg文件编辑,主要修改两个地方,data,log需要自己创建,我是直接在zookeeper安装目录下创建的
    在这里插入图片描述

  • 启动zookeeper

    进入zookeeper安装目录下的bin目录
    在这里插入图片描述

    进入bin目录如下
    在这里插入图片描述

    启动zookeeper服务端:./zkServer.sh start
    在这里插入图片描述

    查看启动状态:./zkServer.sh status
    在这里插入图片描述

    启动zookeeper客户端

    	[root@localhost bin]# ./zkCli.sh
    

安装kafka(非集群)

  • 首先是下载自己需要的压缩包
    官网下载地址:https://kafka.apache.org/downloads
  • 上传到安装目录并解压
    tar -zxvf kafka压缩包名称
    
  • 修改kafka的server.properties配置文件
    进入解压后的文件能看到如下内容,server.properties配置文件在config目录下
    在这里插入图片描述
    我这里是非集群,修改配置文件也比较简单,使用vi或者vim文本编辑命令打开server.properties文件,修改下面配置
    #kafka服务器地址
    listeners=PLAINTEXT://10.3.1.156:9092 
    
    #zk服务器地址,我这里zk跟kafka在一台虚拟机上
    zookeeper.connect=10.3.1.156:2181
    
    log.dirs=/usr/kafka/kafka-logs
    
    非集群方法配置就完成了,下面就是启动kafka服务,进入kafka文件下的bin目录下,使用命令的方式体验消息发送和接收
    启动kafka
    ./kafka-server-start.sh -daemon ../config/server.properties
    
    创建topic
    ./kafka-topics.sh --bootstrap-server 10.3.1.156:9092 --create --topic test --replication-factor 1 --partitions 1
    
    查看topic
    ./kafka-topics.sh --list --bootstrap-server 10.3.1.156:9092
    
    发消息
    ./kafka-console-producer.sh --bootstrap-server 10.3.1.156:9092 --topic test
    
    接收消息
    ./kafka-console-consumer.sh --bootstrap-server 10.3.1.156:9092 --topic test
    
    删除topic
    ./kafka-topics.sh --bootstrap-server 10.3.1.156:9092 --delete --topic test
    

springboot项目整合配置

  • 新建的项目pom文件中导入kafka依赖
    <dependency>
       <groupId>org.springframework.kafka</groupId>
       <artifactId>spring-kafka</artifactId>
    </dependency>
    
  • 配置文件添加kafka相关配置
    server:
      port: 8080
    spring:
      kafka:
        bootstrap-servers: 10.3.1.156:9092 #kafka服务地址
        producer: #生产者
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
        consumer: #消费者
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          auto-offset-reset: earliest
    
  • 客户端发送消息
    @RestController
    public class KafkaProducerController {
    
        @Autowired
        private KafkaTemplate<String, String> kafkaTemplate;
    
        @GetMapping(value = "/sentMessage")
        public String sentMessage(@RequestParam("msg") String msg){
            kafkaTemplate.send("test", msg);
            return "发送成功";
        }
    }
    
  • 消费者接收消息
    @Component
    public class KafkaConsumerListener {
        @KafkaListener(topics = "test", groupId = "my-test-group")
        public void consumer(ConsumerRecord<String,String> record){
            System.out.println("接收到的消息:" + record.value());
        }
    }
    
本文含有隐藏内容,请 开通VIP 后查看

网站公告

今日签到

点亮在社区的每一天
去签到