kafka sals/scram 单机版安装步骤

ubuntu版本

java安装

1. sudo apt-get update
2. sudo apt-get install openjdk-8-jdk
3. java --version

zookeeper安装

1. 下载
   wget https://archive.apache.org/dist/zookeeper/zookeeper-3.4.6/zookeeper-3.4.6.tar.gz

2. 解压缩
   tar -zxvf zookeeper-3.4.6.tar.gz

3. 进入目录 zookeeper-3.4.6/conf，将 zoo_sample.cfg重命名为 zoo.cfg
   cp zoo_sample.cfg zoo.cfg

4. 启动zk
   bin/zkServer.sh start

5. 查看zk启动状态
   bin/zkServer.sh status

kafka安装

1. 下载
   wget https://archive.apache.org/dist/kafka/2.4.1/kafka_2.11-2.4.1.tgz

2. 解压缩
   tar -zxvf kafka_2.11-2.4.1.tgz

3. 进入 kafka_2.11-2.4.1 启动kafka
   bin/kafka-server-start.sh config/server.properties
   后台启动：nohup bin/kafka-server-start.sh config/server.properties 2>&1 &

4. 正常kafka能启动，如果kafka报错如下

  Caused by: java.net.UnknownHostException: xxx: Name or service not known

修改 config/server.properties 加上配置 listeners=PLAINTEXT://10.0.0.3:8090

5. 发送生产消息
   ./bin/kafka-console-producer.sh --broker-list 10.0.0.3:8090 --topic mytest --producer.config config/producer.properties

6. 接收生产消息
   ./bin/kafka-console-consumer.sh --bootstrap-server 10.0.0.3:8090 --topic mytest --from-beginning --consumer.config config/consumer.properties

7. 远程外网连接使用
   以上配置，用外网进行连接的时候，会报超时

Send failed; nested exception is org.apache.kafka.common.errors.TimeoutException: Topic XXX_TOPIC not present in metadata after 30000 ms.

Connection to node 0 (/10.0.0.3:8090) could not be established. Broker may not be available.
[Producer clientId=producer-1] Connection to node 0 (/10.0.0.3:8090) could not be established. Broker may not be available.

增加以下配置即可： advertised.listeners=PLAINTEXT://xx_外网ip地址:8090

sals/scram配置

1. 修改 kafka_2.11-2.4.1/config 下的 server.properties 文件

# 开启 sasl认证，如果要开启ACL，则要加上下面一段配置，是针对topic进行权限控制
listeners=SASL_PLAINTEXT://10.0.0.3:8090
advertised.listeners=SASL_PLAINTEXT://xx_外网ip地址:8090
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
sasl.enabled.mechanisms=SCRAM-SHA-512
# false 只有配置了用户能访问 ； true 所有用户都能访问，只有部分不能访问，针对黑名单
allow.everyone.if.no.acl.found=false
super.users=User:admin


# ACL相关配置，配置了开启针对topic/用户 级别的读写控制，老版本
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

2. 在 config 目录下增加 kafka_server_jaas.conf 文件（名字自定义即可），文件里面的配置的内容

KafkaServer{
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="admin-sec"
};

KafkaServer的意思是指，kafka broker 之间通讯的账号和密码

3. 将账号信息，通过zookeeper的方式进行配置，2中 admin 账号也是通过这种方式配置

# 配置admin
./bin/kafka-configs.sh --zookeeper 127.0.0.1:2181 --alter --add-config 'SCRAM-SHA-512=[password=admin-sec]' --entity-type users --entity-name admin
# 配置生产者账号 producer
./bin/kafka-configs.sh --zookeeper 127.0.0.1:2181 --alter --add-config 'SCRAM-SHA-512=[password=ptest]' --entity-type users --entity-name ptest
# 配置消费者账号 consumer
./bin/kafka-configs.sh --zookeeper 127.0.0.1:2181 --alter --add-config 'SCRAM-SHA-512=[password=ctest]' --entity-type users --entity-name ctest



#如果开启了acl认证，则需要针对topic进行授权，以下是对 ptest 账号授权
./bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=127.0.0.1:2181 --add --allow-principal User:ptest --operation Write --topic mytes
# ctest
./bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=127.0.0.1:2181 --add --allow-principal User:ctest --operation Read --topic mytest
#授权消费组队topic的读权限
./bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=127.0.0.1:2181 --add --allow-principal User:ctest --operation Read --group test-consumer-group

4. 配置kafka启动加载的账号信息，修改 bin/kafka-server-start.sh文件，在配置文件中加上以下配置，在有效配置的最上面贴上就可以

export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka_2.11-2.4.1/config/kafka_server_jaas.conf"

5. Java客户端测试

#增加配置
Map<String, Object> props = new HashMap<>();
props.putxxx ...
props.put("security.protocol","SASL_PLAINTEXT");
props.put("sasl.mechanism","SCRAM-SHA-512");
props.put("sasl.jaas.config","org.apache.kafka.common.security.scram.ScramLoginModule required username='ctest' password='ctest'");

6. 如果要在服务器上测试，则需要做以下配置

# 生产者 - 修改配置 config/producer.properties 增加以下配置
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
    username="ptest" \
    password="ptest";

# 生产者启动命令
./bin/kafka-console-producer.sh --broker-list 10.0.0.3:8090 --topic mytest --producer.config config/producer.properties

# 消费者 - 修改配置 config/consumer.properties 增加以下配置
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
    username="ctest" \
password="ctest";

# 消费者启动命令
./bin/kafka-console-consumer.sh --bootstrap-server 10.0.0.3:8090 --topic mytest --from-beginning --consumer.config config/consumer.properties

问题及概念理解

1 .listeners与 advertised.listeners 的区别
listeners： 定义了kafka服务器内部监听的地址和端口
advertised.listeners 定义了向客户端公开的地址和端口，通过zk进行数据共享，它会保存在zk中 /brokers/ids/0 的endpoints中 。
（使用场景：当公网ip不是服务器网卡，而是通过代理绑定，无法通过listener进行绑定，只能通过0.0.0.0绑定，当外部需要访问kafka集群时，通过zk拿所有的broker节点的公网地址进行访问）