Kafka实操简单的一对一生产消费订阅模型生产者Producer：package main import ( "fmt" "github.com/IBM/sarama" "log" ) func main() { // 配置生产者 config := sarama.NewConfig() config.Producer.RequiredAcks = sarama.WaitForAll // 等待所有副本确认 config.Producer.Retry.Max = 3 // 重试次数 config.Producer.Return.Successes = true // 返回成功的消息 // 创建同步生产者（也可创建异步生产者） producer, err := sarama.NewSyncProducer([]string{"192.168.31.108:9092"}, config) if err != nil { log.Fatalf("Failed to create producer: %s", err) } defer producer.Close() // 发送消息 topic := "test-topic" for i := 0; i < 5; i++ { msg := &sarama.ProducerMessage{ Topic: topic, Value: sarama.StringEncoder(fmt.Sprintf("Hello Sarama! Message %d", i)), } // 同步发送，返回分区和偏移量 partition, offset, err := producer.SendMessage(msg) if err != nil { log.Printf("Failed to send message: %s", err) } else { fmt.Printf("Message sent to partition %d, offset %d\n", partition, offset) } } }消费者Consumer：package main import ( "fmt" "github.com/IBM/sarama" "log" ) func main() { // 配置消费者 config := sarama.NewConfig() config.Consumer.Return.Errors = true // 默认配置消费者偏移量 OffsetOldest: 从最早的消息开始消费 OffsetNewest:从当前最新消息开始消费 config.Consumer.Offsets.Initial = sarama.OffsetOldest // 2. 创建基础消费者（非消费组模式） consumer, err := sarama.NewConsumer([]string{"192.168.31.108:9092"}, config) if err != nil { log.Fatalf("Failed to create consumer: %s", err) } defer consumer.Close() // 3. 指定要消费的主题和分区 topic := "test-topic" partition := int32(0) // 分区编号（从0开始） // 4. 获取分区消费者（从指定偏移量开始消费） // 第三个参数为起始偏移量：可以是 sarama.OffsetOldest / sarama.OffsetNewest 或具体数值 // 无论是哪个参数都会进行持续监听当前的partition pc, err := consumer.ConsumePartition(topic, partition, sarama.OffsetOldest) if err != nil { log.Fatalf("Failed to consume partition: %s", err) } defer pc.Close() for msg := range pc.Messages() { // 成功接收到消息 fmt.Printf("%s %s %d %d\n", msg.Topic, msg.Value, msg.Partition, msg.Offset) } }多生产者单消费者大部分使用kafka的场景就是多个生产者生产消息，往kafka里面发，然后消费者在后面慢慢消费。这种模式不需要做额外配置，生产者只管往这个topic里面发就好了。举例实际应用中，有多个服务产生日志，这些服务就可以每个配置一个日志生产者，生产出来的日志信息传输进kafka，然后对应处理日志的服务作为消费者模型，进行各个服务日志处理，即消费kafka内的日志消息。这就是实战中常用的kafka日志处理订阅发布模型。消费者组我们上面第一个实战操作都是无副本的单机处理，而且消费者在因为某些情况故障或者崩溃的情况下，要么从最开始位置进行消费，要么从最近最新运行开始接收到的消息进行消费，不能做到从崩溃时消费的消息后一条开始消费。即消费者每次中断后都不知道自己消费到了哪里。如果需要知道，我们就得引入消费者组这个概念，使用消费者组可以自动实现 “记录消费位置” 的功能，无需手动管理偏移量，这是消费组最核心的优势之一。消费组会通过 kafka 集群自动跟踪和存储每个分区的消费偏移量（默认存储在 kafka 内部的 __consumer_offsets 主题中），并在消费者重启或重平衡（Rebalance）时自动恢复消费位置，无需手动持久化偏移量。package main import ( "context" "fmt" "github.com/IBM/sarama" "log" "os" "os/signal" "syscall" ) // GroupConsumer 自定义消费者，实现 sarama.ConsumerGroupHandler 接口 type GroupConsumer struct{} // Setup 在分区分配完成后调用（初始化） func (c *GroupConsumer) Setup(_ sarama.ConsumerGroupSession) error { return nil } // Cleanup 在分区被重新分配前调用（清理） func (c *GroupConsumer) Cleanup(_ sarama.ConsumerGroupSession) error { return nil } // ConsumeClaim 处理分区消息（核心逻辑） func (c *GroupConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { // 遍历分区消息 for msg := range claim.Messages() { // 处理消息（实际业务逻辑） fmt.Printf("Received message: %s (topic: %s, partition: %d, offset: %d)\n", string(msg.Value), msg.Topic, msg.Partition, msg.Offset) // 手动提交偏移量（可选，若关闭自动提交则必须手动调用） // 注意：需在消息处理成功后再提交，避免消息丢失 session.MarkMessage(msg, "**") // 标记消息为已处理，会在 session 结束时提交 } return nil } func main() { // 配置消费组 config := sarama.NewConfig() config.Version = sarama.V2_8_1_0 // 指定 Kafka 版本（需与集群版本匹配） // 偏移量配置：默认自动提交（可改为手动提交） config.Consumer.Offsets.AutoCommit.Enable = true // 开启自动提交 config.Consumer.Offsets.AutoCommit.Interval = 5000 // 自动提交间隔（毫秒） config.Consumer.Offsets.Initial = sarama.OffsetOldest // 初始偏移量（首次消费时） // 创建消费组 groupID := "auto-offset-group" // 消费组唯一标识 consumerGroup, err := sarama.NewConsumerGroup( []string{"192.168.31.108:9092"}, // Kafka 地址 groupID, config, ) if err != nil { log.Fatalf("Failed to create consumer group: %v", err) } defer consumerGroup.Close() // 要消费的主题 topics := []string{"test-topic"} consumer := &GroupConsumer{} // 启动消费循环协程 go func() { for { // 持续消费（重平衡后会重新调用） if err := consumerGroup.Consume(context.Background(), topics, consumer); err != nil { log.Printf("Consume error: %v", err) return } } }() // 等待中断信号 sigChan := make(chan os.Signal, 1) signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) <-sigChan fmt.Println("Shutting down consumer group...") } 负载均衡模式提到负载均衡，消费者组有一个概念，即当只有一个分区被多个消费者消费时，为了保证高可用，则默认只有其中一个消费者可以消费消息，其余的消费者阻塞待命，等消费者因为一些原因或者故障导致消费暂停，那么其他的消费者就会自动推选出一个进行当前消费，这种应该被称为故障转移模式。如上图，消费者①消费到39即暂停消费，但生产者生产到了44，则因为故障转移机制，生产者②衔接上上一位消费者接着消费后续消息内容。生产者组也可以进行负载均衡，一个生产者生产消息，消息被轮询到不同的消费者上，实现这个，我们需要在创建topic的时候，指定分区数量。我们先创建一个有3个分区的 topic：package main import ( "fmt" "github.com/IBM/sarama" "log" "time" ) // 检查主题是否存在（旧版API） func topicExists(client sarama.Client, topic string) (bool, error) { // 获取所有主题 topics, err := client.Topics() if err != nil { return false, fmt.Errorf("获取主题列表失败: %v", err) } // 检查目标主题是否在列表中 for _, t := range topics { if t == topic { return true, nil } } return false, nil } // 创建主题（旧版API） func createTopicIfNotExists(brokers []string, topic string, partitions int32, replicationFactor int16) error { // 配置客户端 config := sarama.NewConfig() config.Net.DialTimeout = 5 * time.Second // 创建基础客户端（旧版没有专门的AdminClient） client, err := sarama.NewClient(brokers, config) if err != nil { return fmt.Errorf("创建客户端失败: %v", err) } defer client.Close() // 检查主题是否存在 exists, err := topicExists(client, topic) if err != nil { return fmt.Errorf("检查主题存在性失败: %v", err) } if exists { log.Printf("主题 %s 已存在，无需创建", topic) return nil } // 构建创建主题的请求 request := sarama.CreateTopicsRequest{ TopicDetails: map[string]*sarama.TopicDetail{ topic: { NumPartitions: partitions, ReplicationFactor: replicationFactor, ConfigEntries: make(map[string]*string), }, }, Timeout: 10 * time.Second, // 10秒超时 } // 通过客户端发送创建请求（旧版需手动选择broker） broker, err := client.Controller() if err != nil { return fmt.Errorf("无法获取控制器broker") } // 发送请求 response, err := broker.CreateTopics(&request) if err != nil { return fmt.Errorf("发送创建请求失败: %v", err) } // 检查响应结果 for topic, err := range response.TopicErrors { fmt.Println(topic, err) } log.Printf("主题 %s 创建成功，分区数: %d，副本数: %d", topic, partitions, replicationFactor) return nil } func main() { brokers := []string{"192.168.31.108:9092"} topicName := "test-topic2" partitionCount := int32(3) replicationFactor := int16(1) err := createTopicIfNotExists(brokers, topicName, partitionCount, replicationFactor) if err != nil { log.Fatalf("操作失败: %v", err) } log.Println("操作完成") }然后我们使用三消费者监听这个 topic：我们会发现生产者生产出来了三个 partition 的消息，三个消费者就分别对应了一个 partition 进行消息消费。但是要注意，消费者数量必须是能够被 partition 整除的，例如：3个partition/3个消费者=1。如果 partition 为3，但是消费者只有2的时候，那么所有的消息都会被一个消费者包揽，另一个消费者阻塞待命，回到了最初单机故障转移机制的模型。安全认证kafka安全认证加密脚本：#!/bin/bash # 创建证书目录（与 Docker 挂载目录一致） mkdir -p kafka-certs && cd kafka-certs # 生成 CA 证书 openssl genrsa -out ca.key 2048 openssl req -new -x509 -key ca.key -out ca.crt -days 3650 \ -subj "/C=CN/ST=Beijing/L=Beijing/O=YourCompany/CN=kafka-ca" # 生成 Broker 证书（含 SAN 扩展） cat > san.cnf <<EOF [req] distinguished_name = req_distinguished_name req_extensions = v3_req prompt = no [req_distinguished_name] C = CN ST = Beijing L = Beijing O = YourCompany CN = kafka-broker [v3_req] subjectAltName = @alt_names ## 重要配置 [alt_names] IP.1 = 192.168.31.108 ## 运行kafka虚拟机本机ip DNS.1 = kafka-broker IP.2 = 10.3.0.2 ## 运行kafka容器配置的子网ip DNS.2 = kafka-broker EOF openssl genrsa -out kafka.server.key 2048 openssl req -new -key kafka.server.key -out kafka.server.csr -config san.cnf openssl x509 -req -in kafka.server.csr -CA ca.crt -CAkey ca.key -CAcreateserial \ -out kafka.server.crt -days 3650 -extensions v3_req -extfile san.cnf # 生成 JKS 格式证书（Bitnami 镜像专用） openssl pkcs12 -export \ -in kafka.server.crt -inkey kafka.server.key \ -name kafka-ssl -out kafka.keystore.p12 \ -password pass:yourpassword -certfile ca.crt keytool -importkeystore \ -srckeystore kafka.keystore.p12 -srcstoretype pkcs12 \ -destkeystore kafka.keystore.jks -deststoretype JKS \ -alias kafka-ssl -srcstorepass yourpassword -deststorepass yourpassword # 生成 Truststore keytool -keystore kafka.truststore.jks \ -alias ca-cert -import -file ca.crt \ -storepass yourpassword -noprompt echo "证书已生成到 kafka-certs 目录，请挂载到容器的 /bitnami/kafka/config/certs"docker-compose.yaml 配置文件：services: kafka-server: image: bitnami/kafka:3.9.0 restart: always ports: - "9095:9095" # 对外暴露 9095 端口 environment: # KRaft 基础配置 - KAFKA_ENABLE_KRAFT=yes - KAFKA_CFG_PROCESS_ROLES=controller,broker - KAFKA_CFG_NODE_ID=1 - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@10.3.0.2:9093 - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER # SSL 核心参数 - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=SSL - KAFKA_CFG_SSL_PROTOCOL=TLSv1.2 - KAFKA_CFG_SSL_CLIENT_AUTH=none - KAFKA_CFG_SSL_KEYSTORE_LOCATION=/bitnami/kafka/config/certs/kafka.keystore.jks - KAFKA_CFG_SSL_KEYSTORE_PASSWORD=yourpassword - KAFKA_CFG_SSL_KEY_PASSWORD=yourpassword - KAFKA_CFG_SSL_TRUSTSTORE_LOCATION=/bitnami/kafka/config/certs/kafka.truststore.jks - KAFKA_CFG_SSL_TRUSTSTORE_PASSWORD=yourpassword # 监听器配置 - KAFKA_CFG_LISTENERS=INTERNAL://:9092,CONTROLLER://:9093,SSL://:9095 - KAFKA_CFG_ADVERTISED_LISTENERS=SSL://192.168.31.108:9095,INTERNAL://10.3.0.2:9092 - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,SSL:SASL_SSL,INTERNAL:PLAINTEXT # SASL 认证（与 SSL 叠加） - KAFKA_CFG_SASL_ENABLED_MECHANISMS=PLAIN - KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL=PLAIN - KAFKA_CLIENT_USERS=user1 - KAFKA_CLIENT_PASSWORDS=password1 volumes: - /root/kafka/auth/kafka-certs:/bitnami/kafka/config/certs # 重要，网络分配 networks: kafka_network: ipv4_address: 10.3.0.2 # 分配其中一个子网给kafka，作为服务无需认证鉴权即可通过 # 重要，容器网络配置 networks: kafka_network: driver: bridge # 桥接 ipam: driver: default config: - subnet: 10.3.0.0/24 # 设置子网高可用集群简单实例version: '3.8' services: kraft-kafka-1: # 容器名：kraft-kafka-1 image: bitnami/kafka:3.9.0 container_name: kraft-kafka-1 restart: unless-stopped environment: - KAFKA_ENABLE_KRAFT=yes - KAFKA_KRAFT_CLUSTER_ID=abcdef1234567890 - KAFKA_CFG_PROCESS_ROLES=controller,broker - KAFKA_CFG_NODE_ID=1 - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kraft-kafka-1:9093,2@kraft-kafka-2:9093 - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=INTERNAL # 监听器：INTERNAL(容器间)、EXTERNAL(宿主机)、CONTROLLER(Raft) - KAFKA_CFG_LISTENERS=INTERNAL://:9092,EXTERNAL://:9095,CONTROLLER://:9093 - KAFKA_CFG_ADVERTISED_LISTENERS=INTERNAL://kraft-kafka-1:9092,EXTERNAL://192.168.31.108:9095 - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT ports: - "9095:9095" # 外部生产/消费端口 - "9093:9093" # 控制器端口 networks: - kafka-net kraft-kafka-2: # 容器名：kraft-kafka-2 image: bitnami/kafka:3.9.0 container_name: kraft-kafka-2 restart: unless-stopped environment: - KAFKA_ENABLE_KRAFT=yes - KAFKA_KRAFT_CLUSTER_ID=abcdef1234567890 - KAFKA_CFG_PROCESS_ROLES=controller,broker - KAFKA_CFG_NODE_ID=2 - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kraft-kafka-1:9093,2@kraft-kafka-2:9093 - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=INTERNAL - KAFKA_CFG_LISTENERS=INTERNAL://:9092,EXTERNAL://:9095,CONTROLLER://:9093 - KAFKA_CFG_ADVERTISED_LISTENERS=INTERNAL://kraft-kafka-2:9092,EXTERNAL://192.168.31.108:19095 - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT ports: - "19095:9095" # 外部生产/消费端口（避免冲突） - "19093:9093" # 控制器端口（避免冲突） networks: - kafka-net networks: kafka-net: driver: bridge然后生产者和消费者用之前的一般简易实现就能够进行互动了。多机 kafka 的好处就是，当生产者生产消息的时候，kafka 即便挂了一个，另一个也能自动被选举上去进行顶替使用，有一定的可用性与稳定性保证。（代码示例是单机多容器模拟多机，如果是多机的话，对应端口的网络地址配置不能用容器名代替，需要用实际IP代替）此次学习参照的枫枫大佬的文档：https://www.fengfengzhidao.com/article/9-OXrJgB8lppN5cbw2jc在此感谢大佬。
前言：在部署前请自行解决服务器和域名相关问题。n8n 是一款开源、低代码的工作流自动化平台。部署也相当容易，可选择 npm 安装，但是相对机器性能要求比较高；也可以选择 docker 进行安装，方便快捷但是需要会 docker 相关操作。安装和部署我这边的话推荐用 docker 进行安装，比如 1panel 面板上的一键安装，或者像这位 大佬 教的进行安装。如果有需要中文版，也可以使用 n8n-i18n-chinese 该Github仓库进行根据MD文档操作进行安装，该仓库支持 1.86.1 以上的版本直接使用仓库内的 docker-compose.yml 进行快速安装，不需要自己过多的操作。操作前提示：一定要打开自己域名管理里面的 Websocket 开关。因为 n8n 中的环境变量N8N_PUSH_BACKEND决定了后端如何向 UI 发送实时更新。它可以设置为 websocket 或 sse（服务器发送事件）：默认： websocket选项：websocket：使用 WebSockets 进行推送通知和实时更新。sse：使用服务器发送事件。1panel安装部署在应用商店直接搜索 n8n ，并点击安装即可，版本自选，端口自己设置在自己想的，n8n 默认端口为 5678。但记住，从 n8n 的 1.79 版本之后，官方便对 n8n 的持续服务网络连接进行了更改，往上版本如果环境变量参数缺少了部分参数，会导致部署好之后使用时提示 Connection lost 。这不是自己操作的问题，是官方并没有明确提出的问题。# 以下是我完整的容器环境变量 N8N_SECURE_COOKIE=false # 如果想要cookie验证这里可以改成true WEBHOOK_URL=https://xxx.xxx.com/ # 缺少需补上，1.79版本以上刚需该变量 N8N_HOST=xxx.xxx.com # 缺少需补上 N8N_DEFAULT_LOCALE=zh-CN # 如果是中文版需要加上，不是的话不需要，可以无视 PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin NODE_VERSION=20.19.0 YARN_VERSION=1.22.22 NODE_ICU_DATA=/usr/local/lib/node_modules/full-icu N8N_VERSION=1.86.1 NODE_ENV=production N8N_RELEASE_TYPE=stable SHELL=/bin/sh之后在面板里域名配置好对应端口的反向代理就好了。Docker直接部署这里以上述中文版仓库为例，将仓库 git clone 下来之后，按照作者的 README.md 说的操作就好了。之后配置好域名对应端口 15678:5678 的反向代理就可以了。但是环境变量相关问题同上1panel安装部署的参数一样，如果是高于 1.79 的版本都需要额外写入参数，保证服务连接。部署完在这之后只需要跳转到自己 n8n 的域名，就会有注册登录的页面，弄完之后就可以快乐的玩耍了。这之后再附带一个简单的自动化实战案例。案例这里我们就实现一个简单的 Github的 Golang 仓库接收到推送后，自动同步到服务器内编译并后台运行跑起服务的案例。这里需要在对应仓库内的 Settings → Webhook 里面配置好：仓库需要配置两个 Webhook，一个是测试的，一个是生产实际用的。这里的 Payload URL 要和 n8n 工作流里的 URL 一致，测试URL 即 https://域名/webhook-test/path 这里的 path 是 n8n 的 Webhook 里你自己设置的；如果是 生产URL即 https://域名/webhook/path。这样 Github 和 n8n 工作流的 Webhook 同步配置好后，就能接通了。Webhook 就像一个触发器，这样配 Github 仓库只要有 push 事件发生，就会像对应 URL 发送 POST 请求，n8n 工作流只要处于 active 状态，就会检测到 POST 请求并自动执行之后的工作流。Webhook 触发后就到 Execute Command 即在 n8n 所在系统执行你所需要的命令，这里我因为是 docker 部署的，所以命令都在容器里。这里就需要 ssh 连接到宿主机去对应文件区域执行我们所需要的命令。这里参考两位大佬写的文章：Docker容器如何优雅地访问宿主机网络Github Actions 实现 Nuxt 3 自动构建并部署因为我是容器 → 宿主机，所以需要将密钥放在容器里，公钥在宿主机。即生成好密钥之后，通过容器挂载配置直接找到 n8n 挂载的目录，把密钥丢进去。之后执行我们需要的命令就好：ssh -i /usr/local/lib/node_modules/n8n/node_modules/n8n-editor-ui/dist/密钥 用户名@host.docker.internal "sudo bash /path/path/xxx.sh"这里还是推荐访问到宿主机之后，用 shell 脚本去执行命令，会比较方便。我需要的使用场景就是到对应目录，将原先运行着的后台进程删掉，然后拉去仓库最近代码，修改对应代码之后打包在后台运行。#!/bin/bash # 切换到项目目录 cd /xxx/xxx/path/xxx.xxx.com/ # 杀掉 ./gacha 进程的第一个实例 sudo kill -9 $(ps aux | grep './xxx' | awk '{print $2}' | head -n 1) || true # 拉取最新的代码 sudo git pull # 修改 xxx.js 文件中的 URL sudo sed -i 'xxs|http://xxx/xxx|https://xxx/xxx|' /xxx/path/asset/xxx.js || true # 编译 Go 程序 sudo go build -o xxx main.go # 后台运行 Go 程序 nohup ./xxx > /dev/null 2>&1 &另外容器 ssh 连接到宿主机，会有一些问题，例如我的：sudo kill -9 $(ps aux | grep './xxx' | awk '{print $2}' | head -n 1) || true sudo sed -i 'xxs|http://xxx/xxx|https://xxx/xxx|' /xxx/path/asset/xxx.js || true nohup ./xxx > /dev/null 2>&1 &同是一个用户，但是容器 ssh 连接进来第1、2条命令必须加 sudo 才能正常执行，而且都需要使用一些手段处理命令的回参，不然会阻塞住 n8n 自动化执行命令的会话，这在人为手操相同命令是不会出现这类问题的。最后，后台运行编译好的文件必须舍弃掉 nohup 的回参，不然会因为不知名原因导致该命令变成前台运行切阻塞当前连接会话，让 n8n 自动化执行命令部分直接无限阻塞，导致自动化运行不下去。之后邮件发送通知那里自己弄好邮箱的 smtp 配置就好了，网上应该有挺多教的，就不赘述了。运行测试在真正使用前，最好每个自动化节点测试一下，看看命令啥的有没问题，会不会阻塞，有没有写错参数啥的。Webhook 的话，可以只 push 一次，Github 仓库的 Webhook 有个 Recent Deliveries，开启 n8n 的 Webhokk 测试监听后，去 Recent Deliveries 重发请求就好了，如果监听到了会有对应仓库的所有回参的。至于 Execute Command 的命令，需要自己去一步步试错了，不同的命令可能都会有不同的差异影响，但应该大体上没这么多问题。如果出问题了就当锻炼大伙排错能力了。最后邮件那里如果配置对的话就能发送你配置的邮件给你。当你用 生产URL 进行自动化的时候，n8n 概述里有个 执行 ，里面自己的自动化流程执行的什么样都可以看得到。总结感谢 Bhao 大佬的友情耐心协助。这次部署和配置学到了许多的东西，也真正的自己手动配置了一次自动化流程，感受到了自动化的方便快捷。希望该篇文章能够对大伙有些许帮助。
MySQL的InnoDB和MyISAM对于锁粒度的区别在MySQL中，InnoDB和MyISAM是两种常用的存储引擎，它们在锁粒度（Lock Granularity）上有显著的区别。锁粒度决定了在并发操作时，锁的范围和粒度，从而影响并发性能和数据一致性。以下是InnoDB和MyISAM在锁粒度上的主要区别：1. InnoDB锁粒度：行级锁（Row-Level Locking）。特点：行级锁：InnoDB使用行级锁来管理并发操作，这意味着在执行更新或删除操作时，锁的范围仅限于被操作的行，而不是整个表。高并发性能：由于锁的粒度较小，多个事务可以同时对表中的不同行进行操作，从而提高了并发性能。支持事务：InnoDB支持事务（ACID特性），在事务中可以使用多种隔离级别（如读已提交、可重复读等），这些隔离级别通过行级锁来实现。死锁检测：InnoDB支持死锁检测，当两个或多个事务相互等待对方释放锁时，InnoDB会检测到死锁并自动回滚其中一个事务。锁的类型：共享锁（S Lock）：允许其他事务读取同一行，但不允许修改。排他锁（X Lock）：禁止其他事务读取或修改同一行。适用场景：高并发的读写操作，如电子商务系统、金融系统等。需要事务支持的应用。2. MyISAM锁粒度：表级锁（Table-Level Locking）。特点：表级锁：MyISAM使用表级锁来管理并发操作，这意味着在执行更新或删除操作时，整个表会被锁定，其他事务无法对表中的任何行进行操作。低并发性能：由于锁的粒度较大，多个事务不能同时对表中的不同行进行操作，从而降低了并发性能。不支持事务：MyISAM不支持事务，因此在需要事务支持的应用中不适用。读写分离：虽然MyISAM使用表级锁，但在读操作时，多个读操作可以同时进行，只有写操作会阻塞其他操作。锁的类型：读锁（Read Lock）：允许其他事务读取表中的数据，但不允许修改。写锁（Write Lock）：禁止其他事务读取或修改表中的数据。适用场景：以读操作为主的应用，如内容管理系统（CMS）。不需要事务支持的应用。3. 锁粒度对比特性InnoDB（行级锁）MyISAM（表级锁）锁粒度行级锁表级锁并发性能高并发性能，适合读写混合场景低并发性能，适合读多写少场景事务支持支持事务（ACID）不支持事务锁类型共享锁（S Lock）、排他锁（X Lock）读锁（Read Lock）、写锁（Write Lock）适用场景高并发读写操作、需要事务支持以读操作为主、不需要事务支持4. 示例InnoDB（行级锁）-- 创建InnoDB表 CREATE TABLE innodb_table ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(50), age INT ) ENGINE=InnoDB; -- 事务1 BEGIN; UPDATE innodb_table SET age = 26 WHERE id = 1; -- 事务2 BEGIN; UPDATE innodb_table SET age = 27 WHERE id = 2; -- 不会被事务1阻塞，因为锁的是不同行MyISAM（表级锁）-- 创建MyISAM表 CREATE TABLE myisam_table ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(50), age INT ) ENGINE=MyISAM; -- 事务1 LOCK TABLES myisam_table WRITE; UPDATE myisam_table SET age = 26 WHERE id = 1; -- 事务2 UPDATE myisam_table SET age = 27 WHERE id = 2; -- 会被事务1阻塞，因为整个表被锁定5. 总结InnoDB：适合高并发的读写操作。支持事务和行级锁，能够有效提高并发性能。适用于需要事务支持的应用，如金融系统、电子商务等。MyISAM：适合以读操作为主的应用。使用表级锁，适合读多写少的场景。不支持事务，适用于不需要事务支持的应用。在选择存储引擎时，需要根据应用的具体需求和场景来决定。如果需要高并发性能和事务支持，建议选择InnoDB；如果以读操作为主且不需要事务支持，可以选择MyISAM。
Redis string大数据量如何扩容Redis字符串（SDS）的扩容策略当字符串长度小于1MB时：如果当前SDS的剩余空间不足以容纳新数据，Redis会将SDS的容量扩展为当前长度的两倍。这种策略可以减少频繁的内存分配操作，提高性能。当字符串长度超过1MB时：每次扩容只会增加1MB的空间，而不是成倍扩展。这种策略可以避免一次性分配过多内存，导致资源浪费。最大长度限制：Redis字符串的最大长度为512MB。扩容过程当SDS的剩余空间不足以容纳新内容时，Redis会触发扩容操作：重新分配内存：根据上述策略计算新的内存分配大小，并分配新的内存空间。复制数据：将原有数据复制到新的内存空间中。释放旧内存：释放旧的内存空间。如果扩容导致SDS的类型发生变化（例如，从SDS_TYPE_8变为SDS_TYPE_16），Redis还会更新SDS的编码类型，并可能需要移动数据到新的内存位置。优化策略空间预分配：Redis通过预分配空间来减少内存分配的频率。当字符串长度小于1MB时，每次扩容会分配双倍空间；当长度超过1MB时，每次扩容增加1MB。惰性空间释放：当字符串缩短时，SDS不会立即释放多余的空间，而是保留这些空间以备后续使用，减少后续扩容的开销。这种扩容机制使得Redis能够在处理大量数据时保持高效的性能，同时避免了内存分配和释放带来的开销。和Go的切片扩容机制类似，但也只是形似。Go的是超过1024大小之后就以1.25倍扩容，redis的string是超过1mb就扩1mb，小于1024/1mb的时候就是扩当前长度的两倍。
Redis使用Lua脚本示例三则一则：比较简易的使用Lua语言新增HSET的值package main import ( "context" "fmt" "log" "github.com/go-redis/redis/v8" ) var ctx = context.Background() func main() { // 连接到 Redis rdb := redis.NewClient(&redis.Options{ Addr: "localhost:6379", // Redis 服务器地址 Password: "", // Redis 密码，没有则留空 DB: 0, // 默认数据库编号 }) // 测试连接 _, err := rdb.Ping(ctx).Result() if err != nil { log.Fatalf("Failed to connect to Redis: %v", err) } fmt.Println("Connected to Redis successfully!") // 定义 Lua 脚本 // 这个脚本会插入一个字段到 Hash 中，并返回该字段的值 luaScript := ` local key = KEYS[1] local field = KEYS[2] local value = ARGV[1] redis.call("HSET", key, field, value) return redis.call("HGET", key, field) ` // 调用 Lua 脚本 key := "myHash" field := "myField" value := "myValue" // 使用 EVAL 命令运行 Lua 脚本 result, err := rdb.Eval(ctx, luaScript, []string{key, field}, value).Result() if err != nil { log.Fatalf("Failed to execute Lua script: %v", err) } // 打印结果 fmt.Printf("Inserted and retrieved value: %v\n", result) }二则：将多个值新增进SET内package main import ( "context" "fmt" "github.com/go-redis/redis/v8" "log" ) //var ctx = context.Background() func main() { rdb := redis.NewClient(&redis.Options{ Addr: "127.0.0.1:6379", Password: "", DB: 0, //0就用默认的DB PoolSize: 100, //连接池的大小 }) _, err := rdb.Ping(context.Background()).Result() if err != nil { fmt.Printf("%s, redis连接失败\n", err.Error()) return } fmt.Printf("redis连接成功\n") // 定义 Lua 脚本 luaScript := ` for i = 1, #KEYS do redis.call("SET", KEYS[i], ARGV[i]) end local res = {} for i = 1, #KEYS do table.insert(res, redis.call("GET", KEYS[i])) end return res ` // 调用 Lua 脚本 keys := []string{"name", "email", "age"} values := []string{"leijie", "rolingg@qq.com", "22"} // 使用 EVAL 命令运行 Lua 脚本 results, err := rdb.Eval(context.Background(), luaScript, keys, values).Result() if err != nil { log.Fatalf("Failed to execute Lua script: %v", err) } fmt.Println(results) }三则：将多个值新增进HSETpackage main import ( "context" "fmt" "log" "github.com/go-redis/redis/v8" ) var ctx = context.Background() func main() { // 连接到 Redis rdb := redis.NewClient(&redis.Options{ Addr: "localhost:6379", // Redis 服务器地址 Password: "", // Redis 密码，没有则留空 DB: 0, // 默认数据库编号 }) // 测试连接 _, err := rdb.Ping(ctx).Result() if err != nil { log.Fatalf("Failed to connect to Redis: %v", err) } fmt.Println("Connected to Redis successfully!") // 定义 Lua 脚本 luaScript := ` local keys = KEYS local values = ARGV local ttl = tonumber(values[#values]) -- 获取过期时间（以秒为单位） -- 每个键的字段数量 local fieldsPerKey = 3 -- 初始化值的索引 local valueIndex = 1 -- 批量插入字段并设置过期时间 for i = 1, #keys, fieldsPerKey + 1 do local key = keys[i] for j = 1, fieldsPerKey do local field = keys[i + j] local value = values[valueIndex] redis.call("HSET", key, field, value) valueIndex = valueIndex + 1 end -- 设置过期时间 redis.call("EXPIRE", key, ttl) end -- 检索指定字段 local res = {} for i = 1, #keys, fieldsPerKey + 1 do local key = keys[i] for j = 1, fieldsPerKey do local field = keys[i + j] table.insert(res, redis.call("HGET", key, field)) end end return res ` // 调用 Lua 脚本 keys := []string{ "myHash1", "name", "email", "age", "myHash2", "name", "email", "age", } values := []string{ "John Doe", "john.doe@example.com", "30", "Jane Smith", "jane.smith@example.com", "25", } ttl := 60 // 设置过期时间为 60 秒 // 将过期时间添加到 values 数组的末尾 values = append(values, fmt.Sprintf("%d", ttl)) // 使用 EVAL 命令运行 Lua 脚本 results, err := rdb.Eval(ctx, luaScript, keys, values).Result() if err != nil { log.Fatalf("Failed to execute Lua script: %v", err) } // 打印结果 fmt.Printf("Inserted and retrieved values: %v\n", results) } 这里EVAL命令里面包含四个值，分别是context、lua命令脚本、KEYS、VALUES，其中 KEYS 和 VALUES 可以是 []string 或者 []interface。为什么在并发场景要使用Lua来操纵Redis？减少网络往返次数直接命令：如果你从客户端发送多个 Redis 命令（例如 HSET、EXPIRE 等），每个命令都会产生一次网络往返。在并发场景下，多个客户端同时发送多个命令，网络延迟会显著增加。Lua 脚本：Lua 脚本将多个操作封装在一个命令中，通过 EVAL 或 EVALSHA 命令一次性发送给 Redis 服务器。这样可以显著减少网络往返次数，从而提高性能。原子性直接命令：多个命令在客户端发送时是独立执行的，可能会被其他命令插入，导致操作不原子。例如，如果你先执行 HSET，然后执行 EXPIRE，在两者之间可能会有其他客户端的命令插入，导致不一致。Lua 脚本：Lua 脚本在 Redis 服务器端执行时是原子的。一旦脚本开始执行，Redis 会将其作为一个整体处理，不会被其他命令打断。这确保了操作的一致性和完整性。减少客户端负载直接命令：客户端需要处理多个命令的发送和响应，这会增加客户端的负载，尤其是在并发场景下。Lua 脚本：客户端只需要发送一个 EVAL 命令，由 Redis 服务器端执行脚本并返回结果。这减轻了客户端的负载，使得客户端可以更高效地处理其他任务。减少 Redis 服务器的上下文切换直接命令：每个命令都会触发一次 Redis 服务器的上下文切换，尤其是在高并发场景下，这会导致服务器性能下降。Lua 脚本：Lua 脚本在 Redis 服务器端执行时，所有操作都在一个上下文中完成，减少了上下文切换的开销。优化性能直接命令：多个命令的执行需要多次解析和处理，增加了 Redis 服务器的处理时间。Lua 脚本：Lua 脚本在 Redis 服务器端执行时，所有操作都在一个脚本中完成，减少了解析和处理的开销。Redis 服务器可以更高效地执行脚本中的命令。综合来说，Lua脚本的作用就是将多个Reids命令集成在一个上下文内一起进行操作，从而减少操作次数，优化负载（无论是网络还是服务端负载），并且所有操作都是具有原子性。原子性：一个操作或多个操作要么全部成功，要么全部失败，不会出现部分成功的情况。原子性示例假设我们有一个应用，需要对某个特定事件进行计数。在高并发环境下，我们需要确保计数的原子性，避免出现数据不一致的问题。local key = KEYS[1] -- 获取第一个键 local value = tonumber(ARGV[1]) -- 获取第一个参数，作为初始值 local increment = tonumber(ARGV[2]) -- 获取第二个参数，作为增量 -- 获取当前键的值 local current_value = redis.call('GET', key) -- 如果键存在且有值 if current_value then current_value = tonumber(current_value) -- 合并当前值与新增值 redis.call('SET', key, current_value + increment) return current_value + increment else -- 如果键不存在或值为空，设置初始值 redis.call('SET', key, increment) return increment end具体更多的用法看这位大佬写的吧：https://blog.csdn.net/z_344791576/article/details/143996794示例对比假设你需要为多个键批量设置字段并设置过期时间，以下是两种方式的对比：直接命令for _, key := range keys { for _, field := range fields { _, err := rdb.HSet(ctx, key, field, value).Result() if err != nil { log.Fatalf("Failed to set field: %v", err) } } _, err := rdb.Expire(ctx, key, time.Duration(ttl)*time.Second).Result() if err != nil { log.Fatalf("Failed to set expire: %v", err) } } // 多次循环执行命令，负载高Lua 脚本-- Lua 脚本 local keys = KEYS local values = ARGV local ttl = tonumber(values[#values]) for i = 1, #keys, fieldsPerKey + 1 do local key = keys[i] for j = 1, fieldsPerKey do local field = keys[i + j] local value = values[j] redis.call("HSET", key, field, value) end redis.call("EXPIRE", key, ttl) endGo 调用 Lua 脚本keys := []string{ "myHash1", "name", "email", "age", "myHash2", "name", "email", "age", } values := []string{ "John Doe", "john.doe@example.com", "30", "Jane Smith", "jane.smith@example.com", "25", } ttl := 60 values = append(values, fmt.Sprintf("%d", ttl)) results, err := rdb.Eval(ctx, luaScript, keys, values).Result()性能对比直接命令：每次操作都需要一次网络往返，高并发时性能较差。Lua 脚本：一次网络往返，原子性操作，性能显著提升。总结在并发场景下，使用 Lua 脚本操作 Redis 有以下优势：减少网络往返次数，提高性能。保证操作的原子性，确保一致性。减轻客户端负载，提高客户端效率。减少 Redis 服务器的上下文切换，优化服务器性能。因此，在需要批量操作或高并发场景下，推荐使用 Lua 脚本。补充上面提到的 EVALSHA 是 Redis 提供的一个命令，用于执行已经加载到 Redis 的 Lua 脚本。package main import ( "context" "fmt" "log" "github.com/go-redis/redis/v8" ) var ctx = context.Background() func main() { // 连接到 Redis rdb := redis.NewClient(&redis.Options{ Addr: "localhost:6379", // Redis 服务器地址 Password: "", // Redis 密码，没有则留空 DB: 0, // 默认数据库编号 }) // 测试连接 _, err := rdb.Ping(ctx).Result() if err != nil { log.Fatalf("Failed to connect to Redis: %v", err) } fmt.Println("Connected to Redis successfully!") // 定义 Lua 脚本 luaScript := ` local keys = KEYS local values = ARGV local ttl = tonumber(values[#values]) -- 获取过期时间（以秒为单位） -- 每个键的字段数量 local fieldsPerKey = 3 -- 批量插入字段并设置过期时间 for i = 1, #keys, fieldsPerKey + 1 do local key = keys[i] local valueIndex = 1 -- 初始化值的索引 for j = 1, fieldsPerKey do local field = keys[i + j] local value = values[valueIndex] -- 获取对应的值 redis.call("HSET", key, field, value) valueIndex = valueIndex + 1 -- 移动到下一个值 end -- 设置过期时间 redis.call("EXPIRE", key, ttl) end -- 检索指定字段 local res = {} for i = 1, #keys, fieldsPerKey + 1 do local key = keys[i] for j = 1, fieldsPerKey do local field = keys[i + j] table.insert(res, redis.call("HGET", key, field)) end end return res ` // 加载 Lua 脚本并获取 SHA1 哈希值 scriptSHA, err := rdb.ScriptLoad(ctx, luaScript).Result() if err != nil { log.Fatalf("Failed to load Lua script: %v", err) } fmt.Printf("Script SHA1: %s\n", scriptSHA) // 调用 Lua 脚本 keys := []string{ "myHash1", "name", "email", "age", "myHash2", "name", "email", "age", } values := []string{ "John Doe", "john.doe@example.com", "30", "Jane Smith", "jane.smith@example.com", "25", } ttl := 60 // 设置过期时间为 60 秒 // 将过期时间添加到 values 数组的末尾 values = append(values, fmt.Sprintf("%d", ttl)) // 使用 EVALSHA 命令运行 Lua 脚本 results, err := rdb.EvalSha(ctx, scriptSHA, keys, values).Result() if err != nil { log.Fatalf("Failed to execute Lua script: %v", err) } // 打印结果 fmt.Printf("Inserted and retrieved values: %v\n", results) }性能优化：使用 EVALSHA 可以避免重复加载相同的脚本，从而提高性能。如果脚本已经加载到 Redis 中，EVALSHA 命令会直接执行脚本，而不需要重新解析和加载。但其实普通用和 Eval 用法差距不大。
[Golang] Map的底层实现原理与并发安全（待填坑）
Map的底层实现原理Go语言采用的是哈希查找表，并且使用链表法解决哈希冲突。另外在面试里面，一般问到Map都会问它是否是并发安全的。想要回答这类问题，只需要记住这一句话：Go语言提供的数据类型中，只有channel是并发安全的。Map自身并发安全失败场景func main() { //创建map m := make(map[int]string) var wg sync.waitGroup wg.Add(2) //协程1写 go func() { for true { m[0] = "xiaoming" } wg.Done() }() //协程2读 go func() for true { _ = m[0] } wg.Done() }() wg.wait() fmt . Print1n("结束") }编译后运行会报错：fatal error: concurrent map read and map write 原因笔记来源：详解Go语言如何解决map并发安全问题实现并发安全的Map读写锁+map只需要在map进行读操作的时候，加读取锁后再进行读取；添加、更新、删除、遍历、获取长度这些操作加写锁再进行操作。每次进行这些map相关的操作都要进行加锁，我们通过学习Go语言的锁就会知道，读锁可以实现并发读，但是写锁只能单个写，阻塞其他对于此处的读写操作，是独占的。因此这样虽然能够保证数据的安全性，但也会影响程序的运行效率。所以这种加读写锁的并发安全做法一般适用于读多写少的场景。分片+锁分片+锁，顾名思义就是将map分成多个片段，每个片段都由一个锁进行保护。数据通过散列函数将键分布到不同的小map片段中，从而减少锁的竞争，提高并发性能。这种方法适用于写操作较多，且写操作分布均匀的场景。分片map的原理是将map划分为多个片段，每个片段由一个单独的锁保护。当对map进行读写操作时，只需要锁定对应的map片段即可。这样可以大大提高并发读写的性能。sync.MapGo语言标准库提供的 sync.Map 类型，它不是为了替代普通的map，而是用于特定的场景。sync.Map 适用于读多写少的场景，并且写操作的key分布比较分散。sync.Map内部使用了一个只读的map和一个可写的map，读操作默认是无锁的，而写操作需要加锁。当 misses（未命中读map的次数）超过一定阈值时，会将可写的map复制到只读map中。sync.Map 不能被复制，且在高频率写操作时性能可能下降。sync.Map 是 Go 语言中提供的一个并发安全的 map 类型，它允许多个 goroutine 安全地使用 map 进行读写操作而不需要使用互斥锁（mutex）。sync.Map 包里有增删改查等 Go 官方封装好的方法，用起来很方便嗷！具体的话还请看我之前写的：[[Golang基础] Sync包](https://rolingg.top/index.php/archives/103/)读写锁+map 和 Sync.Map 相比较：读写锁+map优点：简单直观：使用sync.RWMutex 来保护一个普通的map，读操作使用读锁，写操作使用写锁。适合读多写少的场景：允许多个读操作并发执行，而写操作是独占的，适合读操作远多于写操作的情况。缺点：读操作性能受限：在高并发读操作下，虽然可以并发读取，但如果写操作频繁，读操作会因为锁的争用而受到性能影响。写操作独占：写操作需要独占锁，这会阻塞所有的读和写操作，可能导致性能瓶颈。sync.Map优点：并发性能高：sync.Map 通过读写分离和延迟删除机制，优化了读多写少的场景下的并发性能。空间换时间：使用两个map（read和dirty）来提高性能，减少了锁的使用，提高了并发读取的效率。延迟删除：删除操作不会立即清理内存，而是标记为删除，直到下次同步时才清理，减少了锁的操作。缺点：不适合大量写入的场景：在写操作频繁的场景下，sync.Map的性能可能下降，因为需要频繁地将dirty中的数据同步到read中。内存使用高：由于维护了两个map，可能会增加内存的使用。综上，sync.Map 对读操作加锁的优化更好，但是对于一些需要复制map，且对内存占用要求小的场景，用简单实现的读写锁+map会更好。总结写场景频繁，对性能要求也比较高的话，还是使用 分片+锁 的方式吧；其余的场景得看对内存的要求，对内存占用要求小的话，就用 读写锁+map，没这个需求的话，就用 sync.Map，方便快捷。
Channel原理上来先贴一段源码type hchan struct { qcount uint // 队列中的总元素个数 dataqsiz uint // 环形队列大小，即可存放元素的个数 buf unsafe.Pointer // 环形队列指针 elemsize uint16 //每个元素的大小 closed uint32 //标识关闭状态 elemtype *_type // 元素类型 sendx uint // 发送索引，元素写入时存放到队列中的位置 recvx uint // 接收索引，元素从队列的该位置读出 recvq waitq // 等待读消息的goroutine队列 sendq waitq // 等待写消息的goroutine队列 lock mutex //互斥锁，chan不允许并发读写 } //recvq 那些已经准备好从 channel 中读取数据，但因为 channel 当前没有可供读取的数据而处于等待状态的 goroutines 的队列。 //sendq 等待机会将数据发送到 channel 中的 goroutine 的队列向 channel 写数据:若等待接收队列 recvq 不为空，则缓冲区中无数据或无缓冲区，将直接从 recvq 取出 G ，并把数据写入，最后把该 G 唤醒，结束发送过程。说明：如果接收队列 recvq 不为空，说明有 goroutine 正在等待从 channel 中读取数据。此时，如果 channel 的缓冲区中没有数据或者 channel 没有缓冲区，将直接从 recvq 中取出一个等待的 goroutine G，把要发送的数据写入 G，然后将 G 唤醒，结束发送过程。也就是说，原本写入的数据是写进 channel 的缓冲区中，但因为缓冲区没有数据或者channel没有缓冲区，导致这时，发送数据的操作不是将数据放入缓冲区，而是直接将数据传递给这个等待读取的 goroutine G。有人可能会有疑问，为什么channel有缓冲区，而且缓冲区没有数据，写入的数据不写入缓冲区而是写入等待读消息的goroutine G内。这是因为有等待读消息的goroutine G要的就是缓冲区的数据，但是由于缓冲区没有数据了，又有新数据来，不如直接将 数据 → 等待读消息的goroutine G，而不是多一步的 数据 → channel缓冲区 → 等待读消息的goroutine G。这样做①是可以提高效率，直接将数据传递给等待读取的 goroutine 可以减少数据复制的开销，因为数据不需要先写入缓冲区再从缓冲区读取出来。这样可以提高程序的效率。②是可以避免死锁，如果缓冲区为空，而发送的数据被写入缓冲区，那么在没有其他 goroutine 准备读取的情况下，发送操作将无法完成，这可能导致死锁。通过直接将数据传递给等待读取的 goroutine，可以避免这种情况。若等待接收队列 recvq 为空，且缓冲区中有空余位置，则将数据写入缓冲区，结束发送过程。若等待接收队列 recvq 不为空，且缓冲区中没有空余位置，则将发送数据写入 G，将当前 G 加入 sendq ，进入睡眠，等待被读 goroutine 唤醒。从 channel 读数据：若等待发送队列 sendq 不为空，且没有缓冲区，直接从 sendq 中取出 G ，把 G 中数据读出，最后把 G 唤醒，结束读取过程。说明：当从无缓冲的 channel 读取数据时，如果发送队列 sendq 不为空，这意味着有 goroutine 正在等待发送数据到这个变量。无缓冲的 channel 要求发送和接收操作同步进行，即发送方必须等待接收方准备好接收数据。因为 channel 是无缓冲的，所以数据是在等待发送的 goroutine G 的本地变量中。因为数据已经被接收方读取（也就是channel读取了）。G 可以继续执行它之后的代码，通常是在 G 中的 send 操作之后的代码。在有缓冲区的情况下，若等待发送队列 sendq 不为空，说明缓冲区已满，从缓冲区中首部读出数据，把 G 中数据写入缓冲区尾部，把 G 唤醒，结束读取过程。此情况基于有一个 goroutine 尝试从 channel 读取数据。如果缓冲区中有数据，则从缓冲区取出数据（指有一个 goroutine 尝试从 channel 读取数据），结束读取过程。这个操作不需要考虑 sendq，因为缓冲区中已经有数据可以被读取。图同2差不多将当前 goroutine 加入 recvq ，进入睡眠，等待被写 goroutine 唤醒。在这种情况下，因为没有数据可读且没有发送操作在等待（也就是没有缓冲区或者缓冲区内没有数据，同时 sendq 为空），执行读取操作的 goroutine 将自身加入到与 channel 关联的接收队列 recvq 中。上面的图多少有点误，比如说 两个队列里不为空的话其实都是有限个 goroutine，我为了方便复制粘贴了几个空白在里面，想着意思到位就行。其实这样看就会发现 channel 有点像科幻电影的能量传输仓，有一个三仓机器，左边一个仓右边一个仓，顶部可以按能量存储器（也就是缓冲区）就分别对应 channel 上面的四种情况。关闭 channel：1.关闭 channel 时会将 recvq 中的 G 全部唤醒，本该写入 G 的数据位置为 nil（因为要被读消息的对象没了）。将 sendq 中的 G 全部唤醒，但是这些 G 会 panic（因为没有写消息的对象了）。panic 出现的场景还有：关闭值为 nil 的 channel关闭已经关闭的 channel向已经关闭的 channel 中写数据备注：上面的data啊，变量啊，都是 goroutine 里面的，channel 说白了就是用来在 goroutine 里面传输数据的。无缓冲 Chan 的发送和接收是否同步？无缓冲的 channel：当 channel 无缓冲时，发送操作必须等待另一个 goroutine 来接收数据。如果还没有接收方，发送方将会阻塞，直到有接收方准备好接收数据。同样，接收操作也必须等待数据被发送。如果没有发送方，接收方将会阻塞，直到有发送方准备好发送数据。因此，无缓冲的 channel 确保了发送和接收操作是同步的，即它们必须同时发生。也就是说，无缓冲的 channel 通过阻塞强制让数据同步。有缓冲的 channel：当 channel 有缓冲时，发送操作可能不需要等待接收方。如果缓冲区未满，发送方可以直接将数据放入缓冲区，而不必立即等待接收方。同样，接收操作也可能不需要等待发送方。如果缓冲区未空，接收方可以直接从缓冲区读取数据，而不必立即等待发送方。然而，如果缓冲区满了，发送方在缓冲区有空间之前将会阻塞。类似地，如果缓冲区空了，接收方在缓冲区中有数据之前将会阻塞。有缓冲的 channel 允许在缓冲区非空或未满的情况下发送和接收操作异步进行。
Kafka高可用实例kafka 建议不要用 单分区的Topic 去进行消息的持久化存储，这要会导致一些问题。例如：一个集群原本有三个节点，但是只用了一个节点。Kafka集群中一个节点出问题了，而有需要用到的服务对应的 Topic 单分区在这个节点上，服务所有的副本都部署在这个节点上，就会导致服务的功能异常，整个功能不可用。这种叫 单点故障，不满足高可用。解决方法也很简单，调整 Topic 的分区数量即可。调整 Topic 的分区数量，需要分区的副本数量和 kafka 集群节点数量保持一致，确保当 kafka 其中一个节点出现问题时，对应的服务功能能够正常工作。这个问题，Topic 分区对应的服务全都在一个节点内，其他节点没有对应的 Topic 分区时，当这个节点出现问题，就会导致整的服务挂掉。但是如果多分区，就能让问题服务能够正常运行，这其实归于Kafka的分区策略、负载均衡策略和故障转移策略。当从 1 个分区增加到 3 个分区时，增加分区数量，这样可以分散负载，提高并行处理能力。一般来说，还是会手动将 3 个分区分配到 3 个broker中，一个broker内分别有3个分区及他们的副本，保证可用性，也应证了这句话 需要分区的副本数量和 kafka 集群节点数量保持一致。这样做之后，一个Topic就在多个节点上有多个分区，即便一个节点挂了，其他分区也有对应的分区及副本，保证可用性这样当有这个 Topic 分区的节点出问题时，上面的服务所在的分区就会自行进行新的Leader选举，因为其他的节点有可用分区及副本，及保证了的高服务可用。总结不要那么傻傻的一个Topic里只有一个节点挂载着服务的分区和副本，要保证可用性，最好就是多节点都有对应的服务分区和副本。记得别忘了把消费者上涨到上调到 Topic 分区数+1或以上个的数量，这是为了保证每个分区的消息队列都有消费者进行消费，即便消费者挂掉了一个，也有新的消费者填上，一致保持着良性的消费过程，不让消息堆积在消息队列中。
Kubernetes调度策略及相关内容Scheduler调度的过程和策略事先补充：Node 是 Kubernetes 集群中的工作节点一个 Node 可以运行多个 Pod，而一个 Pod 只能运行在一个 Node 上使用标签和选择器可以管理 Node 和 Pod 之间的关系，从而实现灵活的调度和管理。Scheduler 是 Kubernetes 集群的调度器，把 Pod 分配给集群的节点（Node）scheduler 是一个单独运行的程序，只要启动之后就会一直监听apiserver。获取报文中的字段（即spec中的nodeName字段）创建 Pod 时，为每个 Pod 创建一个 binding，表示该往哪个节点上部署。调度规则：公平——每个节点都能够分配资源资源高效利用——集群中的资源可以被最大化使用效率——调度的性能要好，能够尽快的完成大批量 Pod 的调度工作灵活——允许用户根据自己的需求，控制和改变调度的逻辑kubernetes对Node的选择策略预算策略：用于过滤出合适的节点优选策略：用于选择出合适的部署节点选择 Node 的过程：先执行预算策略，再执行优先策略。这两步的操作都必须成功，否则立刻返回报错。以上都是自带的算法策略，由K8s集群自己来选择。预算策略podfitsresources：Pod 的适应策源，检查节点上剩余的资源是否满足pod请求的资源（主要是CPU和内存）podfitshost：Pod 适应主机，如果Pod指定了 Node 的 name，检测主机名是否存在，如果存在要和 Pod 指定的名称匹配，这才能调度过去podselectormarches：Pod 选择器匹配，根据选择器选择的 Node 标签进行匹配，如果 Node 上存在匹配的标签，则选择该 Node。nodeskconflict：无磁盘冲突，确保已经挂载的卷和Pod的卷不发生冲突，除非目录是只读。如果预算策略不满足，Pod 将始终处于pending状态，不断的重试调度，直到节点满足条件为止。优选策略leastrequestedpriority：最低请求优先级，通过算法计算节点上的CPU和内存使用率，确定节点的权重。使用率越低的节点，相应的权重就越高。调度时会更倾向于这些使用率低的节点。实现资源合理的利用。越低，被使用的资源越少，则被调用的权重越高。balancereasourceallocation：平衡资源分配，算CPU和内存的使用率，给节点赋予权重。权重算的是CPU和内存使用率接近，权重越高。和上面的最低请求优先级一起使用。使用率越接近，两种资源的量越接近，因为其中一个资源不够用的概率越小，被调用的权重越高。例如：Node① CPU：20%，内存：80%Node② CPU：50%，内存：50%如果选择Node①，则因为内存不足的可能极大，导致使用出错；如果使用Node②，两个资源剩余的量都很平均，出现其中一个资源不足导致使用出错的问题概率变小；主打一个要么都够用，要么都不够用。imagelocalitypriority：节点上是否已经有了要部署的镜像。镜像的总数成正比，满足的镜像数越多，权重越高。已经有的镜像越多，则要做的准备工作就越少（因为未部署的镜像要拉取配置啥的）。满足的镜像越多部署的就越快。kubernetes对Pod的调度策略在 Kubernetes 中，调度是指将 Pod 放置到合适的节点上，以便对应节点上的 Kubelet 能够运行这些 Pod。Kubelet 是 Kubernetes 集群中的一个关键组件，它是集群中的每个节点（Node）上运行的代理服务。Kubelet 的主要作用是确保容器都运行在 Pod 中，并且这些 Pod 符合集群的状态。它负责维护容器的生命周期，包括创建、更新和销毁容器。定向调度：使用 nodeName 字段指定 Node 名称；使用 nodeSelector 字段指定 Node 标签。亲和性调度：使用 Node/Pod 亲和性（NodeAffinity、PodAffinity、PodAntiAffinity）；污点与容忍：使用 Node 设置污点，结合 Pod 设置容忍。全自动调度：运行在哪个节点上完全由 Scheduler 经过一系列的算法计算得出。定向调度nodeName：指定节点对应的节点名称，用于将 Pod 调度到指定的 Node 上，不经过调度器。apiVersion: v1 kind: Pod metadata: name: my-pod spec: containers: - name: my-container image: my-image nodeName: "node-name"在参数中设置了nodeName，指定了节点的名称。这个规则是强制匹配,会跳过scheduler的调度策略。相当于说用了这个 Pod 有了指定的主人，不会被调到别的 Node。nodeSelector：在 Pod 定义文件的 spec 字段下的 nodeSelector 字段中设置一个标签选择器，在 Pod 调度的时候，只有具有这些标签的 Node 才会被考虑用来运行这个 Pod。apiVersion: v1 kind: Pod metadata: name: my-pod spec: containers: - name: my-container image: my-image nodeSelector: disktype: "ssd" #←这就是一例子，本质上就是tag-key: value指定节点标签的Pod，这些Pod会尽可能的在满足条件的情况下给予这些Node节点使用，如果节点不满足条件，Pod会进入pending状态。直到节点满足条件为止。亲和性调度这里其实不止 Pod 方面，Node 也有，所以就一起讲了。但本质上都是在 Pod 上进行配置的。亲和性分为：①Node亲和性、②Pod亲和性亲和性策略：①软策略、②硬策略Node节点亲和性（Node Affinity）Preferred During Scheduling Ignored During Execution：软策略这意味着这些亲和性规则在调度时会被考虑，以帮助调度器做出决策，但一旦 Pod 被调度并运行，这些规则就不再被考虑。如果在 Pod 运行期间，节点不再满足这些偏好规则，Pod 将继续在该节点上运行，不会被迁移。举个简单的例子：选择 Pod 节点时，声明了 Pod 最好能部署在Node①。如果是软策略，他会尽量满足这个条件，不一定会完全部署在Node①节点上。apiVersion: v1 kind: Pod metadata: name: my-pod spec: containers: - name: my-container image: my-image affinity: nodeAffinity: preferredDuringSchedulingIgnoredDuringExecution: - preference: matchExpressions: - key: disktype operator: In values: - ssd weight: 1preference: 定义了节点选择的偏好条件。这里使用 matchExpressions 来指定节点必须具有 disktype 标签，且其值为 ssd。当调度器在集群中寻找合适的节点来调度 Pod 时，它会首先考虑所有必需的亲和性规则（如果有的话）。然后，它会根据 preferredDuringSchedulingIgnoredDuringExecution 中定义的偏好规则来选择最佳节点。如果多个节点都满足必需的亲和性规则，调度器会根据这些偏好规则的权重来决定将 Pod 调度到哪个节点上。required During Scheduling Ignored During Execution：硬策略这意味着这些亲和性规则在调度时会被强制执行，如果节点不满足这些规则，Pod 将不会被调度到该节点上。然而，一旦 Pod 被调度并运行，即使节点不再满足这些规则，Pod 也会继续在该节点上运行，不会被迁移。举个简单的例子：选择 Pod 时，声明了 Pod 部署在Node①上。如果是硬策略，Pod 在满足硬策略的条件时，会被强制性要求部署在Node①上。apiVersion: v1 kind: Pod metadata: name: my-pod spec: containers: - name: my-container image: my-image affinity: nodeAffinity: requiredDuringSchedulingIgnoredDuringExecution: nodeSelectorTerms: - matchExpressions: - key: disktype #匹配策略 operator: In #节点信息内标签的需匹配的值 values: - ssd硬策略通过 matchExpressions 去搜索有相关 key 和相关 value 的 Pod，然后根据匹配策略 operator 去进行筛选匹配符合策略的节点。Pod亲和性（Pod Affinity）PodAffinity（Pod 亲和性）允许你指定一个 Pod 应该（或优选）被调度到与具有特定标签的 Pod 在同一节点或拓扑域上的规则。PodAntiAffinity（Pod 反亲和性）允许你指定一个 Pod 应该避免被调度到与具有特定标签的 Pod 在同一节点或拓扑域上的规则。PodAffinity 和 PodAntiAffinity 也有 requiredDuringSchedulingIgnoredDuringExecution 和 preferredDuringSchedulingIgnoredDuringExecution 两种形式，分别作为硬性和软性规则。硬性 Pod 亲和性规则要求调度器必须满足这些规则才能调度 Pod，而软性规则则是优选的，调度器会尝试满足，但不是必须的。亲和性实例：apiVersion: apps/v1 kind: Deployment metadata: name: nginx labels: app: nginx spec: replicas: 3 selector: matchLabels: app: nginx template: metadata: labels: app: nginx spec: containers: - name: nginx image: nginx:1.22 affinity: #选择Node/Pod亲和性 nodeAffinity: #选择亲和性规则 requiredDuringSchedulingIgnoredDuringExecution: #调度器根据设置的节点标签，选择哪个Node执行该亲和性规则 nodeSelectorTerms: - matchExpressions: #选择一个符合条件的节点 - key: tag3 #匹配策略 operator: In #节点信息内标签的需匹配的值 values: - tagVal1这里插入一个匹配策略的表格：匹配策略关键词意义解释In在选择的标签值在node节点上存在Notin不在选择的标签值不在node节点上存在Gt大于仅限Node有；要大于选择的标签值（只能比较整数）Lt小于仅限Node有；要小于选择的标签值（只能比较整数）Exists存在选择具有指定标签的对象（不考虑值，不能使用values字段）DoesNotExist不存在选择具有指定标签的对象（不考虑值，不能使用values字段）假设您有以下两个节点：节点 A，带有标签：label3: labelVal1节点 B，带有标签：label3: labelVal2根据上面的配置，节点需要带的标签得是key为label3，value为labelVal1，K8s就必定会选择拥有label3: labelVal1标签的节点，上面节点A符合条件，而节点B不符合，固然只选择节点A。相应的命令配置标签如果有需要的话，可以使用下面的命令进行配置节点标签：kubectl label nodes <node-name> <labelKey>=<labelValue>替换 <node-name> 为您的实际节点名称，<labelKey>为对应的键，<labelValue>为对应的键值。如果要删除一个 Pod 的标签的话，可以使用下面命令删除：kubectl label nodes <node-name> <labelKey>-如果要更新一个 Pod 的标签的话，可以使用下面命令更改：kubectl label nodes <node-name> memory=1000 --overwrite部署配置到 Kubernetes 集群使用以下命令将 Deployment 部署到您的 Kubernetes 集群：kubectl apply -f nginx-deployment.yaml验证 Pod 是否调度到正确的节点kubectl get pods -o wide #获取集群内所有Pod的详细信息不过一般直接用 kubectl get pod 就够了，直接看STATUS和AGE列的内容就能辨认出来 Pod 部署的是否正确。举个我实习中遇到的 Pod 实例：NAMEREADYSTATUSRESTARTSAGEIPNODENOMINATED NODEREADINESS GATESgitee-code-xxxxxx1/1Running0277dxx.xx.xxx.xxxx.xx.xxx.xxgitee-api-xxxxx0/1Error101（41d ago）7d19hxx.xx.xxx.xxxx.xx.xxx.xx如果是直接使用 kubectl get pod 的话，AGE 之后的 Pod 信息是没有的。上面的例子是硬策略，强制规则匹配，想换成软策略主要改的就是 requiredDuringSchedulingIgnoredDuringExecution 变成 preferredDuringSchedulingIgnoredDuringExecution ，然后后续的配置根据具体的关键词在进行细小的调整就行。另外补充一点K8s deployment 里面常见的字段：annotations：注解（Annotations）是一种附加于 Kubernetes 对象的非标识性信息。它们通常用于存储工具或库可以检索的额外信息，但不会用于识别和选择对象。注解可以包含构建信息、版本控制信息、监控和日志链接、调试信息等。例如，kubectl describe deployment 命令会显示 Deployment 的注解，如 deployment.kubernetes.io/revision 和 kubernetes.io/change-cause。这些注解有助于追踪 Deployment 的版本和变更原因。replicas：这个字段指定了期望运行的 Pod 副本数量。Deployment 控制器会确保始终有指定数量的 Pod 在运行。如果 Pod 副本数少于这个值，Deployment 控制器会创建新的 Pod 以补足数量；如果多于这个值，它会删除多余的 Pod。strategyType：这个字段定义了更新策略的类型，可以是 RollingUpdate 或 Recreate。RollingUpdate 是默认值，它允许逐步更新 Pod，而 Recreate 会先删除所有旧的 Pod，然后再创建新的 Pod。（如果想要服务不怎么受影响，能够动态更新，最好还是用 RollingUpdate 进行 Pod 的滚动更新比较好。）minReadySeconds：这是一个可选字段，用于指定新创建的 Pod 应该准备好并且不崩溃的最小秒数，才会被视为可用。这个字段的默认值是 0，意味着 Pod 一旦准备就绪就会被视为可用。通过设置这个值，可以减缓 Deployment 的滚动更新速度，确保在继续更新之前新 Pod 已经稳定运行一段时间。（如果不配置这个字段，默认行为就是一旦 Pod 准备好就立即被视为可用。）rollingUpdateStrategy：在使用 RollingUpdate 更新策略时，这个字段定义了更新过程中可以同时处于不可用状态的最大 Pod 数量（maxUnavailable）和可以超出期望 Pod 数量的最大 Pod 数量（maxSurge）。这些参数可以设置为绝对数量或百分比。默认情况下，这两个值都设置为 25%，意味着在更新过程中，可以有最多 25% 的 Pod 处于不可用状态，同时可以有最多 25% 的额外 Pod 被创建。pod template：这是定义 Pod 规格的模板，它指定了 Pod 的容器、卷、环境变量等。在 Deployment 中，Pod 模板定义了 Deployment 所管理的 Pod 的期望状态。当 Deployment 更新时，Pod 模板中的任何更改都会应用于新的 ReplicaSet 和随后创建的 Pod。综上，如果 Pod 使用软策略，一般来说都是多个软策略看权重实施，权重高的，则执行指定的软策略；如果 Pod 使用硬策略，一般来说就是指定标签去进行匹配实施，符合规则的，就执行指定的硬策略；当然也有软、硬策略一起实施的，先满足硬策略，再考虑软策略。若硬策略无法满足，软策略一个都不会执行。额外面试题：①你在部署pod的时候选择什么样的策略：根据node的亲和性：性能不一致，尽量把 Pod 往性能高的多部署，选择软策略；节点故障或者节点维护中，只能选择硬策略，把故障节点剔除。调度策略匹配标签操作符拓扑域调度目标Node的亲和性主机的标签In、NotIn、Exists、DoesNotExist、Gt、Lt不支持指定主机Pod的亲和性Pod的标签In、NotIn、Exists、DoesNotExist支持Pod和指定标签的Pod部署在同一个拓扑域Pod的反亲和性Pod的标签In、NotIn、Exists、DoesNotExist支持Pod和指定标签的Pod部署在不同拓扑域拓扑域：K8S集群节点当中的一个组织结构，可以根据节点的物理关系或者逻辑关系进行划分。可以用来标识节点之间的空间关系，网络关系，或者其他类型的关系。上面讲的拓扑域指的是Pod的标签，标签可以用来对Node/Pod进行逻辑关系上的划分。Pod亲和性里一般的拓扑域会用关键字段 topologyKey 进行标注出来，例如： topologyKey: test1 ←这里的拓扑域就是 test1 啦注意点:在配置 Pod 的亲和性策略时，必须加上拓扑域的关键字topologykey，指向的是节点标签。Pod 亲和性的策略分为硬策略和软策略。Pod 亲和性的NotIn可以替代反亲和性。②在部署的时候，该怎么考虑Node节点：部署 Node 节点的时候，需要考虑软硬策略的实施，以及污点和容忍度的配置，其中这里配置还可以和 Node 的亲和性一起使用。被设为污点的 Node，不会被部署 Pod，这点和反亲和性一样，脏了，就不会被 Pod 亲和，会被 Pod 远离。污点（taint）污点的种类：污点种类作用NoSchedule这个污点会阻止新的 Pod 被调度到该节点上，除非这些 Pod 明确地容忍（tolerate）这个污点。PreferNoSchedule这个污点的作用是建议调度器尽量不要将 Pod 调度到该节点上。如果调度器找不到其他合适的节点，它仍然可以调度 Pod 到这个节点。NoExecute这个污点不仅会阻止新的 Pod 被调度到该节点上，还会驱逐该节点上所有没有容忍（tolerate）这个污点的 Pod。NoExecute 驱逐细节：因为 NoExecute 类型而从节点上被驱逐出去的非容忍该污点的 Pod，对于由控制器（如 Deployment、StatefulSet、DaemonSet）管理的 Pod，它们会被控制器注意到并重新创建。这意味着这些 Pod 会由控制器在其他合适的节点上重新调度。但对于没有控制器管理的”孤儿” Pod（例如，直接使用 kubectl run 创建的 Pod），它们将被直接杀死，并且不会自动重新创建。注意点：节点服务器需要维护的，服务器如果关机了或者宕机了，节点上的业务将会全部失效。在工作中主要部署 Pod 的方式是控制器部署，尤其 deployment 控制器最多。所以 NoExecute 污点类型适用的其中一个场景类型就是维护升级场景。在这种场景下，节点再升级维护后，就可以将对应的驱逐(NoExecute)污点给去除了。一旦设置为驱逐，控制器创建的 Pod 会在其他节点重新部署。所以驱逐的业务场景主要在于业务维护和业务回收。但是系统集群组件不会被驱逐。污点相关命令#添加污点 kubectl taint nodes <node-name> KEY=VALUE:TAINT_EFFECT #删除污点 kubectl taint nodes <node-name> KEY:TAINT_EFFECT- #删除指定Node指定类型的污点 kubectl taint nodes <node-name> - #删除指定Node所有类型的污点 #查看污点 kubectl describe nodes <node-name> | grep -i taintsNODE_NAME 是您要添加污点的节点的名称。KEY 是污点的键。VALUE 是与键相关联的值（可选）。TAINT_EFFECT 是污点的效果，可以是 NoSchedule、PreferNoSchedule 或 NoExecute。容忍允许调度到有污点的节点：如果一个节点有污点，只有那些在其配置中声明了相应容忍的 Pod 才能被调度到该节点上。保持服务连续性：在节点维护或升级期间，通过设置容忍，可以确保关键的 Pod 继续在节点上运行，即使节点被添加了污点。管理 Pod 生命周期：对于 NoExecute 污点，容忍可以指定一个生命周期，在这个生命周期内，Pod 可以继续在节点上运行，但一旦超过这个时间，Pod 将被驱逐。注意： NoExecute 污点设置的容忍 Pod 必须指定生命周期。Pod 被驱逐之后是否重新创建取决于这些被驱逐的 Pod 是否被控制器管理。容忍污点Pod配置示例apiVersion: v1 kind: Pod metadata: name: my-pod spec: containers: - name: my-container image: my-image tolerations: #设置容忍 - key: "key1" operator: "Exists" #匹配规则：存在，不用设置value effect: "NoSchedule" #阻止类型 - key: "key2" operator: "Exists" #匹配规则：存在，不用设置value effect: "NoExecute" #驱逐类型 tolerationSeconds: 3600 # 容忍时间，单位为秒 生命周期在 Kubernetes 中，如果您想要创建一个容忍（Toleration），它不特定于任何特定的污点类型，您可以省略 effect 字段。这样，这个容忍将适用于所有可能的污点类型。#容忍所有污点类型示例： apiVersion: v1 kind: Pod metadata: name: my-pod spec: containers: - name: my-container image: my-image tolerations: - key: "my-key" operator: "Exists"应用场景这种类型的容忍通常用于以下场景：通用污点处理：需要确保 Pod 可以被调度到所有可能带有特定键的污点的节点上时。灵活的调度策略：在某些自动化系统中，可能会动态地向节点添加不同类型的污点，使用这种通用容忍可以确保 Pod 总是有资格被调度。注意事项使用这种通用容忍时要小心，因为它可能会使 Pod 被调度到不希望它们运行的节点上。确保这种容忍的使用符合应用程序的部署策略和集群的运维策略。这种配置提供了高度的灵活性，但也需要谨慎使用，以避免潜在的调度问题。cordon和draincordon：将 Node 标记为不可用状态。#标记命令 kubectl cordon <node-name> #取消标记命令 kubectl uncordon <node-name>被标记了 cordon 的节点不可以部署。cordon 使用的场景和驱逐污点 noExecute 类似，可以用在节点升级维护的场景。注意：因为被 cordon 标记之后，对应的节点会不可部署，所以在节点升级维护完成之后，记得取消标记。drain：将该节点下的 Pod 全部转移到其他的 Node 节点下运行。一旦执行了drain，被执行该命令的 Node 就会变成不可调度状态，且该 Node 上的所有 Pod 都将会被驱逐。举例命令：kubectl drain <node-name> --ignore-daemonsets --delete-local-data --force命令关键字段功能--ignore-daemonsets忽视 daemonset 方式部署的 Pod，daemonset部署的 Pod 不会被转移（daemonset 要部署的一般是后台运行较为重要的 Pod，例如系统 Pod，所以不动）--delete-local-data有本地挂载的 Pod 会被强制杀死--force强制释放不是控制器管理的 Pod如果要取消 Node 的状态的话，使用kubectl uncordon <node-name>即可。注意：uncordon 操作不会自动重启被驱逐的 Pods，它只是允许新的 Pods 被调度到该节点上。如果需要，你需要手动重启之前被驱逐的 Pods。额外补充资源调整除了上述这种配置：apiVersion: apps/v1 kind: Deployment metadata: name: nginx labels: app: nginx spec: replicas: 3 selector: matchLabels: app: nginx template: metadata: labels: app: nginx spec: containers: - name: nginx image: nginx:1.22 affinity: #选择Node/Pod亲和性 nodeAffinity: #选择亲和性规则 requiredDuringSchedulingIgnoredDuringExecution: #调度器根据设置的节点标签，选择哪个Node执行该亲和性规则 nodeSelectorTerms: - matchExpressions: #选择一个符合条件的节点 - key: tag3 #匹配策略 operator: In #节点信息内标签的需匹配的值 values: - tagVal1通常实际应用场景上还会在 containers 下加上 request 和 limit：Containers: code-go: Image: docker-hub.xxx/xxx-xxx/xxx-xxx:v8.19.0 Port: 3000/TCP Host Port: 0/TCP Limits: cpu: 4 memory: 6Gi Requests: cpu: 100m memory: 256Mi Environment: <none> Mounts: /app/config.yaml from xxx (rw,path="config.yaml") /app/configs/xxx_setting.yaml from xxx-xxx-setting (rw,path="xxx_setting.yaml")Reques ：容器使用的最小资源需求，作为容器调度时资源分配的判断依赖。只有当节点上可分配资源量 >= 容器资源请求数时才允许将容器调度到该节点。但 Request 参数不限制容器的最大可使用资源。Limit：容器能使用资源的资源的最大值，设置为0表示使用资源无上限。Request 能够保证 Pod 有足够的资源来运行，而 Limit 则是防止某个 Pod 无限制地使用资源，导致其他 Pod 崩溃。两者之间必须满足关系: 0 <= Request <= Limit <= Infinity如果 Limit 为0表示不对资源进行限制，这时可以小于 Request。实际场景在 K8s 集群中部署服务时不设置 CPU requests 或将 CPU requests 设置得过低（这样"看上去"就可以在每个节点上容纳更多 Pod ）。在业务比较繁忙的时候，节点的 CPU 全负荷运行。业务延迟明显增加，有时甚至机器会莫名其妙地进入 CPU 软死锁等“假死”状态。其原因就在于在业务繁忙的情况，节点的 CPU 全负荷运行下， CPU requests 设置得过低得 Pod 的资源会被其他高设置的 Pod 削减能使用的资源量，直至到 requests 设置的最低资源请求点。因为 Pod 对应的服务所能使用的 CPU 资源少，导致业务服务延迟明显增加。类似地，部署服务时，不设置内存 requests 或者内存 requests 设置得过低，这时会发现有些 Pod 会不断地失败重启。而不断重启的这些 Pod 通常是遭遇了OOM。被不断的OOM导致服务重启其主要的原因在于 Pod 所占用的内存经常超过了设置的内存占用量，导致K8S觉得你这个 Pod 不正常，就让操作系统把你 OOMKilled 了。重启又因为设置的量太少超过了占用量就又重复上述情况。相反如果给你的应用设置较高的 request 资源需求限制，而实际占用资源长期远小于它的 request 资源需求限制，则会导致节点整体的资源利用率较低。当然这对时延非常敏感的业务除外，因为敏感的业务本身不期望节点利用率过高，影响网络包收发速度。但是对于一些非核心、资源不长期占用的应用，可以适当减少 request 资源需求限制，以提高资源利用率。驱逐节点优先级根据前面的调度策略我们了解到，Pod 在资源不足的时候，会根据调度策略决定 Pod 的优先级。当一些节点（一般是核心业务节点）资源不足时，会触发自动驱逐，将一些低优先级的 Pod 删除掉以释放资源让节点自愈。在没有设置 request，limit 的 Pod 会被调度策略认为优先级最低，容易被驱逐；因为没有限制，相当于这个 Pod 有多少资源都无所谓，放养。request 不等于 limit 的 Pod 其次； request 等于 limit 的 Pod 优先级较高，不容易被驱逐。所以如果是重要的线上应用，不希望在节点故障时被驱逐导致线上业务受影响，就建议将 request 和 limit 设成一致，也就是pod中的每个容器都必须指定limits.memory=requests.memory，limits.cpu=limits.cpu。尽量避免使用过大的 Request 与 Limit如果你的服务使用单副本或者少量副本，给很大的 request 与 limit，让它分配到足够多的资源来支撑业务，那么某个副本故障对业务带来的影响可能就比较大，并且由于 request 较大，当集群内资源分配比较碎片化，如果这个 Pod 所在节点挂了，其它节点又没有一个有足够的剩余可分配资源能够满足这个 Pod 的 request 时，这个 Pod 就无法实现漂移，也就不能自愈，加重对业务的影响。相反，建议尽量减小 request 与 limit，通过增加副本的方式来对你的服务支撑能力进行水平扩容，让你的系统更加灵活可靠。总结上面的话：与其让 Pod 多拿资源防止它崩溃，不如让它拿适当的资源去进行多副本部署。这样既不会发生资源不够导致 Pod 启动不了，也不会因为 Pod 服务没启动起来业务受到影响。多副本一个挂了其他能顶上，资源占用也不算多，够了就重启再顶回去。
腾讯csig运维实习生面经面试岗位：运维实习生 150~200一天 腾讯，针不戳先说体验吧，整体感觉真的很好，面试官老哥人好，沟通也很流畅，问问题讲的也清楚，生怕你没听懂还可以多讲几遍给你听。面试主要问的Go语言的内容比较多，或者说问的都是Go相关的，前面问的都挺简单的：①常量与变量的区别②GC回收（老生常谈）③反射与断言（这个我没答出来，我拉了，用少了也看少了）但是后面叫我根据场景写需求的时候我就懵了，咱第一次被这样面，咱现在代码水平也比较拉胯，所以铁寄了：场景：爬虫爬网站，用channel去接收与传出结果，要求爬失败了的网站该怎么用goroutine去解决。这个场景其实挺简单的，但奈何自己太久没用channel和goroutine，没怎么写过相应的代码，拉垮了，写不出一点。之后就是聊天环节了，老哥也是一眼发现了我的问题，让我去好好地把这些基础打好，然后看看k8s的源码或者大佬的代码，去理解他们代码的设计与思路，之后就可以尝试一下二次开发。总结：这次虽然好好地介绍了自己，但因为自己基础能力综合运用真的不行，Golang特性也没有好好学好。通过面试官老哥的指点也让我明白了之后的方向该怎么走，至于算法，连写的机会都没有，更得提升。这次面试体验真的很好，算是遇到了贵人指点，过程老哥也知道我自学Go方向有点差错，指点了我。我估计这辈子遇不到多几个这样的贵人了。希望之后沉淀完能真的达到老哥说的境界，这样对自己也好，对之后自己拿对应岗位的薪资也有支柱能撑得起自己。最后附上面试时候场景题的解答，确实不难：package main import ( "fmt" "net/http" "sync" "time" ) var waitGroup sync.WaitGroup func main() { ch := make(chan string) waitGroup.Add(1) //开一个协程 go func() { defer waitGroup.Done() res, err := http.Get("https://rolingg.top") if err != nil { ch <- fmt.Sprintf("爬虫爬网站出错, err: %s", err.Error()) } else { defer res.Body.Close() // 确保关闭响应体 ch <- fmt.Sprintf("爬虫爬网站成功！status: %s\n", res.Status) } }() //注意了，这里default获取结果要比case快的，因为协程中运行，肯定要比主线程慢一点，而且channel赋值要时间，就那一丢丢时间就足以让default先符合条件触发 //select { //case result := <-ch: // fmt.Println(result) //case <-time.After(3 * time.Second): // fmt.Println("爬虫爬取网站超时") // return //default: // fmt.Println("奇奇怪怪") //} //注意，select在获得到结果之前也会阻塞，直至有一个条件符合才会往下走 select { case result := <-ch: fmt.Println(result) case <-time.After(3 * time.Second): fmt.Println("爬虫爬取网站超时") return } close(ch) }