部署目录

/app/kafka -> kafka_2.12-1.0.0

数据目录

/app/data/kafka-logs

配置

一. 配置zookeeper

二. 配置kafka

* 证书配置

        #!/bin/bash
        #Step 1
        keytool -keystore server.keystore.jks -alias localhost -validity 365 -keyalg RSA -genkey
        #Step 2
        openssl req -new -x509 -keyout ca-key -out ca-cert -days 365
        keytool -keystore server.truststore.jks -alias CARoot -import -file ca-cert
        keytool -keystore client.truststore.jks -alias CARoot -import -file ca-cert
        #Step 3
        keytool -keystore server.keystore.jks -alias localhost -certreq -file cert-file
        openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days 365 -CAcreateserial -passin pass:test1234
        keytool -keystore server.keystore.jks -alias CARoot -import -file ca-cert
        keytool -keystore server.keystore.jks -alias localhost -import -file cert-signed

• 为客户端生成证书：

  keytool -keystore client.keystore.jks -alias localhost -certreq -file cert-file
  keytool -keystore client.keystore.jks -alias localhost -validity 365 -keyalg RSA -genkey
  keytool -keystore client.keystore.jks -alias localhost -certreq -file ofo-cert-file
  openssl x509 -req -CA ca-cert -CAkey ca-key -in ofo-cert-file -out ofo-cert-signed -days 1024 -CAcreateserial -passin pass:*******
  keytool -keystore client.keystore.jks -alias CARoot -import -file ca-cert
  keytool -keystore client.keystore.jks -alias CARoot -import -file ofo-cert-signed
  keytool -keystore client.keystore.jks -alias localhost -import -file ofo-cert-signed
  

• 配置PLAIN_SSL

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

cat >> /app/kafka/config/kafka_server_jaas.conf <<EOF
KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="*******"
    user_admin="*******"
    user_ofo="******";
};
EOF
cat >> /app/kafka/config/kafka_client_jaas.conf <<EOF
KafkaClient {
     org.apache.kafka.common.security.plain.PlainLoginModule required
     username="admin"
     password="*********";
 };
 EOF
 
cat >>  /app/kafka/config/server.properties <<EOF
ssl.keystore.location=/app/kafka/config/server.keystore.jks
ssl.keystore.password=********
ssl.key.password=*******
ssl.truststore.location=/app/kafka/config/server.truststore.jks
ssl.truststore.password=********
ssl.client.auth=required
ssl.keystore.type=JKS
ssl.truststore.type=JKS
listeners=SASL_SSL://172.17.1.151:9093
#inter.broker.listener.name=SASL_SSL
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
security.inter.broker.protocol=SASL_SSL
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
allow.everyone.if.no.acl.found=true
super.users=User:admin
EOF

服务启停

1. 启动
   /app/kafka/bin/kafka-server-start.sh -daemon /app/kafka/config/server.properties

2. 停止
   /app/kafka/bin/kafka-server-stop.sh

添加用户

vi /app/kafka/config/kafka_server_jaas.conf
在user_admin下面添加一行
user_newusername="yourpassword"

权限管理

/app/kafka/bin/kafka-acls.sh --help

客户端配置

1. 添加jvm参数
   -Djava.security.auth.login.config=/app/kafka/conf/kafka_client_jaas.conf
2. 配置kafka连接,添加属性：

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

   security.protocol=SSL sasl.mechanism=PLAIN ssl.truststore.location=/app/kafka/config/client.truststore.jks ssl.truststore.password=********** ssl.keystore.location=/app/kafka/config/client.keystore.jks ssl.keystore.password=********** ssl.key.password=*********** security.protocol=SASL_SSL sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="admin" \ password="*******";

例子：

1
2
3

export KAFKA_OPTS='-Djava.security.auth.login.config=/app/kafka/conf/kafka_client_jaas.conf'
/app/kafka/bin/kafka-console-consumer.sh --bootstrap-server 172.17.1.151:9093,172.17.1.152:9093,172.17.1.153:9093 --topic test_topic --from-beginning --consumer.config /app/kafka/config/consumer.properties
/app/kafka//bin/kafka-console-producer.sh --broker-list 172.17.1.151:9093,172.17.1.152:9093,172.17.1.153:9093 --topic test_topic --producer.config /app/kafka/config/producer.properties