备案控制台
开发者社区 安全 文章 正文

基于SASL和ACL的Kafka安全性解析

2023-07-14 292
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
服务治理 MSE Sentinel/OpenSergo，Agent数量 不受限
简介： 本文主要介绍基于SCRAM进行身份验证，使用Kafka ACL进行授权，SSL进行加密以及使用camel-Kafka连接Kafka群集以使用camel路由生产和消费消息的过程。


       本文主要介绍基于SCRAM进行身份验证，使用Kafka ACL进行授权，SSL进行加密以及使用camel-Kafka连接Kafka群集以使用camel路由生产和消费消息的过程。

       加密与认证

       AMQ Streams支持加密和身份验证，这是侦听器配置的一部分。

       侦听器配置

       Kafka代理中的加密和身份验证是针对每个侦听器配置的。

       Kafka代理中的每个侦听器都配置有自己的安全协议。配置属性"listener.security.protocal"定义哪个侦听器使用哪种安全协议。然后将每个侦听器名称映射到其安全协议中。

       支持的安全协议有：

  • 纯文本：侦听器，无需任何加密或身份验证。
  • SSL协议：使用TLS加密的侦听器，以及使用TLS客户端证书的身份验证（可选）。
  • SASL_PLAINTEXT：侦听器不加密，但具有基于SASL的身份验证。
  • SASL_SSL：具有基于TLS的加密和基于SASL的身份验证的侦听器。

       以下为指定SASL_SSL协议侦听器配置：


listeners=SASL_SSL://localhost:9092
advertised.listeners=SASL_SSL://localhost:9092
security.inter.broker.protocol=SASL_SSL

       TLS加密

      为了使用TLS加密和服务器身份验证，必须提供包含私钥和公钥的密钥库。通常使用Java密钥存储（JKS）格式的文件来完成此操作。在"ssl.keystore.location"属性中设置此文件的路径。"ssl.keystore.password"为集群中的所有Kafka代理生成TLS证书。证书应在其通用名称或主题备用名称中具有其公告的地址和引导地址。

     编辑所有Kafka集群节点上的/opt/kafka/config/server.properties 配置文件，以进行以下操作：

     更改"listener.security.protocol.map"字段以为要使用TLS加密的侦听器指定SSL协议。

     将"ssl.keystore.location"选项设置为带有代理证书的JKS密钥库的路径。

     将"ssl.keystore.password"选项设置为用于保护密钥库的密码。

     SASL认证

     使用Java身份验证和授权服务（JAAS）配置SASL身份验证。JAAS还用于验证Kafka和ZooKeeper之间的连接。

     JAAS使用其自己的配置文件。该文件位于/opt/kafka/config/jaas.conf，通过普通的未加密连接以及通过TLS连接都支持SASL身份验证。可以分别为每个侦听器启用SASL。要启用它，listener.security.protocol.map中的安全协议必须是SASL_PLAINTEXT或SASL_SSL。

       Kafka中的SASL身份验证支持几种不同的机制：

  • 普通

      根据用户名和密码实施身份验证。用户名和密码以Kafka配置存储在本地。

  • SCRAM-SHA-256和SCRAM-SHA-512

     使用Salted Challenge Response身份验证机制（SCRAM）实现身份验证。SCRAM凭证集中存储在ZooKeeper中。SCRAM可以用于ZooKeeper群集节点在专用网络中隔离运行的情况。

  • GSSAPI

     针对Kerberos服务器实施身份验证

     通过JAAS配置文件配置SASL机制。Kafka使用名为Kafka服务器的JAAS上下文。在JAAS中配置它们之后，必须在Kafka配置中启用SASL机制。这可以使用sasl.enabled.mechanisms属性完成

  • SASL SCRAM

     Kafka中的SCRAM身份验证包含两种机制：SCRAM-SHA-256和SCRAM-SHA-512。这些机制仅在使用的哈希算法上有所不同-SHA-256与更强的SHA-512。要启用SCRAM身份验证，JAAS配置文件必须包含以下配置：


