Kafka集群配置SASL+ACL

                                 **  Kafka 集群配置SASL+ACL

测试环境：**

创新互联公司专业为企业提供兴城网站建设、兴城做网站、兴城网站设计、兴城网站制作等企业网站建设、网页设计与制作、兴城企业网站模板建站服务，10年兴城做网站经验，不只是建网站，更提供有价值的思路和整体网络服务。

                    系统: CentOS 6.5 x86_64
                    JDK : java version 1.8.0_121
                    kafka: kafka_2.11-1.0.0.tgz
                    zookeeper: 3.4.5
                    ip: 192.168.49.161 （我们这里在一台机上部署整套环境）

kafka 名词解析：

    Broker: Kafka 集群包含一个或多个服务器，这种服务器被称为broker
    Topic: 每条发布到Kafka 集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic 的消息
    分开存储，逻辑上一个Topic 的消息虽然保存于一个或多个broker 上但用户只需指定消息的Topic 即可生产或
    消费数据而不必关心数据存于何处）
    Partition: Parition 是物理上的概念，每个Topic 包含一个或多个Partition
    Producer: 负责发布消息到Kafka broker 即生产者
    Consumer: 消息消费者，向Kafka broker 读取消息的客户端即消费者
    Consumer Group：每个Consumer 属于一个特定的Consumer Group（可为每个Consumer 指定group
    name，若不指定group name 则属于默认的group）

Kafka 拓扑结构(借用别人的图)
Kafka 集群配置SASL+ACL
一个典型的Kafka 集群中包含若干Producer（可以是web 前端产生的Page View，或者是服务器日志，系统 CPU、Memory 等），若干broker（Kafka 支持水平扩展，一般broker 数量越多，集群吞吐率越高），若干Consumer Group，以及一个Zookeeper 集群。Kafka 通过Zookeeper 管理集群配置，选举leader，以及在Consumer Group 发生变化时进行rebalance。Producer 使用push 模式将消息发布到broker，Consumer 使用pull 模式从broker 订阅并消费消息

kafka 集群部署
一 Zookeeper 集群部署(这里部署的伪集群)

zookeeper 部署比较简单，解压修改配置文件后即可使用, 默认的配置文件为zoo_sample.cfg 这里我们直接新建一个zoo.cfg 的文件内容如下:

各参数意义这里不作详细解释，想了解可以自行参考官网

                                        tickTime=2000
                                        initLimit=5
                                        syncLimit=2
                                        dataDir=/opt/zook/zookeeper1/data //指定数据路径
                                        dataLogDir=/opt/zook/zookeeper1/logs //指定日志路径
                                        clientPort=2181 //监听端口
                                        server.1=127.0.0.1:7771:7772
                                        server.2=127.0.0.1:7771:7772
                                        server.3=127.0.0.1:7771:7772

  2. 在zookeeper/data 下创建一个名为myid 的文件,并在文件内输入数字1
 3. 集群部署则将上述配置的单节点再复制两份，并将配置文件中
                            ```
                            dataDir , dataLogDir 两个参数根据实际路径修改
                            clientPort 分别配为2182，2183
                            myid 分别配置为2 3
                            ```
  4. 启动集群

二 Kafka 集群部署

 由于集群最终是对外部网络开放的，出于安全考虑本文采用的是SASL+ACL 控制和授权

1. broker 配置
1.1. 将配置server.properties 复制2 份并分别重命名成server-1.properties server-2.properties server-3.properties 修改server.properties 配置:(请根据实际环境填写)

broker.id = 1 //另外两个节点分别为2 3
host.name=192.168.49.161
log.dirs=/tmp/kafka-logs-1 //另外两个节点分别为kafka-logs-2 kafka-logs-3 可以不
放在/tmp 下，如果放在其他地方必须注意目录权限
zookeeper.connect=192.168.49.161:2181,192.168.49.161:2182,192.168.49.161:2183
port=9092 //另外两个节点分别为9093 9094 也可以自定义
listeners=SASL_PLAINTEXT://192.168.49.161:9092 //另外两个节点分别为9093 9094

