Kafka基于Kraft下的权限控制

你挺能闹_6726 2023-03-28 原文

Kafka基于Kraft下的权限控制

本文基于kafka的版本 3.2.0, 之前的版本无法使用本文所提到的方法。

本文方法对kafka源代码有修改
修改部分如下(metadata\src\main\java\org\apache\kafka\metadata\authorizer\StandardAuthorizerData.java):

    void addAcl(Uuid id, StandardAcl acl) {
        try {
            StandardAcl prevAcl = aclsById.putIfAbsent(id, acl);
            if (prevAcl != null) {
                log.warn("An ACL with ID " + id + " already exists.");
//                throw new RuntimeException("An ACL with ID " + id + " already exists.");
            }
            else if (!aclsByResource.add(acl)) {
                aclsById.remove(id);
                log.warn("Unable to add the ACL with ID " + id +" from aclsByResource");
                // throw new RuntimeException("Unable to add the ACL with ID " + id +
                //     " to aclsByResource");
            }
            else if (log.isTraceEnabled()) {
                log.trace("Added ACL " + id + ": " + acl);
            }
        } catch (Throwable e) {
            log.error("addAcl error", e);
 //           throw e;
        }
    }

    void removeAcl(Uuid id) {
        try {
            StandardAcl acl = aclsById.remove(id);
            if (acl == null) {
                log.warn("ID " + id + " not found in aclsById.");
//                throw new RuntimeException("ID " + id + " not found in aclsById.");
            }
            else if (!aclsByResource.remove(acl)) {
                log.warn("Unable to remove the ACL with ID " + id +" from aclsByResource");
               // throw new RuntimeException("Unable to remove the ACL with ID " + id +
                //    " from aclsByResource");
            }
            else if (log.isTraceEnabled()) {
                log.trace("Removed ACL " + id + ": " + acl);
            }
        } catch (Throwable e) {
            log.error("removeAcl error", e);
            //throw e;
        }
    }

实现作用是把抛出异常换为了输出警告，抛出异常的方式会导致kafka启动的时候无法正常启动，至于为什么kafka启动的时候要执行添加/删除 acl 的操作，暂时还不清楚。无法正常启动时出现的异常如下：

