Kafka集群部署

前置條件

安裝環(huán)境：操作系統(tǒng)為CentOS，已配置免密登錄，已配置Java環(huán)境變量。
依賴環(huán)境：zookeeper環(huán)境，參考章節(jié)：大數(shù)據(jù)集群搭建，Zookeeper集群安裝搭建。
安裝介質(zhì)：kafka_2.12-3.2.0.tgz，地址：https://kafka.apache.org/

上傳安裝介質(zhì)并解壓

$ tar -zxvf kafka_2.12-3.2.0.tgz -C /opt/module/
$  cd /opt/module
$ mv kafka_2.12-3.2.0 kafka
$ cp config/server.properties config/server.properties.bak
$ vim config/server.properties

配置：config/server.properties，修改為：

broker.id=3
log.dirs=/opt/module/kafka/data
zookeeper.connect=node53:2181,node54:2181,node55:2181/kafka

分發(fā)至其他服務(wù)器：$ xsync.sh kafka
分別修改：broker.id，node54為：4，node55為：5。
配置kafka環(huán)境變量： sudo vim /etc/profile.d/my_env.sh

#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:KAFKA_HOME/bin

分發(fā)：sudo /home/admin/bin/xsync.sh /etc/profile.d/my_env.sh
注：需要配置root的免密登錄。

$ source /etc/profile
$ echo $KAFKA_HOME

Kafka啟停命令

啟動(dòng)： bin/kafka-server-start.sh -daemon config/server.properties
停止： bin/kafka-server-stop.sh
啟動(dòng)后查看進(jìn)程，進(jìn)程為：Kafka

Kafka集群統(tǒng)一管理

kaf.sh管理腳本

#!/bin/bash

#############
# sed 's/\r//' -i kaf.sh
#############

case $1 in
"start")
    for host in node53 node54 node55
    do
        echo "================啟動(dòng) $host================="
        ssh $host "/opt/module/kafka/bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties"
    done
;;
"stop")
    for host in node53 node54 node55
    do
        echo "================關(guān)閉 $host================="
        ssh $host "/opt/module/kafka/bin/kafka-server-stop.sh"
    done
;;
esac

操作腳本

啟動(dòng)： $ kaf.sh start
關(guān)閉： $ kaf.sh stop

Kafka主題、生產(chǎn)者和消費(fèi)者命令

主題命令行

# 版本
bin/kafka-topics.sh --version
# 幫助
bin/kafka-topics.sh --help
# 集群列表
bin/kafka-topics.sh --bootstrap-server node53:9092,node54:9092 --list
# 創(chuàng)建topic，1個(gè)分區(qū)3個(gè)副本
bin/kafka-topics.sh --bootstrap-server node53:9092,node54:9092 --topic hello --create --partitions 1 --replication-factor 3
# 查看主題的詳細(xì)信息
bin/kafka-topics.sh --bootstrap-server node53:9092,node54:9092 --topic hello --describe
# 修改分區(qū)(分區(qū)只能增加不能減少)
bin/kafka-topics.sh --bootstrap-server node53:9092,node54:9092 --topic hello --alter --partitions 2

生產(chǎn)者命令行

# 版本
bin/kafka-console-producer.sh --version
# 幫助
bin/kafka-console-producer.sh --help
# 生產(chǎn)消息
bin/kafka-console-producer.sh --bootstrap-server node53:9092,node54:9092 --topic hello
> hello1
> hello2
> hello3

消費(fèi)者命令行

# 版本
bin/kafka-console-consumer.sh --version
# 幫助
bin/kafka-console-consumer.sh --help
# 消費(fèi)消息
bin/kafka-console-consumer.sh --bootstrap-server node53:9092,node54:9092 --topic hello
# 消費(fèi)消息(歷史數(shù)據(jù))
bin/kafka-console-consumer.sh --bootstrap-server node53:9092,node54:9092 --topic hello --from-beginning

Java客戶端

pom.xml

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.2.0</version>
</dependency>

消費(fèi)者客戶端代碼

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Properties;

public class ConsumerDemo {

    public static void main(String[] args) {
        Properties props = new Properties();
        // 多臺(tái)服務(wù)用,號(hào)分割
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.2.53:9092,192.168.2.54:9092,192.168.2.55:9092");
        // 鍵值序列化
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "hello");
        // Kafka消費(fèi)者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props);
        // 訂閱消息
        kafkaConsumer.subscribe(Collections.singletonList("hello"));
        boolean flag = true;
        while (flag) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(String.format("topic:%s,offset:%d,消息:%s", record.topic(), record.offset(), record.value()));
            }
        }
    }

}

生產(chǎn)者客戶端代碼

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.Random;

public class ProductDemo {

    public static void main(String[] args) {
        Properties props = new Properties();
        // 多臺(tái)服務(wù)用,號(hào)分割
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.2.53:9092,192.168.2.54:9092,192.168.2.55:9092");
        // 鍵值序列化
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // Kakfa生產(chǎn)者
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(props);
        try {
            String topic = "hello";
            while (true) {
                String msg = "Hello," + new Random().nextInt(100);
                ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, msg);
                kafkaProducer.send(record);
                System.out.println("消息發(fā)送成功:" + msg);
                Thread.sleep(3000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            kafkaProducer.close();
        }
    }

}