1.2 要配置SASL 和ACL，需要在broker 端进行两个方面的设置。
首先是创建包含所有认证用户信息的JAAS 文件,本例中我们有admin qjld 两个用户，其中admin 为管理员用于broker 集群间的认证， qjld 为远程应用的认证用户，新增认证信息文件

    kafka_server_jaas.conf 内容为:
    KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-mgr998778123"
    user_admin="admin-mgr998778123"
    user_qjld="123456";
    };

本例中路径为: /opt/kafka_2.11-1.0.0/config/kafka_server_jaas.conf , 我们需要将配置文件中的内容传递给JVM，因此需要修改/opt/kafka_2.11-1.0.0/bin/kafka-server-start.sh 脚本文件如下:

在exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@" 之前一行添加export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka_2.11-1.0.0/config/kafka_server_jaas.conf" 然后保存退出。

其次我们要在所有broker 配置文件(server-x.properties)中增加：

#ACL 入口类
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
#SASL_PLAINTEXT
sasl.enabled.mechanisms=PLAIN
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
allow.everyone.if.no.acl.found=true
#设置admin 为超级用户
super.users=User:admin

现在我们可以启动broker 测试了

bin/kafka-server-start.sh config/server-1.properties &
bin/kafka-server-start.sh config/server-2.properties &
bin/kafka-server-start.sh config/server-3.properties &

启动成功，此时broker 可以接收已认证的客户端连接了，下面我们进行客户端配置

//创建一个topic 名为test
bin/kafka-topics.sh--create --zookeeper 192.168.49.161:2181,192.168.49.161:2182, 192.168.49.161:2183 --replication-factor 1 --partitions 1 --topic test    
//查看topic 是否创建成功
bin/kafka-topics.sh --list --zookeeper 192.168.49.161:2181

2. Client 配置（用于使用kafka命令连接）
2.1 新增一个配置文件如：/opt/kafka_2.11-1.0.0/config/kafka_client_jaas.conf，内容为:

KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="qjld"
password="123456";
};

同理我们也要将配置文件内容传递给JVM, 因此需要修改

/opt/kafka_2.11-1.0.0/bin/kafka-console-producer.sh 脚本文件如下:在exec $base_dir/kafka-run-class.sh
$EXTRA_ARGS kafka.Kafka "$@" 之前一行添加export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka_2.11-1.0.0/config/kafka_client_jaas.conf"
 同样的方法修改kafka-console-consumer.sh

2.2 需要在config/consumer.properties 和config/producer.properties 配置文件中追加：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

这两行配置, 完成后再试运行：

bin/kafka-console-producer.sh --broker-list 192.168.49.161:9092 --topic test \
--producer.config config/producer.properties
bin/kafka-console-consumer.sh --bootstrap-server 192.168.49.161:9092 --topic test
--from-beginning  --consumer.config config/consumer.properties

接下来测试消息的写入和读取, 开启两个终端分别输入下列

bin/kafka-console-producer.sh --broker-list 192.168.49.161:9092 --topic test //消息写入
bin/kafka-console-consumer.sh --bootstrap-server 192.168.49.161:9092 --topic test  --from-beginning 消息读取

如果报错信息如下:
WARN Bootstrap broker 192.168.49.161:9092 disconnected (org.apache.kafka.clients.NetworkClient)
则说明配置的security 已生效, 要想普通用户能读写消息，需要配置ACL
2.3 配置ACL


bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties
zookeeper.connect=192.168.49.161:2181,192.168.49.161:2182,192.168.49.161:2183 --add
--allow-principal User:qjld --operation All --topic test
```这是为qjld 这个用户配置所有权限，如果有更细的要求则可参考网上的，如：Read Write 等只需要将--operation all 改为--operation Read 多种则在其后面加一个-- operation write 即可

**三    测试**
这里我选用python2.7 编写的脚本进行测试
先安装pip27 install kafka-python
![](/upload/otherpic50/5b755d258b838b83d79753ad5aad3d44.png)

标题名称：Kafka集群配置SASL+ACL
链接地址：http://scyanting.com/article/gcoeij.html

Kafka集群配置SASL+ACL

其他资讯