Jul 28 15:29:06 kafka-server-start.sh[123334]: [2022-07-28 15:29:06,133] ERROR [StandardAuthorizer 1] addAcl error (org.apache.kafka.metadata.authorizer.Stand
Jul 28 15:29:06 kafka-server-start.sh[123334]: java.lang.RuntimeException: An ACL with ID eK5n22NLQOeOHTT3gcnf7w already exists.
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.metadata.authorizer.StandardAuthorizerData.addAcl(StandardAuthorizerData.java:169)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.metadata.authorizer.StandardAuthorizer.addAcl(StandardAuthorizer.java:83)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$19(BrokerMetadataPublisher.scala:234)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at java.util.LinkedHashMap$LinkedEntrySet.forEach(LinkedHashMap.java:671)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$18(BrokerMetadataPublisher.scala:232)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$18$adapted(BrokerMetadataPublisher.scala:221)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at scala.Option.foreach(Option.scala:437)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.publish(BrokerMetadataPublisher.scala:221)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener.kafka$server$metadata$BrokerMetadataListener$$publish(BrokerMet
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.$anonfun$run$2(BrokerMetadataListener.scala:
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.$anonfun$run$2$adapted(BrokerMetadataListene
Jul 28 15:29:06 kafka-server-start.sh[123334]: at scala.Option.foreach(Option.scala:437)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.run(BrokerMetadataListener.scala:119)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventContext.run(KafkaEventQueue.java:121)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventHandler.handleEvents(KafkaEventQueue.java:200)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventHandler.run(KafkaEventQueue.java:173)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at java.lang.Thread.run(Thread.java:748)
Jul 28 15:29:06 kafka-server-start.sh[123334]: [2022-07-28 15:29:06,139] ERROR [BrokerMetadataPublisher id=1] Error publishing broker metadata at OffsetAndEpo
Jul 28 15:29:06 kafka-server-start.sh[123334]: java.lang.RuntimeException: An ACL with ID eK5n22NLQOeOHTT3gcnf7w already exists.
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.metadata.authorizer.StandardAuthorizerData.addAcl(StandardAuthorizerData.java:169)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.metadata.authorizer.StandardAuthorizer.addAcl(StandardAuthorizer.java:83)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$19(BrokerMetadataPublisher.scala:234)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at java.util.LinkedHashMap$LinkedEntrySet.forEach(LinkedHashMap.java:671)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$18(BrokerMetadataPublisher.scala:232)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$18$adapted(BrokerMetadataPublisher.scala:221)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at scala.Option.foreach(Option.scala:437)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.publish(BrokerMetadataPublisher.scala:221)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener.kafka$server$metadata$BrokerMetadataListener$$publish(BrokerMet
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.$anonfun$run$2(BrokerMetadataListener.scala:
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.$anonfun$run$2$adapted(BrokerMetadataListene
Jul 28 15:29:06 kafka-server-start.sh[123334]: at scala.Option.foreach(Option.scala:437)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.run(BrokerMetadataListener.scala:119)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventContext.run(KafkaEventQueue.java:121)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventHandler.handleEvents(KafkaEventQueue.java:200)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventHandler.run(KafkaEventQueue.java:173)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at java.lang.Thread.run(Thread.java:748)
Jul 28 15:29:06 kafka-server-start.sh[123334]: [2022-07-28 15:29:06,143] ERROR [BrokerMetadataListener id=1] Unexpected error handling HandleCommitsEvent (kaf
Jul 28 15:29:06 kafka-server-start.sh[123334]: java.lang.RuntimeException: An ACL with ID eK5n22NLQOeOHTT3gcnf7w already exists.
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.metadata.authorizer.StandardAuthorizerData.addAcl(StandardAuthorizerData.java:169)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.metadata.authorizer.StandardAuthorizer.addAcl(StandardAuthorizer.java:83)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$19(BrokerMetadataPublisher.scala:234)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at java.util.LinkedHashMap$LinkedEntrySet.forEach(LinkedHashMap.java:671)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$18(BrokerMetadataPublisher.scala:232)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$18$adapted(BrokerMetadataPublisher.scala:221)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at scala.Option.foreach(Option.scala:437)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.publish(BrokerMetadataPublisher.scala:221)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener.kafka$server$metadata$BrokerMetadataListener$$publish(BrokerMet
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.$anonfun$run$2(BrokerMetadataListener.scala:
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.$anonfun$run$2$adapted(BrokerMetadataListene
Jul 28 15:29:06 kafka-server-start.sh[123334]: at scala.Option.foreach(Option.scala:437)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.run(BrokerMetadataListener.scala:119)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventContext.run(KafkaEventQueue.java:121)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventHandler.handleEvents(KafkaEventQueue.java:200)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventHandler.run(KafkaEventQueue.java:173)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at java.lang.Thread.run(Thread.java:748)

安装

从官网上下载3.2.0的安装包 ,并解压
下载地址: https://www.apache.org/dyn/closer.cgi?path=/kafka/3.2.0/kafka_2.13-3.2.0.tgz

tar -xzf kafka_2.13-3.2.0.tgz
cd kafka_2.13-3.2.0

替换kafka-metadata-3.2.0.jar

基于上面提到的修改代码，重新构建后生成kafka-metadata-3.2.0.jar，替换掉libs/kafka-metadata-3.2.0.jar

# 备份官方的 kafka-metadata-3.2.0.jar
# 一定要把这个包从libs中拿出来
mv libs/kafka-metadata-3.2.0.jar ./
# 然后把自己build的jar包放进去
mv /root/kafka-3.2.0-src/metadata/build/libs/kafka-metadata-3.2.0.jar/kafka-metadata-3.2.0.jar libs/kafka-metadata-3.2.0.jar

修改配置文件

config/kraft/server.properties:

process.roles=broker,controller
node.id=1
# 修改这里，ip替换为实际ip
controller.quorum.voters=1@<ip1>:9093,2@<ip2>:9093,3@<ip4>:9093
# listeners 的PLAINTEXT要修改为SASL_PLAINTEXT
listeners=SASL_PLAINTEXT://<ip1>:9092,CONTROLLER://<ip1>:9093
# 这里也是PLAINTEXT要修改为SASL_PLAINTEXT
inter.broker.listener.name=SASL_PLAINTEXT
# 这里也是PLAINTEXT要修改为SASL_PLAINTEXT
advertised.listeners=SASL_PLAINTEXT://<ip1>:9092
controller.listener.names=CONTROLLER
# 这里 CONTROLLER:PLAINTEXT修改为 CONTROLLER:SASL_PLAINTEXT
listener.security.protocol.map=CONTROLLER:SASL_PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
# 这里，修改为要存放log的地方(实际存放的应该是kafka的数据，log在kafka安装目录的log文件夹下)
log.dirs=/data/kafka_3.2.0/log
num.partitions=1
num.recovery.threads.per.data.dir=2
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
# 认证方式，用了最简单的PLAIN，缺点是不能动态添加用户
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
sasl.mechanism=PLAIN
# 禁用了自动创建topic
auto.create.topics.enable = false
# 设置必须授权才能用
allow.everyone.if.no.acl.found=false
# 设置超级管理员
super.users=User:admin
# 这个是3.2.0版本新引入的认证方式，可以参考 https://cwiki.apache.org/confluence/display/KAFKA/KIP-801%3A+Implement+an+Authorizer+that+stores+metadata+in+__cluster_metadata
authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
# 集群间认证时用的认证方式
sasl.mechanism.controller.protocol=PLAIN

config/kraft/jaas.conf

KafkaServer {
   org.apache.kafka.common.security.plain.PlainLoginModule required
   username="admin"
   password="password"
   user_admin="password"
   user_test="test";
};

username/password 表示了认证时用的用户。
suer_admin="password",这个表示一个用户名为admin用户，密码是password，这个必须要有一个，且要这一个跟上面的username和password保持一致。
user_test="test" 是第二个用户，表示的是用户名为test的账户，密码为test。

service(/usr/lib/systemd/system/kafka.service)

默认kafka的启动方式是通过命令行管理，这里做了一个service用于控制kafka的启动与停止，也作为守护进程。

[Unit]
Description=kafka server daemon

[Service]
Type=simple
# 这里是指定了 jaas.conf文件，用于启用用户认证
Environment="KAFKA_OPTS=-Djava.security.auth.login.config=/data/kafka_3.2.0/package/kafka_2.13-3.2.0/config/kraft/jaas.conf"
# 启动命令
ExecStart=/data/kafka_3.2.0/package/kafka_2.13-3.2.0/bin/kafka-server-start.sh /data/kafka_3.2.0/package/kafka_2.13-3.2.0/config/kraft/server.properties
ExecReload=/bin/kill -HUP $MAINPID
# 停止命令
ExecStop=/data/kafka_3.2.0/package/kafka_2.13-3.2.0/bin/kafka-server-stop.sh
KillMode=process
Restart=on-failure
RestartSec=42s

[Install]
WantedBy=multi-user.target

生成集群clusterid

./bin/kafka-storage.sh random-uuid
./bin/kafka-storage.sh format -t <uuid> -c ./config/kraft/server.properties

启动kafka

systemctl daemon-reload
systemctl start kafka

命令行中使用

先创建一个用于client的认证文件

vim sasl.properties

# 配置上一个用户
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required  username="admin"  password="password";
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAI

执行命令式，后面都要带上 --command-config ./sasl.properties来进行用户认证

创建两个topic

# 创建 topic create-for-test 
bin/kafka-topics.sh --bootstrap-server localhost:9092  --create  --topic create-for-test --partitions 1 --replication-factor 1  --command-config ./sasl.properties
# 创建 topic admin-create-test
bin/kafka-topics.sh --bootstrap-server localhost:9092  --create  --topic admin-create-test --partitions 1 --replication-factor 1  --command-config ./sasl.properties
# 查看topic
bin/kafka-topics.sh --bootstrap-server localhost:9092 --list --command-config ./sasl.properties

为topic create-for-test ，用test赋读权限

bin/kafka-acls.sh  --bootstrap-server localhost:9092 --add --allow-principal User:test --operation Read --topic create-for-test --command-config ./sasl.properties

切换到test用户，查看topic

# 修改用户，把admin改成test
vim sasl.properties
# 查看所有topic，应该只能看到 create-for-test
bin/kafka-topics.sh --bootstrap-server localhost:9092 --list --command-config ./sasl.properties

java中使用

package org.example;



import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Properties;
import java.util.Collections;
import java.util.UUID;

/**
 * Hello world!
 *
 */
public class App 
{
    public static void main( String[] args )
    {
        String username = "test";
        String password = "test";
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "<ip1>:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 这里配置认证协议
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        // 认证方式
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        // 认证用户
        String saslJaasConfig = String.format("org.apache.kafka.common.security.plain.PlainLoginModule required \nusername=\"%s\" \npassword=\"%s\";", username, password);
        props.put(SaslConfigs.SASL_JAAS_CONFIG, saslJaasConfig);

        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        System.out.printf(consumer.listTopics().toString());
        consumer.close();
    }
}

有关Kafka基于Kraft下的权限控制的更多相关文章

Ruby Readline 在向上箭头上使控制台崩溃 - 2
当我在Rails控制台中按向上或向左箭头时，出现此错误:irb(main):001:0>/Users/me/.rvm/gems/ruby-2.0.0-p247/gems/rb-readline-0.4.2/lib/rbreadline.rb:4269:in`blockin_rl_dispatch_subseq':invalidbytesequenceinUTF-8(ArgumentError)我使用rvm来管理我的ruby安装。我正在使用=>ruby-2.0.0-p247[x86_64]我使用bundle来管理我的gem，并且我有rb-readline(0.4.2)(人们推荐的最少
ruby-on-rails - 带 Spring 锁的 Rails 4 控制台 - 2
我正在使用Ruby2.1.1和Rails4.1.0.rc1。当执行railsc时，它被锁定了。使用Ctrl-C停止，我得到以下错误日志:~/.rvm/gems/ruby-2.1.1/gems/spring-1.1.2/lib/spring/client/run.rb:47:in`gets':Interruptfrom~/.rvm/gems/ruby-2.1.1/gems/spring-1.1.2/lib/spring/client/run.rb:47:in`verify_server_version'from~/.rvm/gems/ruby-2.1.1/gems/spring-1.1.
ruby-on-rails - openshift 上的 rails 控制台 - 2
我将我的Rails应用程序部署到OpenShift，它运行良好，但我无法在生产服务器上运行“Rails控制台”。它给了我这个错误。我该如何解决这个问题？我尝试更新rubygems，但它也给出了权限被拒绝的错误，我也无法做到。railsc错误:Warning:You'reusingRubygems1.8.24withSpring.UpgradetoatleastRubygems2.1.0andrun`gempristine--all`forbetterstartupperformance./opt/rh/ruby193/root/usr/share/rubygems/rubygems
叮咚买菜基于 Apache Doris 统一 OLAP 引擎的应用实践 - 2
导读：随着叮咚买菜业务的发展，不同的业务场景对数据分析提出了不同的需求，他们希望引入一款实时OLAP数据库，构建一个灵活的多维实时查询和分析的平台，统一数据的接入和查询方案，解决各业务线对数据高效实时查询和精细化运营的需求。经过调研选型，最终引入ApacheDoris作为最终的OLAP分析引擎，Doris作为核心的OLAP引擎支持复杂地分析操作、提供多维的数据视图，在叮咚买菜数十个业务场景中广泛应用。作者｜叮咚买菜资深数据工程师韩青叮咚买菜创立于2017年5月，是一家专注美好食物的创业公司。叮咚买菜专注吃的事业，为满足更多人“想吃什么”而努力，通过美好食材的供应、美好滋味的开发以及美食品牌的孵
基于C#实现简易绘图工具【100010177】 - 2
C#实现简易绘图工具一.引言实验目的:通过制作窗体应用程序(C#画图软件),熟悉基本的窗体设计过程以及控件设计,事件处理等,熟悉使用C#的winform窗体进行绘图的基本步骤,对于面向对象编程有更加深刻的体会.Tutorial任务设计一个具有基本功能的画图软件**·包括简单的新建文件,保存,重新绘图等功能**·实现一些基本图形的绘制,包括铅笔和基本形状等,学习橡皮工具的创建**·设计一个合理舒适的UI界面**注明:你可能需要先了解一些关于winform窗体应用程序绘图的基本知识,以及关于GDI+类和结构的知识二.实验环境Windows系统下的visualstudio2017C#窗体应用程序三.
C51单片机——实现用独立按键控制LED亮灭（调用函数篇） - 2
说在前面这部分我本来是合为一篇来写的，因为目的是一样的，都是通过独立按键来控制LED闪灭本质上是起到开关的作用，即调用函数和中断函数。但是写一篇太累了，我还是决定分为两篇写，这篇是调用函数篇。在本篇中你主要看到这些东西！！！1.调用函数的方法（主要讲语法和格式）2.独立按键如何控制LED亮灭3.程序中的一些细节（软件消抖等）1.调用函数的方法思路还是比较清晰地，就是通过按下按键来控制LED闪灭，即每按下一次，LED取反一次。重要的是，把按键与LED联系在一起。我打算用K1来作为开关，看了一下开发板原理图，K1连接的是单片机的P31口，当按下K1时，P31是与GND相连的，也就是说，当我按下去时
kvm虚拟机安装centos7基于ubuntu20.04系统 - 2
需求：要创建虚拟机，就需要给他提供一个虚拟的磁盘，我们就在/opt目录下创建一个10G大小的raw格式的虚拟磁盘CentOS-7-x86_64.raw命令格式：qemu-imgcreate-f磁盘格式磁盘名称磁盘大小qemu-imgcreate-f磁盘格式-o?1.创建磁盘qemu-imgcreate-fraw/opt/CentOS-7-x86_64.raw10G执行效果#ls/opt/CentOS-7-x86_64.raw2.安装虚拟机使用virt-install命令，基于我们提供的系统镜像和虚拟磁盘来创建一个虚拟机，另外在创建虚拟机之前，提前打开vnc客户端，在创建虚拟机的时候，通过vnc
ruby - rbenv 安装权限被拒绝 - 2
大家好，我正在尝试设置一个开发环境，并且我一直在关注以下教程:Linktotutorial我做得不是很好，除了最基本的版本控制内容外，我对终端命令没有任何实际经验。我点击了第一个链接并尝试运行source~/.bash_profile我得到了错误；mkdir:/usr/local/rbenv/shims:权限被拒绝mkdir:/usr/local/rbenv/versions:权限被拒绝现在每次我加载终端时都会出现错误。bash_profile的内容；exportPATH=/usr/local/rbenv/bin:$PATHexportRBENV_ROOT=/usr/local/rbe
ruby-on-rails - 在 Rails 控制台中使用 asset_path - 2
在我的Character模型中，我添加了:字符.rbbefore_savedoself.profile_picture_url=asset_path('icon.png')end但是，对于数据库中已存在的所有角色，它们的profile_picture_url为nil。因此，我想进入控制台并遍历所有这些并进行设置。在我试过的控制台中:Character.find_eachdo|c|c.profile_picture_url=asset_path('icon.png')end但这给出了错误:NoMethodError:undefinedmethod`asset_path'formain:O
ruby-on-rails - 带有 Pry 的 Rails 控制台 - 2
当我进入Rails控制台时，我已将pry设置为加载代替irb。我找不到该页面或不记得如何将其恢复为默认行为，因为它似乎干扰了我的Rubymine调试器。有什么建议吗？最佳答案我刚发现问题，pry-railsgem。忘记了它的目的是让“railsconsole”打开pry。关于ruby-on-rails-带有Pry的Rails控制台，我们在StackOverflow上找到一个类似的问题： https://stackoverflow.com/question

Kafka基于Kraft下的权限控制

Kafka基于Kraft下的权限控制

安装

命令行中使用

java中使用

有关Kafka基于Kraft下的权限控制的更多相关文章

随机推荐