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

这篇具有很好参考价值的文章主要介绍了kafka 设置用户密码和通过SpringBoot测试。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

叙述

当前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, 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.

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,spring boot,java

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"

kafka 账号密码,笔记记录,中间件开发,kafka,spring boot,java

2、启动提供者

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

kafka 账号密码,笔记记录,中间件开发,kafka,spring boot,java

消费者

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

vi bin/kafka-console-consumer.sh

顶部写入下面内容

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

kafka 账号密码,笔记记录,中间件开发,kafka,spring boot,java

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

kafka 账号密码,笔记记录,中间件开发,kafka,spring boot,java

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

kafka 账号密码,笔记记录,中间件开发,kafka,spring boot,java
文章参考:https://blog.csdn.net/guaotianxia/article/details/121094383文章来源地址https://www.toymoban.com/news/detail-584715.html

到了这里,关于kafka 设置用户密码和通过SpringBoot测试的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Springboot配置kafka用户名密码

    今天在写配置kafka用户名密码的yml文件时遇到IDE无法直接 . 出 sasl 的问题,过程中也尝试过切换kafka版本的方法,但是都不管用。最后的解决方案是直接配置就好,不用在意能不能 . 出来的问题。 以下为配置部分代码

    2024年02月12日
    浏览(39)
  • 通过docker安装es,并且设置安全权限(用户名和密码)_docker es 设置用户名和密码

    cluster.name: “docker-cluster” network.host: 0.0.0.0 xpack.security.enabled: true  xpack.security.transport.ssl.enabled: true 3、进入容器内 docker exec -it elasticsearch /bin/bash 4、cd /bin 5、可以进行设置密码(下面几步都要输入密码) 6、重新启动docker 7、浏览器访问es 自我介绍一下,小编13年上海交大毕业

    2024年04月22日
    浏览(55)
  • 通过设置向导设置dlink无线路由器包括登陆用户名和密码等等

    dlink路由器设置有设置向导,按照向导进行设置,很简单。 首先,你要知道路由器的IP地址,登陆用户名和密码,这些都在路由器的背面。 步骤 1、打开浏览器,在浏览器地址栏中输入你的路由器的地址,我的路由器的地址是192.168.0.1.所以我就输入192.168.0.1,点击回车 2、弹出输

    2024年02月06日
    浏览(58)
  • 通过docker安装es,并且设置安全权限(用户名和密码)

    1、首先拉去es 2、配置elasticsearch.yml(挂载到容器外) cluster.name: \\\"docker-cluster\\\" network.host: 0.0.0.0 xpack.security.enabled: true  xpack.security.transport.ssl.enabled: true 3、进入容器内 docker exec -it elasticsearch /bin/bash 4、cd /bin 5、可以进行设置密码(下面几步都要输入密码) 6、重新启动docker 7、

    2024年04月13日
    浏览(47)
  • Linux 用户账号管理-修改用户密码

    Linux 用户账号管理是运维人员必须要掌握的技能之一。其中,修改用户密码是比较常见的操作,本文将介绍在 Linux 系统中修改用户密码的方法,包括语法、实操和各种修改用户密码方法之间的区别。 在 Linux 中,修改用户密码的命令为 passwd ,其使用方法为: 其中,超级用户

    2024年02月14日
    浏览(40)
  • 如何配置Kafka账号密码

    我们需要与第三方系统进行数据同步,需要搭建公网Kafka,Kafka默认是没有用户密码校验的,所以我们需要配置用户名密码校验。 新增JAAS配置文件 在conf目录下新增kafka_server_jaas.conf文件,文件内容如下: 修改kafka-server-start.sh脚本 -Djava.security.auth.login.config=jaas配置文件的路径

    2024年01月23日
    浏览(42)
  • git页面管理用户账号和密码(新建用户和初始化密码)

    本文章适用:没有固定的用户认证(AD域或者LDAP等),git私有仓库单独管理用户(新建用户和初始化密码)。 1.创建新用户  以管理员的身份登录,进入git的管理模式,点击“New user”,创建新用户,填写相应的信息之后,拉到页面的后面 ,点击“Create user”新创建用户。此

    2024年02月11日
    浏览(50)
  • 通过账号、密码使用git拉代码

    1. git clone https://用户名:密码@gitee.com/use-items/lake-vue.git 例如: git clone https://username:password@gitee.com/use-items/lake-vue.git 2. 注意:账号密码中有特殊字符需要进行转义,字符-URL编码值对应如下: 3. 引用 https://blog.csdn.net/W_Zhulin/article/details/125058404

    2024年02月12日
    浏览(50)
  • MongoDB设置账号密码

    先输入 show dbs 查看mongodb数据库存在的数据库列表,选择需要设置的数据库-账号密码 设置超级管理员账号密码 1、 可以使用navicat15连接mongodb数据库,进入命令行界面,输入以下命令 如出现找不到MongoDB shell为可执行文件,选择设置路径,路径为:MongoDB shell是一个可执行的文

    2024年02月02日
    浏览(40)
  • git 设置全局账号密码

    设置全局仓库的用户名密码 git记住用户名以及密码 以上是全局,会在用户的主目录生成.gitconfig文件 之后在主目录,操作git pull,会提示输入账号密码,需要输一次后面不需要在输

    2024年02月11日
    浏览(52)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包