从零开始搭建Kafka+SpringBoot分布式消息系统

时间 2020-12-08

标签 html java node linux web redis spring shell 数据库 apache 栏目 Kafka 繁體版

原文原文链接

前言

因为kafka强依赖于zookeeper，因此需先搭建好zookeeper集群。因为zookeeper是由java编写的，需运行在jvm上，因此首先应具有java环境。
（ps：默认您的centos系统可联网，本教程就不教配置ip什么的了）
（ps2：没有wget的先装一下：yum install wget）
（ps3：人啊，就是要条理。东边放一点，西边放一点，过段时间就不知道本身装在哪里了。本教程全部下载均放在/usr/local目录下）
（ps4：kafka可能有内置zookeeper，感受能够越过zookeeper教程，可是这里也配置出来了。我没试过）html

文章首发公众号：Java架构师联盟java

1、配置jdk

由于oracle 公司不容许直接经过wget 下载官网上的jdk包。因此你直接wget如下地址下载下来的是一个只有5k的网页文件而已，并非须要的jdk包。（垄断地位就是任性）。
（请经过java -version判断是否自带jdk，个人没带）node

一、官网下载

下面是jdk8的官方下载地址：linux

https://www.oracle.com/technetwork/java/javase/downloads/java-archive-javase8u211-later-5573849.html

二、上传解压

这里经过xftp上传到服务器指定位置：/usr/localweb

对压缩文件进行解压：redis

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

对解压后的文件夹进行更名：spring

mv jdk1.8.0_221 jdk1.8

三、配置环境变量

vim /etc/profile
#java environment
export JAVA_HOME=/usr/local/jdk1.8
export CLASSPATH=.:${JAVA_HOME}/jre/lib/rt.jar:${JAVA_HOME}/lib/dt.jar:${JAVA_HOME}/lib/tools.jar
export PATH=$PATH:${JAVA_HOME}/bin

操做以后的界面以下：shell

运行命令使环境生效数据库

source /etc/profile

2、搭建zookeeper集群

一、下载zookeeper

建立zookeeper目录，在该目录下进行下载：apache

mkdir /usr/local/zookeeper

这一步若是出现链接被拒绝时可多试几回，我就是第二次请求才成功的。

wget http://archive.apache.org/dist/zookeeper/zookeeper-3.4.6/zookeeper-3.4.6.tar.gz

等待下载完成以后解压：

tar -zxvf zookeeper-3.4.6.tar.gz

重命名为zookeeper1

mv zookeeper-3.4.6 zookeeper1
cp -r zookeeper1 zookeeper2
cp -r zookeeper1 zookeeper3

二、建立data、logs文件夹

在zookeeper1目录下建立

在data目录下新建myid文件。内容为1

三、修改zoo.cfg文件

cd /usr/local/zookeeper/zookeeper1/conf/
cp zoo_sample.cfg zoo.cfg

进行过上面两步以后，有zoo.cfg文件了，如今修改内容为：

dataDir=/usr/local/zookeeper/zookeeper1/data
dataLogDir=/usr/local/zookeeper/zookeeper1/logs
server.1=192.168.233.11:2888:3888
server.2=192.168.233.11:2889:3889
server.3=192.168.233.11:2890:3890

四、搭建zookeeper2

首先，复制更名。

cd /usr/local/zookeeper/
cp -r zookeeper1 zookeeper2

而后修改具体的某些配置：

vim zookeeper2/conf/zoo.cfg

将下图三个地方1改为2

vim zookeeper2/data/myid

同时将myid中的值改为2

五、搭建zookeeper3

同上，复制更名

cp -r zookeeper1 zookeeper3

vim zookeeper3/conf/zoo.cfg

修改成3

vim zookeeper3/data/myid

修改成3

六、测试zookeeper集群

cd /usr/local/zookeeper/zookeeper1/bin/

因为启动所需代码比较多，这里简单写了一个启动脚本：

vim start

start的内容以下

