docker-compose部署kafka、SASL模式(密码校验模式)

一.基础kafka部署

zookeeper，kafka，kafka-ui

docker-compose.yml

注意点：192.168.1.20 是宿主机的ip

version: "3"
services:zookeeper:image: wurstmeister/zookeepercontainer_name: zookeeperrestart: alwaysports:- 2181:2181environment:ZOOKEEPER_CLIENT_PORT: 2181kafka:image: wurstmeister/kafkarestart: alwayscontainer_name: kafkadepends_on:- zookeeperports:- 9092:9092environment:KAFKA_BROKER_ID: 0KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.1.20:9092KAFKA_ADVERTISED_PORT: 9092KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092KAFKA_PORT: 9092 KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "false"KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0KAFKA_HEAP_OPTS: "-Xmx512M -Xmx512M"kafka-ui:image: provectuslabs/kafka-ui:latestcontainer_name: kafka-uirestart: alwaysports:- 10010:8080environment:- DYNAMIC_CONFIG_ENABLED=true- SERVER_SERVLET_CONTEXT_PATH=/ui-kafka- KAFKA_CLUSTERS_0_NAME=local- KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=kafka:9092- KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL=PLAINTEXTdepends_on:- zookeeper- kafka

kafka-ui

地址：http://localhost:10010/ui-kafka/

