kafka单机部署（含SASL认证）

Dear 丶 2022-12-20 02:25 224阅读 0赞

适用于搭建zookeeper+kafka本地测试环境，使用密码认证，并给出相关shell测试例子。

### 文章目录 ###

*   *  一 下载安装
     *  二 zookeeper
     *   *  1. 配置
         *  2. 启动
     *  三 kafka
     *   *  1. 配置
         *   *  1. 修改server.properties
             *  2. 添加 kafka\_jaas.conf 文件
         *  2. 启动
     *  四 测试
     *   *   *  0. 编写连接配置文件kafka\_client.properties
             *  1. 创建topic
             *  2. 浏览topic
             *  3. producer
             *  4. consumer
             *  5. 查看consumer group列表

## 一 下载安装 ##

直接去官网下载：https://kafka.apache.org/downloads 即可  
我这里用的是 kafka\_2.12-2.3.1.tgz ，即 scala2.12，kafka2.3.1版本

假设安装包在/opt目录下，且欲设置 KAFKA\_HOME 为 /opt/kafka，则：

tar -zxf kafka_2.12-2.3.1.tgz
    rm -f kafka*.tgz
    mv kafka* kafka
    echo -e "export KAFKA_HOME=$(pwd)/kafka\n">>/etc/profile
    source /etc/profile

解压kafka tgz包到目标文件夹  
修改解压后文件夹名称为kafka  
添加 KAFKA\_HOME 环境变量指向该目录

## 二 zookeeper ##

### 1. 配置 ###

$KAFKA\_HOME/config/zookeeper.properties

根据需求修改，基础修改属性有： zookeeper 的数据存储目录 dataDir。

推荐阅读官方文档：https://cwiki.apache.org/confluence/display/ZOOKEEPER/ZooKeeper+and+SASL  
一般需要启用 Clinet-Server 认证，Server-Server 认证我个人则没有使用。  
生产环境中多使用 Kerberos，测试用DIGEST-MD5密码认证就行了。

这里用的zookeeper是kafka自带的，配置大同小异。

1.  向 zookeeper.properties 添加以下配置：

authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
    requireClientAuthScheme=sasl
    jaasLoginRenew=3600000

1.  添加 zookeeper\_jaas.conf 文件  
    以下配置可拆分成两个文件，Server用于服务端进行认证，可配置多个用户。Client用于客户端连接，一般只需通过username和password指定一个。

Server {
           org.apache.zookeeper.server.auth.DigestLoginModule required
           user_admin="123456"
           user_root="123456"
           ;
    };
    Client {
           org.apache.zookeeper.server.auth.DigestLoginModule required
           username="admin"
           password="123456"
           ;
    };

### 2. 启动 ###

export KAFKA_OPTS="-Djava.security.auth.login.config=$KAFKA_HOME/config/zookeeper_jaas.conf"
    bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

## 三 kafka ##

以下是我的配置  
我把端口改成了19092  
数据存储目录为：log.dirs  
offsets.topic.replication.factor 生产环境建议设置为3，这里由于是单机，如果设置超过1会报错，相关配置参考：https://blog.csdn.net/alinyua/article/details/108087917

### 1. 配置 ###

#### 1. 修改server.properties ####

