Kafka集群搭建與SpringBoot項目集成

本篇文章的目的是幫助Kafka初學者快速搭建一個Kafka集群，以及怎么在SpringBoot項目中使用Kafka。

kafka集群環境包地址：https://pan.baidu.com/s/1Mar6ADov94lUNMCR6eyFEA             提取碼：x9yn

一、Kafka集群搭建

1、准備環境

（1）准備三台LINUX服務器：

xxx.xxx.xxx.1

xxx.xxx.xxx.2

xxx.xxx.xxx.3

（2）jdk版本大於1.8即可，我是1.8.0_181

（3）在三台服務器上創建用戶admin，將環境放到admin用戶下，嫌麻煩的同學也可以直接使用root用戶安裝（真實生產上不建議這么做）

tips：LINUX怎么給普通用戶賦文件夾操作權限？

• 切換到root用戶
• 使用chown -R admin:admin /home/admin命令
• 執行su - admin命令就可以切換用戶並定位到/home/admin下

（4）一定要關閉三台服務器的防火牆，不然安裝肯定會出問題，切記！這個真的很重要！

2、搭建Zookeeper集群

（1）解壓zookeeper-3.4.12.tar.gz，進入zookeeper文件夾

 （2）進入conf文件夾

1）復制zoo.cfg文件
cp zoo.cfg zoo_sample.cfg

2）修改zoo.cfg文件
vim zoo.cfg

 
　　這里的3個IP的作用如下：

2181：對cline端提供服務

3888：選舉leader使用

2888：集群內機器通訊使用（Leader監聽此端口）

（3）進入data文件夾，若沒有自己創建一個

　　在data文件夾下創建myid文件，三台機器分別填入server對應的ID，這里我是1、2、3

（4）啟動zookeeper集群

• 1. 啟動ZK服務:         sh bin/zkServer.sh start
• 2. 查看ZK服務狀態:  sh bin/zkServer.sh status
• 3. 停止ZK服務:         sh bin/zkServer.sh stop
• 4. 重啟ZK服務:         sh bin/zkServer.sh restart

（5）三台機器都需要重復上述操作，注意myid中的ID要對應

 

3、搭建Kafka集群

（1）解壓kafka_2.12-2.5.0.tgz，進入kafka文件夾

（2）進入config文件夾，修改 server.properties內容

# Kafka使用唯一的一個整數來標識每個broker，該參數默認是-1。如果不指定，kafka會自動生成一個唯一值
broker.id=1
# broker監聽器的CSV列表，格式是[協議]://[主機名]:[端口]。
listeners=PLAINTEXT://xxx.xxx.xxx.1:9092
# 非常重要的參數！該參數指定了kafka持久化消息的目錄。該參數可以設置多個目錄，以逗號分隔，比如/home/kafka1,/home/kafka2,多目錄的做法是推薦的
log.dirs=/tmp/kafka-logs
# 同樣是很重要的參數！這個參數完全沒有默認值，是必須要自己設置的
zookeeper.connect=xxx.xxx.xxx.1:2181,xxx.xxx.xxx.2:2181,xxx.xxx.xxx.3:2181
# 是否開啟unclean leader選舉。由於開始可能不能保證數據一致性，所以設置為false
unclean.leader.election.enable=false
# topic 在當前 broker 上的分區個數
num.partitions=1
# 用來恢復和清理 data 下數據的線程數量
num.recovery.threads.per.data.dir=1
# segment文件保留的最長時間，超時將被刪除
log.retention.hours=16
# 刪除 topic 功能使能 ( 允許刪除數據 ) ( 手動指定 )
delete.topic.enable=true
# 處理網絡請求的線程數量
num.network.threads=3
# 用來處理磁盤 IO 的線程數量
num.io.threads=8
# 發送套接字的緩沖區大小
socket.send.buffer.bytes=102400
# 接收套接字的緩沖區大小
socket.receive.buffer.bytes=102400
# 請求套接字的緩沖區大小
socket.request.max.bytes=10485760

（3）配置環境變量

vim ~/.bash_profile

# KAFKA_HOME export KAFKA_HOME=/export/servers/kafka_2.11-0.11.0.0
export PATH=$PATH:$KAFKA_HOME/bin

（4）啟動kafka集群
     啟動 ：bin/kafka-server-start.sh config/server.properties &
     關閉 ：bin/kafka-server-stop.sh stop