java生产者

        <dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.9.0</version></dependency>

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties;public class KafkaProducerTest {public static void main(String[] args) {Properties properties = new Properties();properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");//	KEY: 是kafka用于做消息投递计算具体投递到对应的主题的哪一个partition而需要的properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//	VALUE: 实际发送消息的内容properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//	2.创建kafka生产者对象 传递properties属性参数集合KafkaProducer<String, String> producer = new KafkaProducer<>(properties);for (int i = 0; i < 100; i++) {ProducerRecord<String, String> record = new ProducerRecord<String, String>("topic_test", i + " : testx123测试");//	4.发送消息producer.send(record);System.out.println("发送成功： " + i);}//	5.关闭生产者producer.close();}
}

java消费者

public class KafkaConsumerTest {public static void main(String[] args) {Properties props = new Properties();// bootstrap.servers：kafka服务器地址，多个用逗号隔开props.put("bootstrap.servers", "127.0.0.1:9092");props.put("group.id", "topic-test-group"); // 消费组groupIdprops.put("auto.offset.reset", "earliest");// 序列化方式props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);consumer.subscribe(Collections.singletonList("topic_test")); // 订阅的topicwhile (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000L));for (ConsumerRecord<String, String> record : records) {System.out.printf("主题 = %s, 分区 = %d, 位移 = %d, " + "消息键 = %s, 消息值 = %s\n",record.topic(), record.partition(), record.offset(), record.key(), record.value());}if (!records.isEmpty()) {try {// 提交消费位移consumer.commitSync();} catch (CommitFailedException exception) {System.out.println("commit failed....");}}}}
}

二.SASL模式部署kafka

解释： SASL（Simple Authentication and Security Layer）的配置

目录结构如下：C:/docker/kafka1/

server_jaas.conf配置

 需要新建文件：server_jaas.conf，前两个是zk配置，后两个是kafka配置

Client {org.apache.zookeeper.server.auth.DigestLoginModule requiredusername="admin"password="123456";
};Server {org.apache.zookeeper.server.auth.DigestLoginModule requiredusername="admin"password="123456"user_super="123456"user_admin="123456";
};KafkaServer {org.apache.kafka.common.security.plain.PlainLoginModule requiredusername="admin"password="123456"  user_admin="123456";
};KafkaClient {org.apache.kafka.common.security.plain.PlainLoginModule requiredusername="admin"password="123456";
};

zoo.cfg配置

其他的没改变，就是最后添加的四行

# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial 
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between 
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just 
# example sakes.
dataDir=/opt/zookeeper-3.4.13/data
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# Be sure to read the maintenance section of the 
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature
autopurge.purgeInterval=1## 开启SASl关键配置
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000
zookeeper.sasl.client=true

 docker-compose.yml

# 版本根据你的docker版本来的，目前主流应该都是3.几的版本
version: '3.8'
services:zookeeper:image: wurstmeister/zookeepervolumes:- C:/docker/kafka1/secrets/:/opt/secrets/- C:/docker/kafka1/zookeeper/zoo.cfg:/opt/zookeeper-3.4.13/conf/zoo.cfgcontainer_name: zookeeperenvironment:ZOOKEEPER_CLIENT_PORT: 2181ZOOKEEPER_TICK_TIME: 2000SERVER_JVMFLAGS: -Djava.security.auth.login.config=/opt/secrets/server_jaas.confports:- 2181:2181restart: alwayskafka:image: wurstmeister/kafkacontainer_name: kafkadepends_on:- zookeeperports: - 9092:9092volumes:- C:/docker/kafka1/secrets/:/opt/secrets/environment:KAFKA_BROKER_ID: 0KAFKA_ADVERTISED_LISTENERS: SASL_PLAINTEXT://192.168.1.20:9092KAFKA_ADVERTISED_PORT: 9092 KAFKA_LISTENERS: SASL_PLAINTEXT://0.0.0.0:9092KAFKA_SECURITY_INTER_BROKER_PROTOCOL: SASL_PLAINTEXTKAFKA_PORT: 9092 KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: PLAINKAFKA_SASL_ENABLED_MECHANISMS: PLAINKAFKA_AUTHORIZER_CLASS_NAME: kafka.security.auth.SimpleAclAuthorizerKAFKA_SUPER_USERS: User:adminKAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true" #设置为true，ACL机制为黑名单机制，只有黑名单中的用户无法访问，默认为false，ACL机制为白名单机制，只有白名单中的用户可以访问KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0KAFKA_HEAP_OPTS: "-Xmx512M -Xms16M"KAFKA_OPTS: -Djava.security.auth.login.config=/opt/secrets/server_jaas.confkafka-ui:image: provectuslabs/kafka-ui:latestcontainer_name: kafka-uirestart: alwaysports:- 10010:8080environment:- DYNAMIC_CONFIG_ENABLED=true- SERVER_SERVLET_CONTEXT_PATH=/kafka-ui- KAFKA_CLUSTERS_0_NAME=local- KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=kafka:9092- KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL=SASL_PLAINTEXT- KAFKA_CLUSTERS_0_PROPERTIES_SASL_MECHANISM=PLAIN- KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="123456";depends_on:- zookeeper- kafka

这是一个Docker Compose文件，用于定义和运行多个Docker容器的应用。我会为你详细解释这个文件的内容：

  1. 版本:

    • version: '3.8': 指定了Docker Compose的版本为3.8。

  2. services:

    • 定义了两个服务，分别是zookeeper和kafka。

  3. zookeeper:

    • 使用wurstmeister/zookeeper镜像来创建容器。
    • 挂载了两个卷：一个是本地的C:/docker/kafka1/secrets/目录到容器内的/opt/secrets/目录，另一个是本地的C:/docker/kafka1/zookeeper/zoo.cfg文件到容器内的/opt/zookeeper-3.4.13/conf/zoo.cfg文件。
    • 设置容器的名称为zookeeper。
    • 设置环境变量：如ZOOKEEPER_CLIENT_PORT, ZOOKEEPER_TICK_TIME, SERVER_JVMFLAGS等。
    • 映射容器的2181端口到主机的2181端口。
    • 设置容器在退出后总是重启。

  4. kafka:

    • 使用wurstmeister/kafka镜像来创建容器。
    • 容器名称为kafka。
    • 这个服务依赖于zookeeper服务，这意味着它会等待zookeeper服务启动后再启动。
    • 映射容器的9092端口到主机的9092端口。
    • 挂载了与zookeeper相同的卷。
    • 设置环境变量，包括Kafka的配置如：广告监听器、端口、安全协议、SASL机制等。还定义了Kafka的超级用户为“admin”。
    • 设置Kafka的存储和堆大小。

  5. 总体解释:
    这是一个使用Docker Compose配置Kafka和Zookeeper集群的示例。Zookeeper作为Kafka的协调服务，负责管理集群的状态和配置。这个配置文件中，Zookeeper和Kafka都有详细的配置和环境变量设置，以适应特定的使用场景或安全需求。例如，SASL_PLAINTEXT是用于安全认证的协议，而SimpleAclAuthorizer则是一个简单的访问控制列表作者器，用于权限控制。

这是一个Docker Compose配置文件的一部分，用于设置Kafka UI服务。Kafka UI是一个用于管理和监视Apache Kafka集群的工具。以下是对该配置的详细解释：

  1. image:

    • 使用provectuslabs/kafka-ui:latest镜像来创建容器。这个镜像是专门为Kafka UI提供的，并保持最新。

  2. container_name:

    • 容器的名称为kafka-ui。

  3. restart:

    • 设置容器的重启策略为always，这意味着容器将始终在退出后自动重启。

  4. ports:

    • 将容器的8080端口映射到主机的10010端口，这样你就可以从主机访问Kafka UI。

  5. environment:

    • 这些是传递给容器的环境变量。它们用于配置Kafka UI的各种设置。
      • DYNAMIC_CONFIG_ENABLED=true: 启用动态配置。
      • SERVER_SERVLET_CONTEXT_PATH=/kafka-ui: 设置服务器的Servlet上下文路径。
      • KAFKA_CLUSTERS_0_NAME=local: 第一个Kafka集群的名称为“local”。
      • KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=kafka:9092: 第一个Kafka集群的引导服务器地址是kafka:9092。
      • KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL=SASL_PLAINTEXT: 安全协议设置为SASL_PLAINTEXT。
      • KAFKA_CLUSTERS_0_PROPERTIES_SASL_MECHANISM=PLAIN: SASL机制设置为PLAIN。
      • KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG=...: SASL JAAS配置，其中定义了用户名和密码。

  6. depends_on:

    • 这部分指定了该服务所依赖的其他服务。在这里，它依赖于zookeeper和kafka服务，这意味着Kafka UI服务将在这些服务启动后启动。

总之，这个配置是用于设置Kafka UI服务的Docker容器，该服务依赖于Zookeeper和Kafka集群，并使用特定的环境变量进行配置。

kafka-ui测试

http://localhost:10010/kafka-ui

java生产者

 public class KafkaProducerTest {public static void main(String[] args) {String user = "admin";String password = "123456";String topic = "test-topic";Properties properties = new Properties();//	1.配置生产者启动的关键属性参数properties.put("group.id", "test_group");properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//	VALUE: 实际发送消息的内容properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// ssl加密和认证properties.put("security.protocol", "SASL_PLAINTEXT");properties.put("sasl.mechanism", "PLAIN");properties.put("sasl.jaas.config","org.apache.kafka.common.security.scram.ScramLoginModule "+ "required username=\"" + user + "\" password=\"" + password + "\";");//	2.创建kafka生产者对象 传递properties属性参数集合KafkaProducer<String, String> producer = new KafkaProducer<>(properties);for (int i = 0; i < 1000; i++) {//	3.构造消息内容try {Thread.sleep(500);} catch (InterruptedException e) {throw new RuntimeException(e);}ProducerRecord<String, String> record = new ProducerRecord<String, String>("H0162", i + " : testxiaoxixixiix接口连接123");//	4.发送消息producer.send(record);System.out.println("发送成功： " + i);}//	5.关闭生产者producer.close();}
}

java消费者

 
public class KafkaConsumerTest {public static void main(String[] args) {String user = "admin";String password = "123456";String topic = "test-topic";Properties props = new Properties();// bootstrap.servers：kafka服务器地址，见中间服务平台”机构授权“功能操作说明 36.137.123.54:9092为示例地址props.put("bootstrap.servers", "127.0.0.1:9092");props.put("group.id", topic);props.put("auto.offset.reset", "earliest");// 序列化方式props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");// ssl加密和认证props.put("security.protocol", "SASL_PLAINTEXT");props.put("sasl.mechanism", "PLAIN");props.put("sasl.jaas.config","org.apache.kafka.common.security.scram.ScramLoginModule "+ "required username=\"" + user + "\" password=\"" + password + "\";");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);consumer.subscribe(Collections.singletonList(topic));while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000L));for (ConsumerRecord<String, String> record : records) {System.out.printf("主题 = %s, 分区 = %d, 位移 = %d, " + "消息键 = %s, 消息值 = %s\n",record.topic(), record.partition(), record.offset(), record.key(), record.value());}if (!records.isEmpty()) {try {// 提交消费位移consumer.commitSync();} catch (CommitFailedException exception) {System.out.println("commit failed....");}}}}
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/341663.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

H 指数，经典算法实战。

H 指数，经典算法实战。

&#x1f3c6;作者简介&#xff0c;普修罗双战士&#xff0c;一直追求不断学习和成长&#xff0c;在技术的道路上持续探索和实践。 &#x1f3c6;多年互联网行业从业经验&#xff0c;历任核心研发工程师&#xff0c;项目技术负责人。 &#x1f389;欢迎 &#x1f44d;点赞✍评论…
阅读更多...
GBASE南大通用SQL API 中的 SQL

GBASE南大通用SQL API 中的 SQL

ESQL 产品为GBASE南大通用数据库 GBase 8s SQL API&#xff08;应用程序编程接口&#xff09;。 GBase 为 C 编程语言产生 SQL API。 下图展示 SQL API 产品如何工作。您编写您在其中将 SQL 语句处理作为可执行代码的源 程序。嵌入式 SQL 预处理器处理您的源程序&#xff0c;它…
阅读更多...
绝地求生：【违规处罚工作公示】1月1日-1月7日

绝地求生：【违规处罚工作公示】1月1日-1月7日

1月1日至1月7日期间&#xff0c;共计对113,490个违规账号进行了封禁&#xff0c;其中103,133个账号因使用外挂被永久封禁。 若您游戏中遇到违规行为&#xff0c;建议您优先在游戏内进行举报&#xff1b; 另外您也可以在官方微信公众号【PUBG国际版】中点击“ 服务中心 - 举报…
阅读更多...
mysql原理--redo日志1

mysql原理--redo日志1

1.redo日志是个啥 我们知道 InnoDB 存储引擎是以页为单位来管理存储空间的&#xff0c;我们进行的增删改查操作其实本质上都是在访问页面&#xff08;包括读页面、写页面、创建新页面等操作&#xff09;。我们前边唠叨 Buffer Pool 的时候说过&#xff0c;在真正访问页面之前&a…
阅读更多...
Linux第26步_在虚拟机中安装stm32wrapper4dbg工具

Linux第26步_在虚拟机中安装stm32wrapper4dbg工具

在Ubuntu下编译TF-A 或者 Uboot时&#xff0c;我们需要用到ST公司提供的stm32wrapper4dbg工具。stm32wrapper4dbg工具的源码下载地址为: GitHub - STMicroelectronics/stm32wrapper4dbg 记得我们在前面已经创建过的目录如下&#xff1a; 1&#xff09;、在根目录下&#xf…
阅读更多...
大学物理-实验篇（二）——用分光计测定三棱镜的折射率（光：特定频段电磁波、光线在介质界面折射、平行光与凸透镜）

大学物理-实验篇（二）——用分光计测定三棱镜的折射率（光：特定频段电磁波、光线在介质界面折射、平行光与凸透镜）

目录 预备知识 光&#xff1a;特定频段电磁波 光线在介质界面折射 平行光与凸透镜 实验目的 实验仪器 实验原理 实验步骤 准备分光计 目镜调焦 刻度盘读数 测三棱镜顶角 测三棱镜最小偏向角 数据处理 预备知识 光&#xff1a;特定频段电磁波 光速&#xff1a;…
阅读更多...
2024.1.12每日一题

2024.1.12每日一题

LeetCode 2085.统计出现过一次的公共字符串 2085. 统计出现过一次的公共字符串 - 力扣&#xff08;LeetCode&#xff09; 题目描述 给你两个字符串数组 words1 和 words2 &#xff0c;请你返回在两个字符串数组中 都恰好出现一次 的字符串的数目。 示例 1&#xff1a; 输…
阅读更多...
Tomcat-快速使用

Tomcat-快速使用

关于Tomcat的概念及来由在Tomcat基本知识中进行了介绍&#xff0c;下面我们直接上手快速使用起来。 一、快速使用 &#xff08;1&#xff09;tomcat下载 &#xff08;2&#xff09;解压缩 &#xff08;3&#xff09;启动程序 &#xff08;4&#xff09;访问tomcat&#xff1a…
阅读更多...
PPT插件-大珩助手-《提取选中的幻灯片》-选中新建

PPT插件-大珩助手-《提取选中的幻灯片》-选中新建

选中新建 提取选中的幻灯片到新的幻灯文稿中。PDF编辑器可以提取指定的页面到新的PDF文档中&#xff0c;PPT没有这个功能&#xff0c;因此开发。 软件介绍 PPT大珩助手是一款全新设计的Office PPT插件&#xff0c;它是一款功能强大且实用的PPT辅助工具&#xff0c;支持Wps Wo…
阅读更多...
使用推测解码 (Speculative Decoding) 使 Whisper 实现 2 倍的推理加速

使用推测解码 (Speculative Decoding) 使 Whisper 实现 2 倍的推理加速

Open AI 推出的 Whisper 是一个通用语音转录模型&#xff0c;在各种基准和音频条件下都取得了非常棒的结果。最新的 large-v3 模型登顶了 OpenASR 排行榜&#xff0c;被评为最佳的开源英语语音转录模型。该模型在 Common Voice 15 数据集的 58 种语言中也展现出了强大的多语言性…
阅读更多...
2024最新适用于 Windows 、Mac 的最佳屏幕录制软件

2024最新适用于 Windows 、Mac 的最佳屏幕录制软件

屏幕录制软件可以帮助我们录制 PC 和MacBook的实时屏幕视频。如果您想为 优酷录制视频&#xff0c;或者您正在为您的公司制作基于视频的项目&#xff0c;并且需要捕获屏幕的实时视频录制&#xff0c;那么我们在此列出了 一 款适合您的 Windows 、Mac的 2024 年最佳屏幕录制软件…
阅读更多...
【K8S 云原生】Kurbernets集群的调度策略

【K8S 云原生】Kurbernets集群的调度策略

目录 一、Kubernetes的list-watch机制 1、List-watch 2、创建pod的过程&#xff1a; 二、scheduler调度的过程和策略&#xff1a; 1、简介 2、预算策略&#xff1a;predicate 3、优先策略&#xff1a; 3.1、leastrequestedpriority&#xff1a; 3.2、balanceresourceal…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023