#当前机器在集群中的唯一标识，和zookeeper的myid性质一样
    broker.id=0
    #当前kafka对外提供服务的端口，默认是9092
    listeners=SASL_PLAINTEXT://0.0.0.0:19092
    advertised.listeners=SASL_PLAINTEXT://localhost:19092
    security.inter.broker.protocol=SASL_PLAINTEXT
    sasl.enabled.mechanisms=PLAIN
    sasl.mechanism.inter.broker.protocol=PLAIN
    #这个是borker进行网络处理的线程数
    num.network.threads=3
    #这个是borker进行I/O处理的线程数
    num.io.threads=8
    
    #消息存放的目录，这个目录可以配置为“，”逗号分割的表达式，上面的num.io.threads要大于这个目录的个数这个目录，如果配置多个目录，新创建的topic他把消息持久化的地方是，当前以逗号分割的目录中，那个分区数最少就放那一个
    log.dirs=/data/kafka
    
    #发送缓冲区buffer大小，数据不是一下子就发送的，先回存储到缓冲区了到达一定的大小后在发送，能提高性能
    socket.send.buffer.bytes=102400
    #kafka接收缓冲区大小，当数据到达一定大小后在序列化到磁盘
    socket.receive.buffer.bytes=102400
    #这个参数是向kafka请求消息或者向kafka发送消息的请请求的最大数，这个值不能超过java的堆栈大小
    socket.request.max.bytes=104857600
    
    #默认的分区数，一个topic默认1个分区数
    num.partitions=1
    num.recovery.threads.per.data.dir=1
    offsets.topic.replication.factor=1
    transaction.state.log.replication.factor=1
    transaction.state.log.min.isr=1
    default.replication.factor=1
    
    #默认消息的最大持久化时间，168小时，7天
    log.retention.hours=48
    #这个参数是：因为kafka的消息是以追加的形式落地到文件，当超过这个值的时候，kafka会新起一个文件
    log.segment.bytes=1073741824
    #每隔300000毫秒去检查上面配置的log失效时间（log.retention.hours=168 ），到目录查看是否有过期的消息如果有，删除
    log.retention.check.interval.ms=300000
    
    #设置zookeeper的连接端口
    zookeeper.connect=localhost:2181/kafka
    zookeeper.connection.timeout.ms=60000
    
    group.initial.rebalance.delay.ms=0

#### 2. 添加 kafka\_jaas.conf 文件 ####

建议直接阅读官方文档：https://kafka.apache.org/documentation/\#security\_sasl

因为kafka配置了sasl.mechanism.inter.broker.protocol为PLAIN，所以KafkaServer中需配置username和password用于Kafka内部认证，并提供k1,k2,k3供客户端连接。

KafkaServer使用root用户连接zookeeper

注意，现在一般不推荐kafka客户端直接连接zookeeper，如果需直接连接zookeeper，建议在KafkaServer设置与zookeeper相同的用户和密码，这样同一套jaas配置既可以连接KafkaServer也可以连接zookeeper。

KafkaServer{ 
        org.apache.kafka.common.security.plain.PlainLoginModule required
        username="root"
        password="123456"
        user_root="123456"
        user_k1="k1123456"
        user_k2="k2123456"
        user_k3="k3123456"
        ;
    };
    Client{ 
        org.apache.zookeeper.server.auth.DigestLoginModule required
        username="root"
        password="123456";
    };

### 2. 启动 ###

export KAFKA_OPTS="-Djava.security.auth.login.config=$KAFKA_HOME/config/kafka_jaas.conf"
    $KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties  >> logs/kafka.log 2>&1 &

## 四 测试 ##

#### 0. 编写连接配置文件kafka\_client.properties ####

security.protocol=SASL_PLAINTEXT
    sasl.mechanism=PLAIN
    sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="k1" password="k1123456";

#### 1. 创建topic ####

kafka-topics.sh 需提供command-config指定使用 kafka\_client.properties

bin/kafka-topics.sh --bootstrap-server localhost:19092 --create --topic my_topic_name \
            --partitions 10 --replication-factor 1 --command-config  config/kafka_client.properties

#### 2. 浏览topic ####

bin/kafka-topics.sh --list --bootstrap-server localhost:19092 --command-config  config/kafka_client.properties

#### 3. producer ####

kafka-console-producer.sh 需提供producer.config指定使用 kafka\_client.properties

bin/kafka-console-producer.sh --topic my_topic_name --broker-list localhost:19092 --producer.config config/kafka_client.properties

#### 4. consumer ####

kafka-console-consumer.sh 需提供consumer.config指定使用 kafka\_client.properties

bin/kafka-console-consumer.sh --topic my_topic_name --from-beginning --bootstrap-server localhost:19092 --consumer.config config/kafka_client.properties

#### 5. 查看consumer group列表 ####

kafka-console-groups.sh 需提供consumer.config指定使用 kafka\_client.properties

bin/kafka-consumer-groups.sh --bootstrap-server  localhost:19092 --list  --command-config config/kafka_client.properties