kafka 安装以及 Kraft 模式、安全认证配置

这篇具有很好参考价值的文章主要介绍了kafka 安装以及 Kraft 模式、安全认证配置。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

常见安装：zookeeper + kafka

zookeeper 单节点安装：

apach官网下载对应包：apache-zookeeper-3.7.1-bin.tar.gz

修改对应配置文件 /conf/zoo_sample.cfg ，配置端口以及数据目录

sh zkServer.sh start 启动、sh zkServer.sh stop 停止、sh zkServer.sh status 状态

sh zkCli.sh -server 客户端

zookeeper 集群安装：

在每个节点data 目录，创建一个 myid 的文件，myid内容为服务id，没个节点服务id不同

配置文件zoo.cfg添加集群信息：server.myid=ip:2888:3888

2888为服务器 follower 与集群leader 的通信端口、3888 leader挂掉后，选举用端口

各节点启动集群.

分布式框架提供协调服务，为kafak 管理员，需要 zookeeper 选择机制、节点监听等

集群管理、leader选举

Topic注册

Broker注册

zookeeper 存储 kafka 的 borker信息，topic等信息

kafka 集群安装：

https://kafka.apache.org/ 官网下载对应运行体，eg：kafka_2.12-3.0.0.tgz

config/server.properties 修改对应服务配置

broker.id 服务唯一id

log.dirs=/tmp/kafka-logs 数据目录

zookeeper.connect=192.168.2.207:2181,192.168.2.208:2181/kafka 集群地址配置

各节点启动： sh kafka-server-start.sh -daemon …/config/server.properties

kafka - Kraft 模式（去掉zookeeper）

版本大于 2.8.0 时，可以不依赖zookeeper 环境启动

修改 krafk 目录下 server.properties 配置文件，以下操作每个节点需要同步配置，node.id 为唯一

生成一个随机id： sh kafka-storage.sh random-uuid

初始化：sh kafka-storage.sh format -t ${id} -c /opt/kafka/kafka_2.12-3.0.0/config/kraft/server.properties
-Kraft 模式启动： sh ./bin/kafka-server-start.sh -daemon ./config/kraft/server.properties

Kraft 模式以及安全认证配置：

SASL/GSSAPI：kerberos认证方式，一般使用随机密码的keytab认证方式，密码是加密的，也是企业里使用最多的认证方式；

SASL/PLAIN：这种方式其实就是一个账号/密码的认证方式，不过它有很多缺陷，比如用户名密码是存储在文件中，不能动态添加，密码明文等等！好处是足够简单；

SASL/SCRAM：针对SASL/PLAIN方式的不足而提供的另一种认证方式。这种方式的用户名/密码是存储中zookeeper的，因此能够支持动态添加用户。该种认证方式还会使用sha256或sha512对密码加密，安全性相对会高一些，在0.10.2版本引入；

SASL/PLAIN 安全认证配置：

新建config 文件 config/kafka_server_jaas.conf

👇

KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
serviceName="kafka"
username="admin"
password="admin"
user_admin="admin";
};

copy一份kafka-server-start.sh ，修改名称 kafka-server-start-sasl.sh启动脚本修改名称，引入加密文件；注意路径

修改配置文件内容：

cp一份 server_sasl.properties 修改名称为server_sasl.properties 与非安全认证分开

# 修改以下内容
node.id=1
controller.quorum.voters=1@kraft1:9093
listeners=SASL_PLAINTEXT://192.168.2.207:9092,CONTROLLER://192.168.2.207:9093
inter.broker.listener.name=SASL_PLAINTEXT
advertised.listeners=SASL_PLAINTEXT://192.168.2.207:9092
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G -Djava.security.auth.login.config=/opt/kafka/kafka_2.12-3.0.0/config/kafka_server_jaas.conf"
fi

启动：sh ./bin/kafka-server-start-sasl.sh -daemon ./config/kraft/server_sasl.properties

spring boot yml配置

server:
  port: 8200

spring:
  mvc:
    pathmatch:
      matching-strategy: ant_path_matcher
  application:
    name: @artifactId@
  kafka:
    bootstrap-servers: 192.168.2.207:9092
    # 生产配置
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      properties:
        sasl.mechanism: PLAIN
        security.protocol: SASL_PLAINTEXT
        sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin";
    #消费配置
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      group-id: xiaoshu
      enable-auto-commit: false
      properties:
        sasl.mechanism: PLAIN
        security.protocol: SASL_PLAINTEXT
        sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin";

    listener:
      # 手动调用Acknowledgment.acknowledge()后立即提交
      ack-mode: manual