kafka 设置用户密码和通过SpringBoot测试

分类：Kafka作者：张童瑶

叙述

当前Kafka认证方式采用动态增加用户协议。

自0.9.0.0版本开始Kafka社区添加了许多功能用于提高Kafka群集的安全性，Kafka提供SSL或者SASL两种安全策略。SSL方式主要是通过CA令牌实现，此文主要介绍SASL方式。

1）SASL验证:

验证方式 Kafka版本 特点
SASL/PLAIN 0.10.0.0 不能动态增加用户
SASL/SCRAM 0.10.2.0 可以动态增加用户
SASL/Kerberos 0.9.0.0 需要独立部署验证服务
SASL/OAUTHBEARER 2.0.0 需自己实现接口实现token的创建和验证，需要额外Oauth服务

2）SSL加密: 使用SSL加密在代理和客户端之间，代理之间或代理和工具之间传输的数据。

操作

1、进入kafka目录，修改kafka配置文件，修改config/server.properties文件，并添加如下内容

vi config/server.properties

修改内容：

#配置zookeeper管理kafka的路径
zookeeper.connect=localhost:2181

#配置kafka的监听端口（SASL_PLAINTEXT：动态增加用户协议，PLAINTEXT 不能动态增加用户）
listeners=SASL_PLAINTEXT://:9092

#把kafka的地址端口注册给zookeeper，如果是远程访问要改成外网IP（SASL_PLAINTEXT：动态增加用户协议，PLAINTEXT 不能动态增加用户）
advertised.listeners=SASL_PLAINTEXT://你外网的ip:9092


#使用的认证协议（SASL_PLAINTEXT：动态增加用户协议，PLAINTEXT 不能动态增加用户）
security.inter.broker.protocol=SASL_PLAINTEXT

#SASL机制
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN

#完成身份验证的类
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

#如果没有找到ACL（访问控制列表）配置，则允许任何操作。
allow.everyone.if.no.acl.found=false

#需要开启设置超级管理员,设置visitor用户为超级管理员
super.users=User:visitor

全部内容：

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License,Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# see kafka.server.KafkaConfig for additional details and defaults

############################# Server Basics #############################

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=0

############################# Socket Server Settings #############################

# The address the socket server listens on. It will get the value returned from 
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
#listeners=PLAINTEXT://:9092

# Hostname and port the broker will advertise to producers and consumers. If not set,
# it uses the value for "listeners" if configured.  Otherwise,it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
#advertised.listeners=PLAINTEXT://your.host.name:9092

# Maps listener names to security protocols,the default is for them to be the same. See the config documentation for more details
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# The number of threads that the server uses for receiving requests from the network and sending responses to the network
num.network.threads=3

# The number of threads that the server uses for processing requests,which may include disk I/O
num.io.threads=8

# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400

# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400

# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600


############################# Log Basics #############################

# A comma separated list of directories under which to store log files
log.dirs=/usr/local/kafka/kafka-logs

# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption,but this will also result in more files across
# the brokers.
num.partitions=1

# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1

############################# Internal Topic Settings  #############################
# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
# For anything other than development testing,a value greater than 1 is recommended to ensure availability such as 3.
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

############################# Log Flush Policy #############################

# Messages are immediately written to the filesystem but by default we only fsync() to sync
# the OS cache lazily. The following configurations control the flush of data to disk.
# There are a few important trade-offs here:
#    1. Durability: Unflushed data may be lost if you are not using replication.
#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
#    3. Throughput: The flush is generally the most expensive operation,and a small flush interval may lead to excessive seeks.
# The settings below allow one to configure the flush policy to flush data after a period of time or
# every N messages (or both). This can be done globally and overridden on a per-topic basis.

# The number of messages to accept before forcing a flush of data to disk
#log.flush.interval.messages=10000

# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000

############################# Log Retention Policy #############################

# The following configurations control the disposal of log segments. The policy can
# be set to delete segments after a period of time,or after a given size has accumulated.
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
# from the end of the log.

# The minimum age of a log file to be eligible for deletion due to age
log.retention.hours=168

# A size-based retention policy for logs. Segments are pruned from the log unless the remaining
# segments drop below log.retention.bytes. Functions independently of log.retention.hours.
#log.retention.bytes=1073741824

# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824

# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000

############################# Zookeeper #############################

# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs,each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=localhost:2181

# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=18000


############################# Group Coordinator Settings #############################

# The following configuration specifies the time,in milliseconds,that the GroupCoordinator will delay the initial consumer rebalance.
# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group,up to a maximum of max.poll.interval.ms.
# The default value for this is 3 seconds.
# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
# However,in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary,and potentially expensive,rebalances during application startup.
group.initial.rebalance.delay.ms=0


