ubuntu版本
java安装
- sudo apt-get update
- sudo apt-get install openjdk-8-jdk
- java --version
zookeeper安装
下载
wget https://archive.apache.org/dist/zookeeper/zookeeper-3.4.6/zookeeper-3.4.6.tar.gz解压缩
tar -zxvf zookeeper-3.4.6.tar.gz进入目录 zookeeper-3.4.6/conf，将 zoo_sample.cfg重命名为 zoo.cfg
cp zoo_sample.cfg zoo.cfg启动zk
bin/zkServer.sh start查看zk启动状态
bin/zkServer.sh status
kafka安装
下载
wget https://archive.apache.org/dist/kafka/2.4.1/kafka_2.11-2.4.1.tgz解压缩
tar -zxvf kafka_2.11-2.4.1.tgz进入 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 &正常kafka能启动，如果kafka报错如下
Caused by: java.net.UnknownHostException: xxx: Name or service not known
修改 config/server.properties 加上配置 listeners=PLAINTEXT://10.0.0.3:8090
发送生产消息
./bin/kafka-console-producer.sh --broker-list 10.0.0.3:8090 --topic mytest --producer.config config/producer.properties接收生产消息
./bin/kafka-console-consumer.sh --bootstrap-server 10.0.0.3:8090 --topic mytest --from-beginning --consumer.config config/consumer.properties远程外网连接使用
以上配置，用外网进行连接的时候，会报超时
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配置
- 修改 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
- 在 config 目录下增加 kafka_server_jaas.conf 文件（名字自定义即可），文件里面的配置的内容
KafkaServer{
org.apache.kafka.common.security.scram.ScramLoginModule required
username="admin"
password="admin-sec"
};
KafkaServer的意思是指，kafka broker 之间通讯的账号和密码
- 将账号信息，通过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
- 配置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"
- 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'");
- 如果要在服务器上测试，则需要做以下配置
# 生产者 - 修改配置 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节点的公网地址进行访问）
