Kafka集群部署

前置条件

安装环境：操作系统为CentOS，已配置免密登录，已配置Java环境变量。
依赖环境：zookeeper环境，参考章节：大数据集群搭建，Zookeeper集群安装搭建。
安装介质：kafka_2.12-3.2.0.tgz，地址：https://kafka.apache.org/

上传安装介质并解压

$ 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

分发至其他服务器：$ xsync.sh kafka
分别修改：broker.id，node54为：4，node55为：5。
配置kafka环境变量： sudo vim /etc/profile.d/my_env.sh

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

分发：sudo /home/admin/bin/xsync.sh /etc/profile.d/my_env.sh
注：需要配置root的免密登录。

$ source /etc/profile
$ echo $KAFKA_HOME

Kafka启停命令

启动： bin/kafka-server-start.sh -daemon config/server.properties
停止： bin/kafka-server-stop.sh
启动后查看进程，进程为：Kafka

Kafka集群统一管理

kaf.sh管理脚本

#!/bin/bash

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

case $1 in
"start")
    for host in node53 node54 node55
    do
        echo "================启动 $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 "================关闭 $host================="
        ssh $host "/opt/module/kafka/bin/kafka-server-stop.sh"
    done
;;
esac

操作脚本

启动： $ kaf.sh start
关闭： $ kaf.sh stop

Kafka主题、生产者和消费者命令

主题命令行

# 版本
bin/kafka-topics.sh --version
# 帮助
bin/kafka-topics.sh --help
# 集群列表
bin/kafka-topics.sh --bootstrap-server node53:9092,node54:9092 --list
# 创建topic，1个分区3个副本
bin/kafka-topics.sh --bootstrap-server node53:9092,node54:9092 --topic hello --create --partitions 1 --replication-factor 3
# 查看主题的详细信息
bin/kafka-topics.sh --bootstrap-server node53:9092,node54:9092 --topic hello --describe
# 修改分区(分区只能增加不能减少)
bin/kafka-topics.sh --bootstrap-server node53:9092,node54:9092 --topic hello --alter --partitions 2

生产者命令行

# 版本
bin/kafka-console-producer.sh --version
# 帮助
bin/kafka-console-producer.sh --help
# 生产消息
bin/kafka-console-producer.sh --bootstrap-server node53:9092,node54:9092 --topic hello
> hello1
> hello2
> hello3

消费者命令行

# 版本
bin/kafka-console-consumer.sh --version
# 帮助
bin/kafka-console-consumer.sh --help
# 消费消息
bin/kafka-console-consumer.sh --bootstrap-server node53:9092,node54:9092 --topic hello
# 消费消息(历史数据)
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>

消费者客户端代码

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();
        // 多台服务用,号分割
        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消费者
        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()));
            }
        }
    }

}

生产者客户端代码

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();
        // 多台服务用,号分割
        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生产者
        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("消息发送成功:" + msg);
                Thread.sleep(3000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            kafkaProducer.close();
        }
    }

}