#配置zookeeper管理kafka的路径
zookeeper.connect=localhost:2181

#配置kafka的监听端口（SASL_PLAINTEXT：动态增加用户协议，PLAINTEXT 不能动态增加用户）
listeners=SASL_PLAINTEXT://:9092

#把kafka的地址端口注册给zookeeper，如果是远程访问要改成外网IP（SASL_PLAINTEXT：动态增加用户协议，PLAINTEXT 不能动态增加用户）
advertised.listeners=SASL_PLAINTEXT://你外网的ip:9092


#使用的认证协议（SASL_PLAINTEXT：动态增加用户协议，PLAINTEXT 不能动态增加用户）
security.inter.broker.protocol=SASL_PLAINTEXT

#SASL机制
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN

#完成身份验证的类
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

#如果没有找到ACL（访问控制列表）配置，则允许任何操作。
allow.everyone.if.no.acl.found=false

#需要开启设置超级管理员,设置visitor用户为超级管理员
super.users=User:visitor

2、在kafka目录下新建pass目录，并且创建配置文件，填入如下内容。

创建目录

mkdir pass

创建 kafka_server_jaas.conf 文件

vi kafka_server_jaas.conf

填入如下内容

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
        username="visitor"
        password="qaz@123"
        user_visitor="qaz@123";
};

3、在当前kafka目录下，修改kafka启动脚本

vi bin/kafka-server-start.sh

在文件最上面添加变量

export KAFKA_OPTS=" -Djava.security.auth.login.config=/usr/local/kafka2.12/pass/kafka_server_jaas.conf"

全部配置内容

#!/bin/bash
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License,either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

export KAFKA_OPTS=" -Djava.security.auth.login.config=/usr/local/kafka2.12/pass/kafka_server_jaas.conf"

if [ $# -lt 1 ];
then
	echo "USAGE: $0 [-daemon] server.properties [--override property=value]*"
	exit 1
fi
base_dir=$(dirname $0)

if [ "x$KAFKA_LOG4J_OPTS" = "x" ]; then
    export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:$base_dir/../config/log4j.properties"
fi

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
fi

EXTRA_ARGS=${EXTRA_ARGS-'-name kafkaServer -loggc'}

COMMAND=$1
case $COMMAND in
  -daemon)
    EXTRA_ARGS="-daemon "$EXTRA_ARGS
    shift
    ;;
  *)
    ;;
esac

exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"

在这里插入图片描述

Kafka消费者提供者客户端验证

注意：如果你只需要使用Kafka连接SpringBoot验证，那么这一步可以不做！

进入kafka目录，在pass目录下创建kafka_client_jaas.conf文件，并写入如下内容。

KafkaClient {  
org.apache.kafka.common.security.plain.PlainLoginModule required  
    username="visitor"  
    password="qaz@123";  
};

提供者

1、配置提供者认证，修改提供者启动脚本，并在顶部写入下面内容。

vi bin/kafka-console-producer.sh

顶部写入下面内容

export KAFKA_OPTS=" -Djava.security.auth.login.config=/usr/local/kafka2.12/pass/kafka_client_jaas.conf"

在这里插入图片描述

2、启动提供者

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic testTopic --producer-property security.protocol=SASL_PLAINTEXT --producer-property sasl.mechanism=PLAIN

在这里插入图片描述

消费者

1、配置消费者认证，修改消费者启动脚本，并在顶部写入下面内容。

vi bin/kafka-console-consumer.sh

顶部写入下面内容

export KAFKA_OPTS=" -Djava.security.auth.login.config=/usr/local/kafka2.12/pass/kafka_client_jaas.conf"

在这里插入图片描述

2、启动消费者

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testTopic --from-beginning --consumer-property security.protocol=SASL_PLAINTEXT --consumer-property sasl.mechanism=PLAIN

在这里插入图片描述

SpringBoot中测试验证

1、在SpringBoot中测试验证账号密码，具体SpringBoot项目配置等请参考其他博文。

配置文件