cd /usr/local/zookeeper/zookeeper1/bin/
./zkServer.sh start ../conf/zoo.cfg
cd /usr/local/zookeeper/zookeeper2/bin/
./zkServer.sh start ../conf/zoo.cfg
cd /usr/local/zookeeper/zookeeper3/bin/
./zkServer.sh start ../conf/zoo.cfg

下面是链接脚本：

vim login

login内容以下：

./zkCli.sh -server 192.168.233.11:2181,192.168.233.11:2182,192.168.233.11:2183

脚本编写完成，接下来启动：

sh start
sh login

启动集群成功，以下图：

这里zookeeper就告一段落了，因为zookeeper占用着输入窗口，这里能够在xshell右键标签，新建ssh渠道。而后就能够在新窗口继续操做kafka了！

3、搭建kafka集群

一、下载kafka

首先建立kafka目录：

mkdir /usr/local/kafka

而后在该目录下载

cd /usr/local/kafka/
wget https://archive.apache.org/dist/kafka/1.1.0/kafka_2.11-1.1.0.tgz

下载成功以后解压：

tar -zxvf kafka_2.11-1.1.0.tgz

二、修改集群配置

首先进入conf目录下：

cd /usr/local/kafka/kafka_2.11-1.1.0/config

修改server.properties
修改内容：

broker.id=0
log.dirs=/tmp/kafka-logs
listeners=PLAINTEXT://192.168.233.11:9092

复制两份server.properties

cp server.properties server2.properties
cp server.properties server3.properties

修改server2.properties

vim server2.properties

修改主要内容为：

broker.id=1
log.dirs=/tmp/kafka-logs1
listeners=PLAINTEXT://192.168.233.11:9093

如上，修改server3.properties
修改内容为：

broker.id=2
log.dirs=/tmp/kafka-logs2
listeners=PLAINTEXT://192.168.233.11:9094

三、启动kafka

这里仍是在bin目录编写一个脚本：

cd ../bin/
vim start

脚本内容为：

./kafka-server-start.sh ../config/server.properties &
./kafka-server-start.sh ../config/server2.properties &
./kafka-server-start.sh ../config/server3.properties &

经过jps命令能够查看到，共启动了3个kafka。

四、建立Topic

cd /usr/local/kafka/kafka_2.11-1.1.0
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic

kafka打印了几条日志

在启动的zookeeper中能够经过命令查询到这条topic！

ls /brokers/topics

查看kafka状态

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic

能够看到此时有三个节点 1 ， 2 ， 0

Leader 是1 ，
由于分区只有一个因此在0上面，
Replicas：主从备份是 1,2,0，
ISR（in-sync）：如今存活的信息也是 1,2,0

五、启动生产者

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic

因为不能按删除，不能按左右键去调整，因此语句有些乱啊。em…

六、启动消费者

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic

能够看出，启动消费者以后就会自动消费。

在生产者又造了一条。

消费者自动捕获成功！

4、集成springboot

先贴一张kafka兼容性目录：

不知足的话启动springboot的时候会抛异常的！！！ps：该走的岔路我都走了o(╥﹏╥)o
（个人kafka-clients是1.1.0，spring-kafka是2.2.2，中间那列暂时不用管）

回归正题，搞了两个小时，终于搞好了，想哭…
遇到的问题基本就是jar版本不匹配。
上面的步骤我也都会相应的去修改，争取你们按照本教程一遍过！！！

一、pom文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.gzky</groupId>
    <artifactId>study</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>study</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-redis</artifactId>
            <version>1.3.8.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.2.0.RELEASE</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
        </dependency>

    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

pom文件中，重点是下面这两个版本。

<parent>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-parent</artifactId>
       <version>2.1.1.RELEASE</version>
       <relativePath/> <!-- lookup parent from repository -->
   </parent>
<dependency>
      <groupId>org.springframework.kafka</groupId>
      <artifactId>spring-kafka</artifactId>
      <version>2.2.0.RELEASE</version>
</dependency>

二、application.yml