二、使用kafkatool工具操作Kafka

這里提供一篇詳細操作：https://www.cnblogs.com/frankdeng/p/9452982.html

三、Kafka與SpringBoot集成

1、pom.xml導入

<?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.3.0.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.sunyard.bigdata</groupId>
    <artifactId>springbootkafka</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>springbootkafka</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.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </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.kafka</groupId>
            <artifactId>spring-kafka-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

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

</project>

2、application.properties配置

server.port=9001
spring.application.name=kafka

#### kafka配置生產者 begin ####
# 指定kafka server的地址，集群配多個，中間，逗號隔開
spring.kafka.bootstrap-servers=xxx.xxx.xxx.1:9092,xxx.xxx.xxx.2:9092,xxx.xxx.xxx.3:9092
# 寫入失敗時，重試次數。當leader失效，一個repli節點會替代成為leader節點，此時可能出現寫入失敗，
# 當retris為0時，produce不會重復。retirs重發，此時repli節點完全成為leader節點，不會產生消息丟失。
spring.kafka.producer.retries=0
# 每次批量發送消息的數量,produce積累到一定數據，一次發送
spring.kafka.producer.batch-size=16384
# produce積累數據一次發送，緩存大小達到buffer.memory就發送數據
spring.kafka.producer.buffer-memory=33554432
#procedure要求leader在考慮完成請求之前收到的確認數，用於控制發送記錄在服務端的持久化，其值可以為如下：
#acks = 0 如果設置為零，則生產者將不會等待來自服務器的任何確認，該記錄將立即添加到套接字緩沖區並視為已發送。在這種情況下，無法保證服務器已收到記錄，並且重試配置將不會生效（因為客戶端通常不會知道任何故障），為每條記錄返回的偏移量始終設置為-1。
#acks = 1 這意味着leader會將記錄寫入其本地日志，但無需等待所有副本服務器的完全確認即可做出回應，在這種情況下，如果leader在確認記錄后立即失敗，但在將數據復制到所有的副本服務器之前，則記錄將會丟失。
#acks = all 這意味着leader將等待完整的同步副本集以確認記錄，這保證了只要至少一個同步副本服務器仍然存活，記錄就不會丟失，這是最強有力的保證，這相當於acks = -1的設置。
#可以設置的值為：all, -1, 0, 1
spring.kafka.producer.acks=1
# 指定消息key和消息體的編解碼方式
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
#### kafka配置生產者 end ####

#### kafka配置消費者 start ####
# 指定默認消費者group id --> 由於在kafka中，同一組中的consumer不會讀取到同一個消息，依靠groud.id設置組名
spring.kafka.consumer.group-id=test1
# smallest和largest才有效，如果smallest重新0開始讀取，如果是largest從logfile的offset讀取。一般情況下我們都是設置smallest
spring.kafka.consumer.auto-offset-reset=earliest
# enable.auto.commit:true --> 設置自動提交offset
spring.kafka.consumer.enable-auto-commit=true
#如果'enable.auto.commit'為true，則消費者偏移自動提交給Kafka的頻率（以毫秒為單位），默認值為5000。
spring.kafka.consumer.auto-commit-interval=1000
# 指定消息key和消息體的編解碼方式
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
#### kafka配置消費者 end ####

3、啟動類代碼

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.EnableKafka;
 @SpringBootApplication @EnableKafka public class SpringbootkafkaApplication {
    public static void main(String[] args) {
        SpringApplication.run(SpringbootkafkaApplication.class, args);
    }
}

4、生產者代碼

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/api/kafka/")
public class KafkaController {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    @GetMapping("send")
    @ResponseBody
    public boolean send(@RequestParam String message) {
        try {
            kafkaTemplate.send("test-topic", message);
            kafkaTemplate.send("test-topic2", message);
            System.out.println("消息發送成功...");
        } catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }
    @GetMapping("test")
    @ResponseBody
    public String test() {
        System.out.println("hello world!");
        return "ok";
    }
}

5、消費者代碼

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class ConsumerListener {

    @KafkaListener(topics = "test-topic")
    public void onMessage1(String message) {
        System.out.println("我是第一個消費者:" + message);
    }
    @KafkaListener(topics = "test-topic2")
    public void onMessage2(String message) {
        System.out.println("我是第二個消費者:" + message);
    }
}