#
#   环境选择器
#
#   @author:tongyao
#
spring:
  kafka:
    bootstrap-servers: 82.157.190.245:1803

    # 配置用户名密码
    properties:
      security.protocol: SASL_PLAINTEXT
      sasl.mechanism: PLAIN

      # SASL/PLAIN 不能动态增加用户
      #sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="visitor" password="qaz@123";

      # SASL/SCRAM 可以动态增加用户
      sasl.jaas.config: org.apache.kafka.common.security.scram.ScramLoginModule required username="visitor" password="qaz@123";

    producer:

      # 发生错误后，消息重发的次数。
      retries: 0

      #当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。
      batch-size: 16384

      # 设置生产者内存缓冲区的大小。
      buffer-memory: 33554432

      # 键的序列化方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer

      # 值的序列化方式
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

      # acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。
      # acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
      # acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。
      acks: 1
    consumer:

      # 自动提交的时间间隔 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5D
      auto-commit-interval: 1S

      # 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
      # latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
      # earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录
      auto-offset-reset: earliest

      # 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量
      enable-auto-commit: false

      # 键的反序列化方式
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer

      # 值的反序列化方式
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    listener:

      # 在侦听器容器中运行的线程数。
      concurrency: 5

      #listner负责ack，每调用一次，就立即commit
      ack-mode: manual_immediate
      missing-topics-fatal: false
server:
  port: 8899

在这里插入图片描述


文章参考：https://blog.csdn.net/guaotianxia/article/details/121094383

原文地址：https://blog.csdn.net/u014641168/article/details/130841314

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。

相关推荐