spring:
  redis:
    cluster:
      #设置key的生存时间，当key过时时，它会被自动删除；
      expire-seconds: 120
      #设置命令的执行时间，若是超过这个时间，则报错;
      command-timeout: 5000
      #设置redis集群的节点信息，其中namenode为域名解析，经过解析域名来获取相应的地址;
      nodes: 192.168.233.11:9001,192.168.233.11:9002,192.168.233.11:9003,192.168.233.11:9004,192.168.233.11:9005,192.168.233.11:9006
  kafka:
    # 指定kafka 代理地址，能够多个
    bootstrap-servers: 192.168.233.11:9092,192.168.233.11:9093,192.168.233.11:9094
    producer:
      retries: 0
      # 每次批量发送消息的数量
      batch-size: 16384
      buffer-memory: 33554432
      # 指定消息key和消息体的编解码方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      # 指定默认消费者group id
      group-id: test-group
      auto-offset-reset: earliest
      enable-auto-commit: true
      auto-commit-interval: 100
      # 指定消息key和消息体的编解码方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

server:
  port: 8085
  servlet:
    #context-path: /redis
    context-path: /kafka

没有配置Redis的能够把Redis部分删掉，也就是下图：
想学习配置Redis集群的能够参考：《Redis集群redis-cluster的搭建及集成springboot》

三、生产者

package com.gzky.study.utils;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

/**
 * kafka生产者工具类
 *
 * @author biws
 * @date 2019/12/17
 **/
@Component
public class KfkaProducer {

    private static Logger logger = LoggerFactory.getLogger(KfkaProducer.class);

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 生产数据
     * @param str 具体数据
     */
    public void send(String str) {
        logger.info("生产数据：" + str);
        kafkaTemplate.send("testTopic", str);
    }
}

四、消费者

package com.gzky.study.utils;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * kafka消费者监听消息
 *
 * @author biws
 * @date 2019/12/17
 **/
@Component
public class KafkaConsumerListener {

    private static Logger logger = LoggerFactory.getLogger(KafkaConsumerListener.class);

    @KafkaListener(topics = "testTopic")
    public void onMessage(String str){
        //insert(str);//这里为插入数据库代码
        logger.info("监听到：" + str);
        System.out.println("监听到：" + str);
    }

}

五、对外接口

package com.gzky.study.controller;

import com.gzky.study.utils.KfkaProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

/**
 * kafka对外接口
 *
 * @author biws
 * @date 2019/12/17
 **/
@RestController
public class KafkaController {

    @Autowired
    KfkaProducer kfkaProducer;

    /**
     * 生产消息
     * @param str
     * @return
     */
    @RequestMapping(value = "/sendKafkaWithTestTopic",method = RequestMethod.GET)
    @ResponseBody
    public boolean sendTopic(@RequestParam String str){
        kfkaProducer.send(str);
        return true;
    }
}

六、postman测试

这里首先应该在服务器启动监听器（kafka根目录）,下面命令必须是具体的服务器ip，不能是localhost，是我踩过的坑：

推荐此处重启一下集群
关闭kafka命令：

cd /usr/local/kafka/kafka_2.11-1.1.0/bin
./kafka-server-stop.sh ../config/server.properties &
./kafka-server-stop.sh ../config/server2.properties &
./kafka-server-stop.sh ../config/server3.properties &

此处应该jps看一下，等待全部的kafka都关闭（关不掉的kill掉），再从新启动kafka：

./kafka-server-start.sh ../config/server.properties &
./kafka-server-start.sh ../config/server2.properties &
./kafka-server-start.sh ../config/server3.properties &

等待kafka启动成功后，启动消费者监听端口：

cd /usr/local/kafka/kafka_2.11-1.1.0
bin/kafka-console-consumer.sh --bootstrap-server 192.168.233.11:9092 --from-beginning --topic testTopic

曾经我乱输的测试信息所有被监听过来了！

启动springboot服务

而后用postman生产消息：

而后享受成果，服务器端监听成功。

项目中也监听成功！