[administrator@JavaLangOutOfMemory ~ ]% vi ${kafka_home}/config/kafka_server_jass.conf
KafkaServer {
   org.apache.kafka.common.security.scram.ScramLoginModule required
   username="admin"
   password="admin123";
};

      并在server.properties文件中启用SASL身份验证:


[administrator@JavaLangOutOfMemory ~ ]% vi ${kafka_home}/config/server.properties
 # With SASL & SSL encryption
scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="admin" \
  password="admin@2016.08.19.com";

       在$kafka-home/config目录下创建ssl-user-config.properties文件，如下所示：


[administrator@JavaLangOutOfMemory ~ ]% vi ${kafka_home}/config/ssl-user-config.properties
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="luga" password="luga@2016.08.19.com";
ssl.truststore.location=$kafka_home/config/truststore/kafka.truststore.jks
ssl.truststore.password=luga@2016.08.19.com

       SCRAM机制的用户凭据存储在ZooKeeper中。 可借助kafka-configs.sh工具进行管理。


[administrator@JavaLangOutOfMemory ~ ]%/bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-512=[password='admin@2016.08.19.com']' --entity-type users --entity-name admin
Warning: --zookeeper is deprecated and will be removed in a future version of Kafka.
Use --bootstrap-server instead to specify a broker to connect to.
Completed updating config for entity: user-principal 'admin'.
./kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-512=[password='luga@2016.08.19.com']' --entity-type users --entity-name luga
Warning: --zookeeper is deprecated and will be removed in a future version of Kafka.
Use --bootstrap-server instead to specify a broker to connect to.
Completed updating config for entity: user-principal 'luga'.

       完整的$ {kafka-home} /config/server.properties文件如下所示:


# 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=SASL_SSL://localhost: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=SASL_SSL://localhost:9092
security.inter.broker.protocol=SASL_SSL
ssl.endpoint.identification.algorithm=
ssl.client.auth=required
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
sasl.enabled.mechanisms=SCRAM-SHA-512
# Broker security settings
46
ssl.truststore.location=/${kafka_home}/config/truststore/kafka.truststore.jks
ssl.truststore.password=luga@2016.08.19.com
ssl.keystore.location=/${kafka_home}/config/keystore/kafka.keystore.jks
ssl.keystore.password=luga@2016.08.19.com
ssl.key.password=luga@2016.08.19.com
# ACLs
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:admin
# #zookeeper SASL
zookeeper.set.acl=false
# # With SASL & SSL encryption
scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="admin" \
  password="admin123";
# 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=/tmp/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

       用JAAS启动Kafka，命令如下所示：