Kafka的认证
文章浏览阅读4.1k次。kafka认证_kafka认证
Kafka之常用参数配置整理
文章浏览阅读4.8k次，点赞4次，收藏11次。kafka常用参数_kafka配置
Kafka生产者发送消息的流程
文章浏览阅读1.4k次，点赞25次，收藏10次。Kafka 生产者发送消息的流程涉及多个步骤，从消息的创建到成功存储在 Kafka 集群中。_kafka发送消息流程
第3、4章 Kafka 生产者 和 消费者 ——向 Kafka 写入数据 和读取数据
文章浏览阅读854次，点赞22次，收藏24次。点对点模型：适用于一对一的消息传递，具有高可靠性。发布/订阅模型：适用于广播消息给多个消费者，实现消息的广播。主题模型：适用于根据消息的主题进行灵活的过滤和匹配，处理复杂的消息路由需求。
kafka消息队列的初步探索
文章浏览阅读1.5k次，点赞2次，收藏3次。kafka 自动配置在KafkaAutoConfiguration
【可视化工具】Kafka-Offset Explorer使用
文章浏览阅读1.3w次，点赞6次，收藏33次。Offset Explorer（以前称为Kafka Tool）是一个用于管理和使Apache Kafka ®集群的GUI应用程序。它提供了一个直观的UI，允许人们快速查看Kafka集群中的对象以及存储在集群主题中的消息。它包含面向开发人员和管理员的功能。二、环境信息系统环境：windows 10版本：2.2Kafka版本：Kafka2.0.0三、安装和使用3.1 下载Offset Explorer 和安装下载到本地的 .exe文件Next安装路径 ，Next。_offset explorer
【云原生进阶之PaaS中间件】第三章Kafka-4.3.2-broker网络模型
文章浏览阅读1.3k次，点赞12次，收藏19次。kafka broker 在启动的时候，会根据你配置的listeners 初始化它的网络组件，用来接收外界的请求，这个listeners你可能没配置过，它默认的配置是listeners=PLAINTEXT://:9092就是告诉kafka使用哪个协议，监听哪个端口，如果我们没有特殊的要求的话，使用它默认的配置就可以了，顶多是修改下端口这块。
Kafka: 详解、使用教程和示例
文章浏览阅读1.3k次，点赞2次，收藏2次。Kafka 是一个强大的分布式流处理平台，用于实时数据传输和处理。通过本文详细的介绍、使用教程和示例，你可以了解 Kafka 的核心概念、安装、创建 Topic、使用生产者和消费者，从而为构建现代分布式应用打下坚实的基础。无论是构建实时数据流平台、日志收集系统还是事件驱动架构，Kafka 都是一个可靠、高效的解决方案。_博客系统怎么使用kafka
【Spring从成神到升仙系列 四】从源码分析 Spring 事务的来龙去脉
文章浏览阅读3.5k次，点赞42次，收藏56次。对于Java开发者而言，关于 Spring ，我们一般当做黑盒来进行使用，不需要去打开这个黑盒。但随着目前程序员行业的发展，我们有必要打开这个黑盒，去探索其中的奥妙。本期 Spring 源码解析系列文章，将带你领略 Spring 源码的奥秘。本期源码文章吸收了之前 Kafka 源码文章的错误，将不再一行一行的带大家分析源码，我们将一些不重要的分当做黑盒处理，以便我们更快、更有效的阅读源码。废话不多说，发车！
Kafka篇——Kafka消费者端常见配置，涵盖自动手动提交offset、poll消息细节、健康状态检查、新消费组消费offset规则以及指定分区等技术点配置，全面无死角，一篇文章拿下！
文章浏览阅读1.1k次，点赞14次，收藏16次。一、自动提交offset1、概念Kafka中默认是自动提交offset。消费者在poll到消息后默认情况下，会自动向Broker的_consumer_offsets主题提交当前主题-分区消费的偏移量2、自动提交offset和手动提交offset流程图3、在Java中实现配置4、自动提交offset问题自动提交会丢消息。因为如果消费者还没有消费完poll下来的消息就自动提交了偏移量，那么此时消费者挂了，于是下一个消费者会从已经提交的offset的下一个位置开始消费消息。_kafka中自动提交offsets
分布式 - 消息队列Kafka：Kafka生产者架构和配置参数
文章浏览阅读1.6k次。如果生产者发送消息的速度超过发送到服务器的速度，则会导致生产者空间不足，这个时候KafkaProducer的send（）方法调用要么被阻塞，要么抛出异常，这个取决于参数max.block.ms的配置，此参数的默认值为60000，即60秒。在默认情况下，生产者发送的消息是未经压缩的。如果应用程序调用send()方法的速度超过生产者将消息发送给服务器的速度，那么生产者的缓冲空间可能会被耗尽，后续的send()方法调用会等待内存空间被释放，如果在max.block.ms之后还没有可用空间，就抛出异常。_kafka producer 参数
Kafka快速入门(最新版3.6.0)
文章浏览阅读2.9k次，点赞3次，收藏10次。kafka解决通信问题_kafka3.6
nginx+rsyslog+kafka+clickhouse+grafana 实现nginx 网关监控
文章浏览阅读1.5k次，点赞9次，收藏11次。上面都配置完了之后可以先验证下，保证数据最终到ck，如果有问题，需要再每个节点调试，比如先调试nginx->rsyslog ，可以先不配置kafka 输出，配置为console或者文件输出都可以，具体这里就不写了。这里做了一个类型转换，因为nginx，request-time 单位是s，我想最终呈现在grafana 中是ms，所以这里做了转换，当然grafana中也可以做。kafka 相关部署这里不做赘述，只要创建一个topic 就可以。
Kafka中的enable-auto-commit和auto-commit-interval配置
文章浏览阅读1.4k次，点赞22次，收藏16次。Kafka中的enable-auto-commit和auto-commit-interval配置_auto-commit-interval
thingsboard规则链调用外部API——kafka
文章浏览阅读742次。thingsboard规则链调用外部 kafka_thingsboard kafka
Kafka 简介
文章浏览阅读1.3k次，点赞18次，收藏22次。Kafka_简介
从kafka如何保证数据一致性看通常数据一致性设计
文章浏览阅读1.1k次，点赞16次，收藏14次。在数据库系统中有个概念叫事务，事务的作用是为了保证数据的一致性，意思是要么数据成功，要么数据失败，不存在数据操作了一半的情况，这就是数据的一致性。在很多系统或者组件中，很多场景都需要保证数据的一致性，有的是高度的一致性。特别是在交易系统等这样场景。有些组件的数据不一定需要高度保证数据的一致性，比如日志系统。本节从从kafka如何保证数据一致性看通常数据一致性设计。
元数据管理-Atlas的介绍和使用(集成Hive、Solr、Kafka、Kerberos)
文章浏览阅读1.4k次。概述介绍架构发展架构原理类型系统介绍类型hive_table类型介绍DataSet类型定义Asset类型定义Referenceable类型定义Process类型定义Entities(实体)Attributes(属性)安装安装环境准备安装Solr-7.7.3安装Atlas2.1.0Atlas配置Atlas集成HbaseAtlas集成SolrAtlas集成KafkaAtlas Server配置Kerberos相关配置Atlas集成HiveAtlas启动Atlas使用Hive元数据初次导入Hive元数据增量同步。_atlas元数据管理
zookeeper + kafka
文章浏览阅读659次。Zookeeper是一个开源的分布式服务管理框架。存储业务服务节点元数据及状态信息，并负责通知再 ZooKeeper 上注册的服务几点状态给客户端。
Kafka-Kraft 模式架构部署
文章浏览阅读1.4k次。Kafka-Kraft 模式架构部署_kafka kraft部署
pythonJavaScriptjavaHTMLPHPreactjsC#AndroidCSSNode.jssqlrpython-3.xMysqLjQueryc++pandasFlutterangularIOSdjangolinuxswifttypescript路由器JSON路由器设置无线路由器h3c华三华三路由器设置华三路由器电脑软件教程arraysdocker软件图文教程Cvue.jslaravelspring-boot