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:

    • 定义了两个服务,分别是zookeeperkafka
  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:

    • 这部分指定了该服务所依赖的其他服务。在这里,它依赖于zookeeperkafka服务,这意味着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.mzph.cn/news/621619.shtml

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

相关文章

Java判断字符串当中是否有中文符号(不是中文名称,是符号)

public static void main(String[] args) throws ParseException, IOException, URISyntaxException {// 测试示例String testString1 "Hello,test&#xff01;";String testString2 "This is a test.";boolean result1 containsChineseSymbols(testStr…

第一次开发基于SpringBoot的Java应用

第一次开发基于SpringBoot的Java应用 一、 方式1&#xff1a;IDEA创建New Project Spring Boot官方文档的Getting Started1、IDEA创建New Project2、Spring Boot官方文档的Getting Started2.1 Creating the POM &#xff08;实际是&#xff0c;更新pom.xml&#xff09;2.2 Add…

【GitHub项目推荐--学算法】【转载】

不想做低级码农&#xff0c;不想成为前端抠图达人或是后台「增删改查」小王子&#xff1f;那你可能需要好好复习下算法与数据结构。 想成为算法工程师&#xff0c;基础知识是绕不开的大山。这次要推荐的项目是数据结构与算法的开源项目集&#xff0c;覆盖多种主流语言&#xff…

呼吸道病毒感染后,为何会引发细菌性肺炎?气道和肠道微生物组改变是关键

谷禾健康 病毒-细菌合并或继发感染 引起呼吸道感染的病毒是导致全世界高发病率和死亡率的原因&#xff0c;数十年来通常发生在冬季。在冬天&#xff0c;空气干燥&#xff0c;那些可能含有病毒的飞沫可以在空气中停留更长时间&#xff0c;并可以进一步传播。此外人的免疫力在冬季…

2023年终总结,一路向阳待花期

回望2023&#xff0c;可谓“苦尽甘来终有时&#xff0c;一路向阳待花期”。这一年&#xff0c;经历很多&#xff0c;收获亦很多。 回望2023 2023年最重要的三件事&#xff0c;想聊聊买房、工作、自我提升。 买房&#xff1a; 众所众知&#xff0c;2023楼市整体的情况不甚乐…

计算机三级(网络技术)——综合题(Sniffer抓包分析)

考点内容&#xff1a; DNS域名解析TCP三次握手FTP(文件传输协议)ICMP(Internet控制报文协议)&#xff1a;ping、tracertHTTP(超文本传输协议)&#xff1a;get、post命令 例题一 下图是校园网某台主机在命令行模式下执行某个命令时用sniffer捕获的数据包。 抓包分析 5~8行为…

使用JGit拉取代码提示未授权not authorized

原因&#xff1a;2021年8月13日后不支持密码登录&#xff0c;需要使用token验证 调用时候需要先去git仓库创建个人令牌 需要在安全中心创建个人token&#xff0c;使用token名称作为账号&#xff0c;使用token作为密码。 另&#xff1a; Github克隆仓库的三种方式对比&#xff…

x-cmd pkg | qrencode - 二维码生成工具

目录 简介首次用户功能特点竞品和相关作品进一步阅读 简介 qrencode 是一个用于生成二维码的命令行工具。它可以将文本、URL、电话号码等信息转换为二维码图像。生成的二维码图像可以保存为图片文件&#xff0c;方便在电子文档、网页、移动应用等各种场景中使用。 它支持的二维…

Redis集群(主从复制)

主从复制&#xff1a;是指将一台 Redis 服务器的数据&#xff0c;复制到其他的 Redis 服务器。 前者称为主节点(master)&#xff0c;后者称为从节点(slave),数据的复制是单向的&#xff0c;只能 由主节点到从节点。 可以实现数据备份。即使当其中一台机器宕机其他机器还可以正…

arcgis javascript api4.x加载天地图web墨卡托(wkid:3857)坐标系

效果&#xff1a; 示例代码&#xff1a; <!DOCTYPE html> <html lang"en"><head><meta charset"UTF-8"><meta name"viewport" content"widthdevice-width, initial-scale1.0"><meta http-equiv&quo…

一、MySQL 卸载

目录 1、软件的卸载准备 2、软件的卸载 方式一&#xff1a;通过控制面板卸载 方式二&#xff1a;通过mysql8的安装向导卸载 1、双击mysql8的安装向导 2、取消更新 3、选择要卸载的mysql服务器软件的具体版本 4、确认删除数据目录 5、执行删除 6、完成删除 3、清理残…

ssm基于web办事大厅政务预约系统+vue论文

摘 要 现代经济快节奏发展以及不断完善升级的信息化技术&#xff0c;让传统数据信息的管理升级为软件存储&#xff0c;归纳&#xff0c;集中处理数据信息的管理方式。本办事大厅政务预约系统就是在这样的大环境下诞生&#xff0c;其可以帮助管理者在短时间内处理完毕庞大的数据…

【GitHub项目推荐--6 个吊炸天的后台模板】【转载】

很多程序员都有过接私活的经历&#xff0c;帮别人开发一个网站&#xff1f;写个软件&#xff1f;不少网站都要有一个后台管理系统&#xff0c;而后台管理系统大多数情况下仅仅是管理员在使用&#xff0c;所以不像前台那样需要去定制设计优美的 UI。 一套既美观又方便的后台框架…

国家注册信息安全专业人员十五类CISP证书

国家注册信息安全专业人员&#xff08;Certified Information Security Professiona&#xff0c;简称CISP&#xff09;&#xff0c;是面向党政机关、关键信息基础设施运营单位、各类企事业单位和社会组织以及网络与信息安全企业、测评和咨询服务机构等工作的信息安全人员颁发的…

第 4 课 创建工作空间与功能包

文章目录 第 4 课 创建工作空间与功能包1.工作环境的创建2.ROS功能包的创建 第 4 课 创建工作空间与功能包 消息和服务的创建、发布器和订阅器的编写、服务端和客户端的编写都是基于Ros功能包进行操作的&#xff0c;因此在进行上述操作前&#xff0c;需要先创建工作空间及功能包…

【信息论安全】:信源编码定理

一. 介绍 在点对点的通信中&#xff0c;信源编码定理&#xff08;source coding theorem&#xff09;满足可达性和可逆性。当信道是无噪声时&#xff0c;那么YX&#xff0c;这时就不需要信道编码。但是&#xff0c;信源编码依旧是有效的&#xff0c;可以提高数据传输效率&…

iOS swift UISlider改变进度条的高度和圆形滑块的大小

文章目录 1.改变进度条的高度&#xff08;亲测有效&#xff09;2.改变圆形滑块的大小&#xff08;亲测有效&#xff09; 1.改变进度条的高度&#xff08;亲测有效&#xff09; import UIKitclass CustomSlider: UISlider {// 设置轨道高度var trackHeight: CGFloat 10// 重写…

Navicat 16 for MySQL:打造高效数据库开发管理工具

随着数据的快速增长和复杂性的提升&#xff0c;数据库成为了现代应用开发中不可或缺的一部分。而在MySQL数据库领域&#xff0c;Navicat 16 for MySQL作为一款强大的数据库开发管理工具&#xff0c;正受到越来越多开发者的青睐。 Navicat 16 for MySQL拥有丰富的功能和直观的界…

Jenkins-Maven Git

整合Maven 安装GIT #更新yum sudo yum update #安装git yum install git 安装Maven插件,在插件管理中心&#xff1a; 配置仓库 配置密码认证 我们可以在这个目录下看到Jenkins 帮我们拉取了代码 /env/liyong/data/docker/jenkins_mount/workspace/maven-job 配置maven打包…

启英泰伦推出「离线自然说」,离线语音交互随意说,不需记忆词条

离线语音识别是指不需要依赖网络&#xff0c;在本地设备实现语音识别的过程&#xff0c;通常以端侧AI语音芯片作为载体来进行数据的采集、计算和决策。但是语音芯片的存储空间有限&#xff0c;通过传统的语音算法技术&#xff0c;最多也只能存储数百条词条&#xff0c;导致用户…