[administrator@JavaLangOutOfMemory ~ ]%export KAFKA_OPTS=-Djava.security.auth.login.config=/${kafka_home}/config/kafka_server_jaas.conf
[administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-server-start.sh   /${kafka_home}/config/server.properties

       创建主题Demo-Topic:


[administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:luga --operation Create --operation Describe  --topic demo-topic
Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`: 
    (principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW)
    (principal=User:luga, host=*, operation=CREATE, permissionType=ALLOW) 
Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`: 
    (principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW)
    (principal=User:luga, host=*, operation=CREATE, permissionType=ALLOW)

       同理，我们给生产者和消费者授予相关权限：


[administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:luga --producer --topic demo-topic
Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`: 
    (principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW)
    (principal=User:luga, host=*, operation=WRITE, permissionType=ALLOW)
    (principal=User:luga, host=*, operation=CREATE, permissionType=ALLOW) 
Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`: 
    (principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW)
    (principal=User:luga, host=*, operation=CREATE, permissionType=ALLOW)
    (principal=User:luga, host=*, operation=WRITE, permissionType=ALLOW)

[administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:luga --consumer --topic demo-topic --group demo-consumer-group
Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`: 
    (principal=User:luga, host=*, operation=READ, permissionType=ALLOW)
    (principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW) 
Adding ACLs for resource `ResourcePattern(resourceType=GROUP, name=demo-consumer-group, patternType=LITERAL)`: 
    (principal=User:luga, host=*, operation=READ, permissionType=ALLOW) 
Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`: 
    (principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW)
    (principal=User:luga, host=*, operation=CREATE, permissionType=ALLOW)
    (principal=User:luga, host=*, operation=WRITE, permissionType=ALLOW)
    (principal=User:luga, host=*, operation=READ, permissionType=ALLOW) 
Current ACLs for resource `ResourcePattern(resourceType=GROUP, name=demo-consumer-group, patternType=LITERAL)`: 
    (principal=User:luga, host=*, operation=READ, permissionType=ALLOW)

       开始生产消息：


[administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic demo-topic --producer.config config/ssl-producer.properties
>Hi this is docker
>

      生产者配置ssl-producer.properties：


[administrator@JavaLangOutOfMemory ~ ]% vi /${kafka_home}/config/ssl-producer.properties
bootstrap.servers=localhost:9092
compression.type=none
### SECURITY ######
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="luga" password="luga@2016.08.19.com";
ssl.truststore.location=/{kafka_home}/config/truststore/kafka.truststore.jks
ssl.truststore.password=luga@2016.08.19.com

        消费者配置ssl-consumer.properties：


[administrator@JavaLangOutOfMemory ~ ]% vi /${kafka_home}/config/ssl-consumer.properties
bootstrap.servers=localhost:9092
# consumer group id
group.id=test-consumer-group
# What to do when there is no initial offset in Kafka or if the current
# offset does not exist any more on the server: latest, earliest, none
#auto.offset.reset=
### SECURITY ######
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="luga" password="luga@2016.08.19.com";
ssl.truststore.location=/${kafka_home}/config/truststore/kafka.truststore.jks
ssl.truststore.password=luga@2016.08.19.com

       开始消费消息：


[administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic demo-topic --from-beginning --consumer.config config/ssl-consumer.properties

       现在基于came路由启动Spring Boot应用程序进行消息的生产与消费：


public class KafkaRouteBuilder extends RouteBuilder {
    @Override
    public void configure() throws Exception {
        //producer
         from("timer://foo?period={{period}}")
         .setBody(constant("Hi This is kafka example")) 
         .to("kafka:{{kafka.topic}}?brokers={{kafka.bootstrap.url}}"
                 + "&keySerializerClass=org.apache.kafka.common.serialization.StringSerializer"
                 + "&serializerClass=org.apache.kafka.common.serialization.StringSerializer" 
                 + "&securityProtocol={{security.protocol}}&saslJaasConfig={{sasl.jaas.config}}"
                 + "&saslMechanism={{sasl.mechanism}}&sslTruststoreLocation={{ssl.truststore.location}}"
                 + "&sslTruststorePassword={{ssl.truststore.password}}&sslTruststoreType={{ssl.truststore.type}}")
         .log("${body}");
         //consumer
        from("kafka:{{consumer.topic}}?brokers={{kafka.bootstrap.url}}&maxPollRecords={{consumer.max.poll.records}}"
                 + "&keySerializerClass=org.apache.kafka.common.serialization.StringSerializer"
                + "&serializerClass=org.apache.kafka.common.serialization.StringSerializer" 
                + "&groupId={{consumer.group}}&securityProtocol={{security.protocol}}&saslJaasConfig={{sasl.jaas.config}}"
                + "&saslMechanism={{sasl.mechanism}}&sslTruststoreLocation={{ssl.truststore.location}}"
                 + "&sslTruststorePassword={{ssl.truststore.password}}&sslTruststoreType={{ssl.truststore.type}}"
               + "&autoOffsetReset={{consumer.auto.offset.reset}}&autoCommitEnable={{consumer.auto.commit.enable}}")
         .log("${body}");
    }

       application.yml文件配置如下：


kafka.topic=demo-topic
kafka.bootstrap.url=localhost:9092
period=10000&repeatCount=5
kafka.key.serializer=org.apache.kafka.common.serialization.StringSerializer
kafka.value.serializer=org.apache.kafka.common.serialization.StringSerializer
consumer.topic=demo-topic
consumer.group=demo-consumer-group
consumer.max.poll.records=1
consumer.threads=10
consumer.consumersCount=1 
consumer.auto.offset.reset=earliest
consumer.auto.commit.enable=true
consumer.receive.buffer.bytes=-1 
security.protocol = SASL_SSL
ssl.truststore.location =/luga/git/{...}/src/main/truststore/kafka.truststore.jks
ssl.truststore.password =luga@2016.08.19.com
ssl.truststore.type = JKS
sasl.mechanism = SCRAM-SHA-512
#sasl.kerberos.service.name=
sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required username="luga" password="luga@2016.08.19.com";

       测试运行：


[administrator@JavaLangOutOfMemory ~ ]% mvn spring-boot:run
[INFO] Scanning for projects...
[INFO] 
[INFO] ---------< org.apache.camel.example:camel-example-kafka-sasl >----------
[INFO] Building Camel :: Example :: Kafka :: sasl 1.0
[INFO] --------------------------------[ jar ]---------------------------------
。。。

       至此，简单的场景运行起来。

文章标签：
云消息队列 Kafka 版
云解析DNS
密钥管理服务
微服务引擎
Java
消息中间件
网络安全
数据安全/隐私保护
Kafka
Spring
安全
算法
存储
关键词：
云解析DNS安全性
安全性云解析DNS
云消息队列 Kafka 版解析
云解析DNS kafka
云消息队列 Kafka 版sasl
架构驿站
目录
相关文章
一只牛博
|
11小时前
|
消息中间件 安全 Kafka
深度解析Kafka中消费者的奥秘
深度解析Kafka中消费者的奥秘
一只牛博
45 0
一只牛博
|
11小时前
|
消息中间件 存储 算法
深度解析Kafka中的消息奥秘
深度解析Kafka中的消息奥秘
一只牛博
42 0
程序三两行
|
12小时前
|
消息中间件 运维 Kafka
kafka使用SASL认证
kafka使用SASL认证
程序三两行
22 0
德迅云安全陈琦琦
|
12小时前
|
SQL 监控 安全
服务器安全性漏洞和常见攻击方式解析
服务器安全性漏洞和常见攻击方式解析
德迅云安全陈琦琦
91 0
倪虹升
|
12小时前
|
消息中间件 缓存 Kafka
探究Kafka原理-5.Kafka设计原理和生产者原理解析（下）
探究Kafka原理-5.Kafka设计原理和生产者原理解析
倪虹升
43 0
倪虹升
|
12小时前
|
消息中间件 存储 负载均衡
探究Kafka原理-5.Kafka设计原理和生产者原理解析（上）
探究Kafka原理-5.Kafka设计原理和生产者原理解析
倪虹升
64 0
倪虹升
|
12小时前
|
消息中间件 分布式计算 Java
探究Kafka原理-3.生产者消费者API原理解析（上）
探究Kafka原理-3.生产者消费者API原理解析
倪虹升
40 0
长梦
|
11小时前
|
消息中间件 监控 大数据
Kafka消息队列架构与应用场景探讨：面试经验与必备知识点解析
【4月更文挑战第9天】本文详尽探讨了Kafka的消息队列架构，包括Broker、Producer、Consumer、Topic和Partition等核心概念，以及消息生产和消费流程。此外，还介绍了Kafka在微服务、实时数据处理、数据管道和数据仓库等场景的应用。针对面试，文章解析了Kafka与传统消息队列的区别、实际项目挑战及解决方案，并展望了Kafka的未来发展趋势。附带Java Producer和Consumer的代码示例，帮助读者巩固技术理解，为面试做好准备。
长梦
28 0
东北赵四
|
12小时前
|
安全 容器
并发编程之的ArrayList安全性的详细解析
并发编程之的ArrayList安全性的详细解析
东北赵四
13 0
极客李华
|
12小时前
|
消息中间件 供应链 Java
Kafka的发布-订阅功能: Java实现与应用场景解析
Kafka的发布-订阅功能: Java实现与应用场景解析
极客李华
77 0

热门文章

最新文章

  • 1
    SpringMVC注解@RequestParam全面解析
  • 2
    dnsmasq搭建简易DNS服务器
  • 3
    Linux内存工具解析之RSS/VSS/USS/PSS区别于联系
  • 4
    dns组件能够删除后恢复
  • 5
    歪理邪说解析架构设计师上午考试试题之六（分析2010下半年系统架构设计师上午试题26-30题）
  • 6
    深入解析 Java @RequestParam 注解：处理请求参数的高效利用
  • 7
    2.3.1 理解动态代理 -《SSM深入解析与项目实战》
  • 8
    U1 NB-IoT开发板上云例程操作解析
  • 9
    DNS view
  • 10
    《 自动化测试最佳实践：来自全球的经典自动化测试案例解析》一一2.6　管理自动化测试
  • 1
    SparkStreaming（SparkStreaming概述、入门、Kafka数据源、DStream转换、输出、关闭）
    56
  • 2
    Kafka【付诸实践 03】Offset Explorer Kafka 的终极 UI 工具安装+简单上手+关键特性测试（一篇学会使用 Offset Explorer）
    253
  • 3
    Kafka【付诸实践 02】消费者和消费者群组+创建消费者实例+提交偏移量（自动、手动）+监听分区再平衡+独立的消费者+消费者其他属性说明（实例源码粘贴可用）【一篇学会使用Kafka消费者】
    92
  • 4
    Kafka【付诸实践 01】生产者发送消息的过程描述及设计+创建生产者并发送消息（同步、异步）+自定义分区器+自定义序列化器+生产者其他属性说明（实例源码粘贴可用）【一篇学会使用Kafka生产者】
    259
  • 5
    Kafka【基础知识 02】集群+副本机制+数据请求+物理存储+数据存储设计（图片来源于网络）
    51
  • 6
    Kafka【基础知识 01】消息队列介绍+Kafka架构及核心概念（图片来源于网络）
    119
  • 7
    Kafka【问题记录 01】kill -9 导致 Kakfa 重启失败问题处理（doesn‘t match stored clusterId xxx in meta.properties）
    28
  • 8
    Flink CDC产品常见问题之用upsert的方式写入kafka失败如何解决
    70
  • 9
    Flink CDC产品常见问题之Flink CDC里从kafka消费的时候顺序混乱如何解决
    61
  • 10
    DataWorks常见问题之kafka数据导入datahub失败如何解决
    69

    • 相关课程

    更多
  • 云计算工程师解析与实战-网络专家篇（体验版）
  • 深入解析Docker容器化技术
  • 分布式消息系统 Kafka 快速入门
  • Java面试疑难点解析 - 面试技巧及语言基础
  • Java面试疑难点解析 - Java Web开发
  • Java面试疑难点解析 - 系统架构及项目设计

    • 相关电子书

    更多
  • 神龙云服务器产品及技术深度解析
  • 弹性创造价值：基于ECS的最佳性价比实践解析
  • 又快又稳：阿里云下一代虚拟交换机解析

    • 相关实验场景

    更多
  • 通过云拨测对指定服务器进行Ping/DNS监测

    • 推荐镜像

    更多
  • DNS
  • NTP
  • kali
    • 下一篇
    2024年阿里云免费云服务器及学生云服务器申请教程参考