0. 计划前提学习目标本计划目标不是“泛泛了解大模型”，而是在 6 个月内形成可以落地的工程能力：理解机器学习、神经网络、Transformer、大模型推理的核心机制。能用 PyTorch 实现基础训练流程、Attention、Transformer Block 和 mini GPT。能做 Prompt、Tool Calling、RAG、Agent、评估等 LLM 应用工程。掌握 Spark、Kafka、Flink、ClickHouse 在 LLM 数据管道中的典型用法。能完成一个“大数据 + LLM 知识库与日志评估平台”综合项目。能做基础 LoRA 微调、评估、性能优化和项目文档整理。时间投入建议最低时间投入：工作日：每天 3 小时。非工作日：每天 8 小时（可弹性延至 10 小时）。每周合计：31 小时（工作日 15h + 周末 16h）。6 个月约 24 周，总投入约 744 小时。如果某周工作压力较大，最低不能低于 20 小时；低于 20 小时的周，优先在下周周末弹性时间内补回核心项目实践（周末额外的 2h/天是天然的追进度缓冲区，不要在正常周挤占它）。学习原则每周必须有代码产出。每两周必须有一个可运行 demo。每个月必须有一个可展示成果。数学不脱离代码学习，每个数学概念都要用 NumPy 或 PyTorch 实现一次。大模型应用不只调 API，要同时记录数据流、失败案例、日志和评估指标。大数据不只学组件概念，要和 LLM 文档处理、日志分析、评估闭环结合。每周固定节奏工作日：理论学习 + 代码实现 + 复盘练习，三件事每天都要有。周六：理论补强 + 项目编码为主，提交代码，写 README 片段。周日：复盘本周知识点，继续实现项目，补测试整理结果，写周总结和下周计划。每周输出模板每周结束时写一份简短总结，建议包含：本周主题： 本周完成： 核心概念： 代码产出： 踩坑记录： 仍不理解的问题： 下周计划：1. 总体路线第 1 月：数学 + 机器学习 + PyTorch 第 2 月：Transformer / LLM 原理 第 3 月：LLM 应用工程 第 4 月：大数据工程 第 5 月：大数据 + LLM 综合项目 第 6 月：微调 + 性能优化 + 作品整理最终产出：project-01-numpy-ml：NumPy 逻辑回归与数学基础实现。project-02-pytorch-template：通用 PyTorch 训练模板。project-03-mini-gpt：mini GPT 训练与生成。project-04-rag-agent：RAG + Tool Calling + Agent demo。project-05-bigdata-llm-pipeline：Spark / Kafka / ClickHouse 日志与文档管道。project-06-llm-bigdata-platform：综合项目。project-07-lora-eval-optimization：LoRA 微调、评估和优化报告。2. 第 1 月：数学 + 机器学习 + PyTorch目标：补齐大模型所需最小数学基础。能用 NumPy 理解模型训练的底层计算。能用 PyTorch 写出可复用训练模板。理解 loss、梯度、optimizer、训练循环和评估指标。第 1 周：线性代数 + 逻辑回归本周目标理解向量、矩阵、点积、矩阵乘法。理解 sigmoid、二分类、逻辑回归。用 NumPy 实现一个二分类模型。必学概念标量、向量、矩阵、张量。向量点积。矩阵乘法。shape 推导。sigmoid 函数。binary cross entropy。梯度下降的直观含义。每日安排周一：- 学习向量、矩阵、点积、矩阵乘法。- 用 NumPy 写矩阵乘法、点积、向量范数。- 整理 shape 笔记，写 10 个 shape 推导例子。周二：- 学习 sigmoid、二分类概率输出。- 写 sigmoid、预测函数 y_hat = sigmoid(X @ w + b)。- 构造 toy dataset，跑一次前向预测。周三：- 学习 binary cross entropy。- NumPy 实现 BCE loss。- 手算一个样本的 loss，和代码结果对比。周四：学习梯度下降。推导逻辑回归参数梯度。实现 dw、db 更新。周五：复盘本周数学。把逻辑回归训练流程串起来。写周总结和问题清单。周六：补线性代数练习，重点看 shape。完成 NumPy 逻辑回归完整训练脚本。画训练 loss 曲线。周日：复盘逻辑回归。添加 train / validation split 和 accuracy。整理 README。总结“矩阵乘法在模型里的作用”。产出NumPy 逻辑回归训练脚本。一份 shape 推导笔记。一张 loss 曲线。验收标准能解释 X @ w + b 的每个 shape。能解释 sigmoid 输出为什么可以当概率。能解释 loss 下降意味着什么。能不用框架训练一个简单二分类模型。第 2 周：概率、Softmax、交叉熵与评估指标本周目标理解概率分布、softmax、cross entropy。理解多分类模型输出。掌握常见评估指标。必学概念概率分布。条件概率。log 概率。softmax。cross entropy。negative log likelihood。accuracy、precision、recall、F1、AUC。每日安排周一：学习概率分布和 log。手写 softmax。测试不同 logits 对 softmax 的影响。周二：学习 cross entropy。手写多分类交叉熵。对比 one-hot label 和 class index label。周三：学习 maximum likelihood 直觉。解释为什么最小化 CE 等价于提高正确 token 概率。写一页笔记：CE 在分类和语言模型中的作用。周四：学习 precision、recall、F1。写混淆矩阵计算代码。构造不平衡数据，观察 accuracy 的问题。周五：学习 AUC、ROC 直觉。给第 1 周模型加评估指标。写周总结。周六：复习 softmax + CE。实现一个 NumPy 多分类模型。整理指标对比表。周日：复盘概率与 CE。完成评估脚本。写 README。整理错题和易混点。产出softmax 实现。cross entropy 实现。多分类评估脚本。验收标准能解释 softmax 为什么输出和为 1。能解释 CE 为什么常用于分类和 next token prediction。能在业务场景中选择 precision / recall / F1。第 3 周：PyTorch 基础与训练模板本周目标掌握 PyTorch Tensor、autograd、Dataset、DataLoader。写出通用训练模板。必学概念Tensor。requires_grad。autograd。loss.backward。optimizer.step。Dataset。DataLoader。batch training。每日安排周一：学习 Tensor 基础操作。用 Tensor 重写 NumPy 计算。整理 Tensor 与 ndarray 差异。周二：学习 autograd。写简单函数自动求导。手算梯度并对比 autograd。周三：学习 nn.Module。实现 LogisticRegression Module。用 optimizer 训练。周四：学习 Dataset / DataLoader。封装自定义 Dataset。把训练改成 batch。周五：学习训练模板结构。整理 train/evaluate/predict 函数。写周总结。周六：补 PyTorch API。完成通用训练模板。加日志输出。周日：复盘训练循环。加 checkpoint 保存和加载。整理 README。列出模板可复用点。产出PyTorch 二分类训练模板。Dataset / DataLoader 示例。checkpoint 保存与恢复。验收标准能解释每次训练循环发生了什么。能独立写出 zero_grad -> forward -> loss -> backward -> step。能用 DataLoader 训练模型。第 4 周：MLP、正则化、LayerNorm 与文本分类本周目标理解 MLP、激活函数、dropout、LayerNorm。完成一个文本或表格分类项目。必学概念MLP。ReLU / GELU。dropout。normalization。overfitting。early stopping。learning rate。每日安排周一：学习 MLP。实现 MLP classifier。观察隐藏层维度对效果的影响。周二：学习激活函数。对比 ReLU / GELU。写实验记录。周三：学习 dropout 和过拟合。给模型加 dropout。观察训练集和验证集曲线。周四：学习 LayerNorm。用 PyTorch 调用 LayerNorm。预习 Transformer 中 LayerNorm 的作用。周五：学习 early stopping。给训练模板加 early stopping。写月度总结。周六：准备文本或表格分类数据。完成分类项目训练。记录指标。周日：复盘第 1 月。整理项目代码。写 README。列出第 2 月 Transformer 前置问题。产出一个文本或表格分类项目。可复用 PyTorch 训练模板。第 1 月学习总结。验收标准能解释 overfitting 和正则化。能用 PyTorch 完成一个完整训练任务。能读懂简单模型的训练日志。3. 第 2 月：Transformer / LLM 原理目标：理解 Attention、Transformer Block、Causal LM。能训练一个 mini GPT。能理解大模型推理阶段的 prefill、decode、sampling。第 5 周：Self-Attention本周目标理解 Q、K、V。理解 scaled dot-product attention。理解 mask 的作用。用 PyTorch 实现单头 attention。必学概念token embedding。query、key、value。attention score。scale by sqrt(dk)。softmax attention weight。causal mask。每日安排周一：学习 Q/K/V 的直觉。写 Q = X @ Wq、K = X @ Wk、V = X @ Wv。整理 Q/K/V shape。周二：学习 attention score。实现 scores = Q @ K.T / sqrt(dk)。可视化 attention score。周三：学习 softmax attention。实现 weights @ V。对比不同 token 的注意力分布。周四：学习 causal mask。实现上三角 mask。解释为什么语言模型不能看未来 token。周五：复盘 self-attention。封装 Attention Module。写周总结。周六：推导 attention shape。完成单头 attention 实现。写测试。周日：复盘 Q/K/V。实现 causal self-attention。整理 README。记录不理解的问题。产出PyTorch Attention 实现。attention shape 笔记。causal mask demo。验收标准能解释为什么是 Q @ K.T。能解释 mask 的作用。能从 shape 判断 attention 代码是否正确。第 6 周：Multi-Head Attention 与 Transformer Block本周目标理解多头注意力。理解 residual、LayerNorm、FFN。实现 Transformer Block。必学概念multi-head attention。concat heads。output projection。residual connection。LayerNorm。feed-forward network。pre-norm / post-norm。每日安排周一：学习多头注意力。实现 head split 和 concat。整理 shape。周二：学习 output projection。完成 MultiHeadAttention。写单元测试。周三：学习 residual。把 residual 加到模块里。解释 residual 为什么有利于训练。周四：学习 LayerNorm。实现 Transformer Block 前半部分。对比 LayerNorm 输入输出。周五：学习 FFN。完成 Transformer Block。写周总结。周六：复盘 MHA。完成 Transformer Block 代码。测试不同 batch/seq/hidden。周日：复盘 residual + LayerNorm。整理模块化代码。写 README。预习 Causal LM。产出MultiHeadAttention。TransformerBlock。shape 测试脚本。验收标准能解释一个 Transformer Block 的数据流。能说明 residual、LayerNorm、FFN 各自作用。能独立实现可运行 Transformer Block。第 7 周：Tokenizer、Causal LM 与训练本周目标理解语言模型训练目标。构造字符级或简单 BPE 级数据集。训练一个小型 causal LM。必学概念tokenizer。vocabulary。token id。next token prediction。causal language model。cross entropy for token prediction。perplexity。每日安排周一：学习 tokenizer。实现字符级 tokenizer。整理 encode/decode 流程。周二：学习 next token prediction。构造 input/target。检查 shift 是否正确。周三：学习 embedding。实现 token embedding + position embedding。测试 embedding 输出 shape。周四：学习 LM head。实现 logits 输出。计算 token cross entropy。周五：学习 perplexity。跑一次小训练。写周总结。周六：准备小文本数据。训练字符级语言模型。记录 loss 曲线。周日：复盘 Causal LM。调 batch size、lr、context length。生成样例文本。写 README。产出字符级 tokenizer。Causal LM 训练脚本。生成文本样例。验收标准能解释输入和 target 为什么错开一位。能解释 logits shape。能说明 perplexity 和 loss 的关系。第 8 周：Mini GPT、推理与采样本周目标完成 mini GPT。理解 prefill、decode、KV cache 概念。实现 temperature、top-k、top-p 采样。必学概念GPT block。autoregressive generation。prefill。decode。KV cache。temperature。top-k。top-p。每日安排周一：组装 mini GPT。跑通 forward。检查 logits shape。周二：学习自回归生成。实现 greedy decode。记录生成结果。周三：学习 temperature。实现 temperature sampling。对比不同 temperature。周四：学习 top-k / top-p。实现 top-k。选做 top-p。周五：学习 KV cache 概念。写 KV cache 原理笔记。写月度总结。周六：整理 mini GPT 代码。训练并生成结果。记录实验参数。周日：复盘第 2 月。完善 README 和架构图。整理 Transformer 问答。预习 LLM 应用工程。产出mini GPT 项目。文本生成脚本。Transformer 原理笔记。验收标准能从 embedding 到 logits 讲完整 GPT 前向过程。能解释 temperature / top-k / top-p。能说明 KV cache 为什么能加速 decode。4. 第 3 月：LLM 应用工程目标：能构建 LLM API 应用。能做结构化输出、工具调用、RAG、Agent。能做基本评估和 bad case 分析。第 9 周：Prompt、结构化输出与 API 使用本周目标掌握常见 Prompt 结构。能要求模型输出稳定 JSON。能写 LLM API wrapper。必学概念system / user / assistant。few-shot。chain-of-thought 的使用边界。structured output。JSON schema。retry / timeout。token usage。每日安排周一：学习消息格式。写 API wrapper。整理调用日志字段。周二：学习 system prompt。写 5 个不同任务 prompt。比较输出差异。周三：学习 few-shot。给分类任务加 few-shot。记录成功和失败样例。周四：学习 structured output。要求模型输出 JSON。写 JSON parse + retry。周五：学习 token 成本。记录 prompt/input/output tokens。写周总结。周六：整理 Prompt 模板。完成结构化抽取 demo。加错误重试。周日：复盘 Prompt。做一个合同/工单/日志结构化抽取小项目。写 README。整理 bad cases。产出LLM API wrapper。JSON 输出 demo。结构化抽取小项目。验收标准能设计 system prompt。能处理 JSON 输出失败。能记录 token 和请求日志。第 10 周：Tool Calling / Function Calling本周目标理解工具调用机制。能定义工具 schema。能完成多轮工具调用。必学概念tool schema。arguments schema。tool_call。tool result。multi-step tool call。tool error handling。每日安排周一：学习工具调用流程。定义天气查询工具。跑通一次工具调用。周二：学习参数 schema。定义 SQL 查询工具。限制 SQL 只读。周三：学习工具结果回传。完成 tool result 到最终回答。记录错误样例。周四：学习多工具选择。加入文件查询工具。实现工具路由。周五：学习工具异常处理。给工具加 timeout / retry。写周总结。周六：整理工具调用框架。实现 SQL + 文件 + HTTP 三工具 demo。写测试样例。周日：复盘工具调用。完成多轮工具调用 demo。写 README。整理安全边界。产出Tool calling demo。SQL / 文件 / HTTP 工具。工具错误处理。验收标准能解释模型本身不执行工具，工具由外部 runtime 执行。能定义清晰工具 schema。能处理工具失败和参数错误。第 11 周：RAG 基础本周目标构建本地知识库问答。理解 chunk、embedding、向量检索、rerank。必学概念document loader。chunking。embedding。vector index。top-k retrieval。rerank。citation。hallucination。每日安排周一：学习 RAG 流程。加载 Markdown / PDF / txt。整理文档元数据。周二：学习 chunk 策略。实现按标题/长度切分。比较 chunk size。周三：学习 embedding。生成向量。存入本地向量库。周四：学习检索。实现 top-k 检索。输出引用来源。周五：学习 prompt 拼接。完成问答链路。写周总结。周六：整理知识库数据。完成 RAG API。记录检索日志。周日：复盘 RAG。加入 rerank 或简单重排。写 README。整理 bad cases。产出本地 RAG 知识库。检索日志。bad case 文件。验收标准能解释 chunk size 对召回的影响。能展示检索到的原文引用。能说明回答错误是检索错还是生成错。第 12 周：Agent 与基础评估本周目标构建简单多工具 Agent。建立 LLM 应用评估意识。必学概念planning。tool routing。memory。state。error recovery。LLM-as-judge。retrieval recall。answer correctness。每日安排周一：学习 Agent 基本结构。实现任务状态对象。串联工具调用。周二：学习 planning。让模型生成简单 plan。执行 plan 中工具。周三：学习 memory。加入短期对话记忆。测试多轮问题。周四：学习评估指标。写 RAG 评估数据集。实现简单召回评估。周五：学习 LLM-as-judge。实现答案评分脚本。写月度总结。周六：整合 RAG + tools。完成 Agent demo。记录失败用例。周日：复盘第 3 月。完成评估脚本。整理 README。预习 Spark。产出多工具 Agent。RAG 评估脚本。LLM-as-judge demo。验收标准能解释 Agent 与普通工具调用的区别。能定位一次失败发生在计划、工具、检索还是生成。能用小评估集比较两个 RAG 配置。5. 第 4 月：大数据工程目标：掌握 Spark、Kafka、Flink、ClickHouse 的核心工程用法。能把 LLM 文档处理和日志分析接入大数据链路。第 13 周：Spark SQL / DataFrame本周目标掌握 Spark DataFrame 和 Spark SQL。用 Spark 做文档清洗和切分。必学概念Spark Driver / Executor。DataFrame。Spark SQL。partition。shuffle。join。Parquet。每日安排周一：学习 Spark 架构。启动本地 Spark。读取 CSV / JSON。周二：学习 DataFrame API。做 select/filter/groupBy。写数据清洗练习。周三：学习 Spark SQL。注册临时表并查询。对比 SQL 和 DataFrame。周四：学习 partition / shuffle。观察 explain plan。记录 join 触发 shuffle 的情况。周五：学习 Parquet。把清洗结果写 Parquet。写周总结。周六：准备文档数据。用 Spark 清洗文档。保存为 Parquet。周日：复盘 Spark。用 Spark 切分文档 chunk。写 README。记录性能指标。产出Spark 文档清洗任务。Spark chunk 生成任务。Parquet 输出。验收标准能解释 Driver / Executor。能写 Spark SQL 和 DataFrame 代码。能解释 shuffle 为什么慢。第 14 周：Kafka 日志管道本周目标掌握 Kafka 基础。将 LLM 问答日志写入 Kafka。必学概念topic。partition。broker。producer。consumer。consumer group。offset。at-least-once。每日安排周一：学习 Kafka 架构。本地启动 Kafka。创建 topic。周二：学习 producer。写 Python producer。发送测试消息。周三：学习 consumer group。写 Python consumer。观察 offset。周四：学习 partition。按 user_id 分区。测试并发消费。周五：学习消费语义。记录失败重试策略。写周总结。周六：设计 LLM 日志 schema。把 RAG 请求日志写入 Kafka。写 producer 封装。周日：复盘 Kafka。写 consumer 入库到文件或数据库。整理 README。记录积压排查命令。产出Kafka producer / consumer。LLM 问答日志 schema。日志采集 demo。验收标准能解释 consumer group 和 partition 的关系。能说明 offset 如何提交。能处理 producer / consumer 基础异常。第 15 周：Flink / 流处理本周目标理解流处理基本概念。实时统计 LLM 请求指标。必学概念stream。window。event time。processing time。watermark。state。checkpoint。每日安排周一：学习流处理概念。搭建 Flink 本地环境。跑 word count。周二：学习 window。实现滚动窗口统计。统计 QPS。周三：学习 event time / watermark。模拟乱序数据。记录 watermark 行为。周四：学习 state。实现按 session 或 user 统计。记录 state 大小问题。周五：学习 checkpoint。整理容错笔记。写周总结。周六：接入 Kafka 日志。实时统计 QPS、错误率、平均延迟。输出到文件或数据库。周日：复盘 Flink。完善指标统计。整理 README。记录状态和 checkpoint 问题。产出Flink 实时指标统计任务。QPS / error_rate / latency 指标。验收标准能解释 window 和 watermark。能说明 checkpoint 的作用。能从 Kafka 流实时计算指标。第 16 周：ClickHouse 日志分析本周目标掌握 ClickHouse 基础建模。存储和分析 LLM 问答日志。必学概念MergeTree。partition key。order by。primary key。materialized view。aggregation。TTL。每日安排周一：学习 ClickHouse 架构。创建本地表。插入测试数据。周二：学习 MergeTree。设计问答日志明细表。写建表 SQL。周三：学习分区和排序键。优化查询条件。跑 explain。周四：学习聚合查询。统计日维度指标。写看板 SQL。周五：学习物化视图。建立聚合表。写月度总结。周六：把 Kafka/Flink 结果写入 ClickHouse。完成日志明细表和聚合表。写查询脚本。周日：复盘第 4 月。完成基础分析看板 SQL。整理 README。预习综合项目。产出ClickHouse 日志明细表。聚合指标查询。物化视图或聚合表。验收标准能解释 MergeTree 排序键对查询的影响。能设计 LLM 日志表。能写常用指标分析 SQL。6. 第 5 月：大数据 + LLM 综合项目项目名称：企业知识库 RAG + 日志评估平台。目标：将第 3 月 LLM 应用和第 4 月大数据组件整合成一个完整系统。形成可演示、可解释、可优化的工程项目。系统架构原始文档 -> Spark 清洗 -> Spark chunk 生成 -> embedding 批处理 -> 向量库 -> RAG API -> LLM 回答 -> Kafka 问答日志 -> Flink/Spark 指标统计 -> ClickHouse 明细与聚合 -> 评估报告 / bad case 数据集第 17 周：离线文档处理与入库本周目标建立离线知识库构建流水线。完成文档清洗、chunk、embedding、向量入库。每日安排周一：设计数据目录结构。定义文档元数据 schema。整理项目 README 目录。周二：Spark 文档清洗。去除空文档、重复文档。记录清洗统计。周三：chunk 策略设计。实现 Spark chunk 生成。记录 chunk size 分布。周四：embedding 批处理设计。实现批量 embedding。处理失败重试。周五：向量库 schema。写入向量库。写周总结。周六：跑完整离线流水线。修复数据质量问题。记录指标。周日：复盘离线管道。补充 README 和流程图。整理数据统计表。列下周 RAG 服务任务。产出离线知识库构建脚本。向量库数据。文档处理统计报告。验收标准能从原始文档一键生成向量库。能统计文档数、chunk 数、失败数。能说明 chunk 策略选择原因。第 18 周：RAG API 服务本周目标构建可用 RAG 服务。支持检索、重排、回答、引用来源。每日安排周一：设计 API。实现检索接口。记录请求日志。周二：实现 prompt 拼接。调用 LLM。返回答案和引用。周三：加入 rerank。比较 rerank 前后结果。记录 bad case。周四：实现无答案判断。加入 fallback。写测试问题集。周五：完善 API 错误处理。记录 token 和 latency。写周总结。周六：端到端测试。修复 RAG 质量问题。整理日志字段。周日：复盘 RAG 服务。补充 README 和接口文档。整理 bad case。准备日志管道。产出RAG API 服务。引用来源输出。bad case 文件。验收标准能回答知识库问题。能展示引用来源。能区分检索失败和生成失败。第 19 周：日志管道与实时指标本周目标将 RAG 请求日志写入 Kafka。使用 Flink 或 Spark Streaming 统计实时指标。写入 ClickHouse。每日安排周一：设计日志 schema。RAG 服务写 Kafka。测试 producer。周二：写 Kafka consumer。落 ClickHouse 明细表。验证数据完整性。周三：实时指标设计。Flink 统计 QPS。统计错误率。周四：统计平均延迟。统计无答案率。写入聚合表。周五：整理查询 SQL。写看板查询。写周总结。周六：端到端压测。修复日志丢失或重复问题。记录指标。周日：复盘日志链路。完善 ClickHouse 查询。写 README。准备评估系统。产出Kafka 日志管道。ClickHouse 明细表和聚合表。实时指标统计任务。验收标准每次 RAG 请求有完整日志。能查询 QPS、错误率、延迟、无答案率。能解释日志链路的投递语义。第 20 周：评估系统与 Bad Case 闭环本周目标建立 RAG 评估集。区分检索评估和生成评估。形成 bad case 分析闭环。每日安排周一：设计评估集格式。整理 50-100 条问题。标注参考答案或参考文档。周二：实现检索评估。计算 recall@k。分析低召回问题。周三：实现生成评估。使用 LLM-as-judge。人工抽查。周四：整理 bad case 分类。按检索、重排、prompt、模型四类归因。提出优化方案。周五：跑一次优化前后对比。写第 5 月项目总结。整理下月优化计划。周六：完成评估脚本。写评估报告。整理图表。周日：复盘综合项目。完善项目 README。画系统架构图。准备第 6 月微调与优化。产出RAG 评估集。检索评估脚本。生成评估脚本。bad case 分析报告。验收标准能量化 RAG 效果。能指出主要失败原因。能提出下一步优化方向。7. 第 6 月：微调、优化与作品整理目标：学会基础 LoRA 微调。对综合项目做性能和质量优化。整理成可展示作品。第 21 周：LoRA / QLoRA 微调本周目标理解 SFT、LoRA、QLoRA。用小模型做一次 instruction tuning。必学概念SFT。instruction data。LoRA rank。alpha。target modules。QLoRA。PEFT。每日安排周一：学习 SFT。整理 instruction 数据格式。构造小数据集。周二：学习 LoRA 原理。安装 PEFT。跑通最小 LoRA demo。周三：学习 target modules。配置训练参数。开始训练。周四：学习 QLoRA 概念。记录显存/速度。保存 adapter。周五：评估微调模型。对比基础模型。写周总结。周六：整理训练数据。完成 LoRA 训练。记录指标。周日：复盘微调。写推理脚本。整理 README。记录适用边界。产出LoRA 微调 demo。instruction 数据样例。微调前后对比。验收标准能解释 LoRA 为什么参数更少。能完成一次可复现微调。能说明何时该用 RAG，何时该微调。第 22 周：性能优化本周目标优化 RAG 服务和离线管道。形成优化前后指标对比。优化方向embedding batch。向量索引参数。检索 top-k。rerank 数量。prompt token 控制。LLM 请求并发。cache。retry。ClickHouse 查询优化。每日安排周一：梳理当前性能瓶颈。记录 baseline 指标。制定优化列表。周二：优化 embedding batch。测离线处理速度。记录结果。周三：优化检索参数。比较 recall 和 latency。记录 trade-off。周四：优化缓存和并发。压测 API。记录 P50/P95。周五：优化 ClickHouse 查询。整理优化报告。写周总结。周六：端到端压测。修复性能问题。整理图表。周日：复盘优化。写优化报告。更新 README。准备作品整理。产出性能优化报告。优化前后指标表。压测脚本。验收标准有 baseline。有优化动作。有指标对比。能解释优化 trade-off。第 23 周：系统文档与作品整理本周目标把综合项目整理成可展示作品。补齐架构图、部署文档、数据流图。每日安排周一：整理项目目录。清理无用代码。补 README 目录。周二：画系统架构图。画数据流图。写模块说明。周三：写部署文档。写环境变量说明。写启动命令。周四：写 API 文档。写数据表说明。写评估说明。周五：整理 demo 脚本。准备展示问题集。写周总结。周六：完整演示跑通。修复文档和代码问题。录制或整理演示流程。周日：复盘作品。完善 README。整理技术亮点。准备最终复盘。产出完整 README。架构图。数据流图。部署文档。演示脚本。验收标准其他人能按 README 跑起来。能用 10 分钟讲清楚系统架构。能说明技术难点和优化点。第 24 周：最终复盘与补缺本周目标查漏补缺。形成最终知识体系和作品集。每日安排周一：复盘数学和 ML。复盘 PyTorch。整理易错点。周二：复盘 Transformer。复盘 LLM 推理。整理问答卡片。周三：复盘 RAG / Agent。复盘评估。整理 bad case 方法论。周四：复盘 Spark / Kafka / Flink / ClickHouse。整理大数据组件对比。写数据管道总结。周五：整理最终汇报。整理项目亮点。写最终总结。周六：最终项目演示。模拟技术讲解。修复最后问题。整理作品列表。周日：写 6 个月复盘。规划下一阶段。整理学习资料索引。备份代码和文档。产出最终复盘文档。项目作品集。技术问答清单。下一阶段计划。验收标准能讲清楚 mini GPT、RAG、大数据日志管道三个核心项目。能解释端到端系统的数据流。能说明下一步深入方向。8. 每日记录每天学习结束后记录以下内容：今天学了什么： 今天写了什么代码： 今天最重要的一个概念： 今天遇到的问题： 明天要做什么：9. 数学补充清单必须掌握矩阵乘法和 shape 推导。点积和余弦相似度。softmax。cross entropy。梯度下降。链式法则直觉。Adam 优化器直觉。不必一开始深入严格数学证明。特征值分解细节。信息论完整体系。优化理论收敛证明。复杂概率图模型。数学与 LLM 对应关系矩阵乘法 -> Linear layer / QKV projection 点积 -> Attention score 余弦相似度 -> Embedding 检索 softmax -> token 概率分布 cross entropy -> 语言模型 loss 梯度下降 -> 参数更新 链式法则 -> 反向传播10. 项目验收总表第 1 月验收NumPy 逻辑回归能跑通。PyTorch 训练模板能复用。能解释 loss 和梯度。第 2 月验收Attention 能独立实现。Transformer Block 能跑通。mini GPT 能训练并生成文本。第 3 月验收能做结构化输出。能做工具调用。能做 RAG。能做简单 Agent 和评估。第 4 月验收Spark 能清洗文档。Kafka 能收集日志。Flink 能实时统计指标。ClickHouse 能查询分析日志。第 5 月验收综合项目端到端跑通。有 RAG 服务。有日志链路。有评估报告。第 6 月验收有 LoRA 微调 demo。有性能优化报告。有完整 README 和架构图。有最终复盘。11. 风险与调整如果时间不够优先保留：PyTorch 训练模板 Transformer / mini GPT RAG Spark Kafka ClickHouse 综合项目可以压缩：Flink 深度 LoRA 深度 Agent 复杂度 top-p 手写实现 复杂看板如果数学卡住不要停在证明上，改成：先写代码 观察输入输出 再回头理解公式如果项目卡住拆成最小链路：单文档 -> 单 chunk -> 单 embedding -> 单检索 -> 单回答 -> 单日志先跑通最小闭环，再扩展。如果大数据环境搭建卡住大数据四个组件（Spark / Kafka / Flink / ClickHouse）统一用 Docker Compose 管理，这是默认策略，不是备用方案。在第 12 周末提前把环境跑通，第 4 月进入时直接写代码。不要在集群模式或原生安装上耗费时间，先理解数据流和 API，工程化后再考虑生产部署细节。12. 推荐学习资料类型不强制绑定具体课程，按资料类型选择：PyTorch 官方教程。Hugging Face Transformers 文档。Andrej Karpathy 的 Neural Networks / GPT 讲解。Spark 官方文档。Kafka 官方文档。ClickHouse 官方文档。RAG / LoRA 原论文或高质量解读。阅读论文优先级：1. Attention Is All You Need 2. GPT 系列概览 3. InstructGPT / RLHF 概览 4. LoRA 5. RAG不要在 6 个月内大量读论文。先做工程闭环，再补论文深度。13. 最终能力画像6 个月结束后，应达到：能从数学角度解释 Attention。能从工程角度解释 RAG。能从数据角度解释 embedding 管道。能从系统角度解释 Kafka / Spark / ClickHouse 在 LLM 应用中的位置。能独立排查 LLM 应用中的检索失败、生成失败、日志缺失、评估偏差。能写出一个可展示的大数据 + LLM 综合项目。
Temporal风控（十八）控制流与数据面分离设计方法
一、为什么很多风控系统越做越重很多风控系统一开始看起来很简单： 主流程负责调步骤，步骤里算出结果，最后返回决策。问题在于，随着业务扩张，主流程往往会逐渐承担越来越多内容：进度管理中间态存储错误摘要大结果拼装复用搜索回放状态最后系统会进入一个很危险的状态： 控制流和数据面不再分离。这意味着主流程既要负责“怎么跑”，又要负责“数据放哪、怎么读、如何复用”。 一旦走到这一步，系统几乎必然变重。二、什么是控制流，什么是数据面1. 控制流控制流负责回答：现在执行到哪一步下一步该执行什么哪些步骤可以并发失败如何传播超时和重试如何处理2. 数据面数据面负责回答：数据结果存在哪里特征如何共享模型结果如何复用中间态如何命名和清理这两者天然相关，但不应混杂。三、为什么控制流不适合承载大量中间态如果把大量中间态直接挂在控制流对象上，会迅速带来几个问题：状态体积膨胀序列化成本变高重放语义更脆弱进度查询被大量无关数据污染控制流最适合保存的是：阶段状态运行索引错误摘要关键进度元信息而不是海量计算结果。四、为什么数据面需要独立上下文系统中间态一旦从控制流中分离出来，就必须有自己的正式承载方式。 这就是上下文系统的意义。上下文系统应负责：命名空间路径组织读写语义搜索复用生命周期清理只有这样，控制流与数据面分离才不只是“把数据丢到别处”，而是真正的架构分层。五、分离之后，系统会获得什么1. 控制流更轻主流程可以专注于阶段推进和生命周期管理。2. 数据面更清晰所有中间态都有统一命名空间和清理语义。3. 回放更容易回放系统可以复用同一控制骨架，只替换数据面来源与执行适配。4. 控制面更容易建设控制面查询控制流状态，上下文治理面查询数据面状态，两者边界清晰。六、分离设计中的关键连接点是什么控制流与数据面虽然分层，但仍然必须通过若干稳定契约连接。这些契约通常包括：步骤身份命名空间身份节点路径约定结果写回协议如果没有这些连接点，分离就会变成断裂； 如果连接点过多，分离又会重新失效。七、常见反模式1. 把所有结果都挂在流程状态里这会让控制流越来越重。2. 数据面没有统一命名和清理语义这会让分离流于形式。3. 控制面同时读取两套不一致状态这会让系统观测变得混乱。4. 回放系统另起一套状态模型这会让分离设计失去复用价值。八、如何判断控制流与数据面是否真正分离1. 主流程是否仍持有大量业务结果如果是，说明分离不彻底。2. 中间态是否都能通过统一上下文系统访问如果不能，说明数据面没有真正建立。3. 控制面能否明确区分运行态与中间态治理如果不能，说明两者边界仍不清晰。九、总结：这层分离为什么这么重要控制流与数据面分离的价值，不只是“结构更好看”。 它真正决定了风控引擎是否具备：轻量可维护的流程骨架可治理的中间态系统可复用的回放模式可解释的控制面从工程角度看，这是复杂风控平台能否长期稳定演进的底层分水岭。 一旦控制流和数据面重新糊在一起，系统很快就会重新走回重流程、重状态、重耦合的老路。
Temporal风控（十六）步骤条件跳过机制设计方法
一、为什么条件跳步不能只是一个 if 语句当流程变复杂之后，总会出现某些步骤不是每次都要执行的情况。 最自然的处理方式是在业务代码里写一个条件判断：if 某个条件: return # 跳过本步骤这种写法在小系统里完全够用，但在平台化风控引擎中会带来三个问题。第一，跳步语义被淹没在业务逻辑里，执行层和控制面都无法识别"这是一次合法跳过"还是"节点没被调度"。 第二，平台无法统一度量跳步行为，无法知道某个步骤的跳过频率。 第三，回放系统不能区分"这次应该跳过"和"这次因为错误被跳过"。因此，步骤条件跳过需要成为流程层的正式语义，而不是各节点各自处理的内部细节。二、跳步语义属于步骤定义还是执行层很多团队会把跳步逻辑放在以下位置之一：业务节点内部工作流的主循环里外部配置开关这三种方式都不是最优解。放在业务节点内部，平台无感知；放在工作流主循环，业务逻辑污染控制流；依赖外部配置开关，缺乏上下文感知能力。更成熟的设计是：跳步逻辑作为步骤定义的一个字段携带，执行层在调度步骤前统一评估，评估结果明确记录为"跳过"。这样做的好处是：步骤是否跳过由步骤自身声明，执行层只负责询问并执行结果，两者分工清晰。三、skip 函数的本质：一个可注入的运行时谓词步骤上的 skip 字段，本质上是一个谓词函数：入参是当前执行上下文（包含已有的中间态和 risk_input）出参是 bool（是否跳过）既支持同步，也支持异步这个设计有几个关键好处。1. 策略与执行解耦skip 函数只负责判断，不执行任何步骤逻辑。 这样跳步条件可以随业务演进，而不影响步骤本身的执行路径。2. 感知当前上下文skip 函数能访问 Context，这意味着它可以读取前置步骤的结果来决定是否跳过。 这是纯外部配置开关无法做到的事情。3. 同步/异步统一接口执行层通过 inspect.iscoroutinefunction 判断 skip 函数是否为协程，统一调用。 对 skip 函数的编写方不需要感知调度器的运行模式。四、skip 评估的时机：为什么必须在 step_setup 之前执行层在调度一个步骤时，通常会经历以下几个阶段：检查是否跳过执行 step_setup（写入步骤级初始化状态）启动子工作流（data / feature / model / rule）skip 评估必须在 step_setup 之前。 原因很简单：如果先执行 setup 再判断要跳过，就会产生一个已初始化但从未被执行的步骤残态。 这不仅浪费资源，还会污染上下文，让后续步骤搜索到不完整的中间态。因此，评估顺序为：check_skip → skip 则直接记录并返回，不继续初始化；不 skip 则进入正常执行路径。五、skip 语义在 Activity 层还是 Workflow 层评估有两个可选位置：在 Workflow 内部直接判断封装成 Activity 调用两种方式各有优劣。在 Workflow 内部判断：延迟极低，但 Temporal 的工作流代码要求确定性，不能任意访问外部状态。封装成 Activity 调用：可以自由访问 Redis、数据库等外部状态，适合绝大多数业务场景下的跳步判断。因此，绝大多数场景应选择 Activity 调用。 skip 函数通常依赖 Context 去读取 Redis 中已有的中间态，这恰好是 Activity 最擅长的事情。六、skip 的对立面：什么情况下不能允许跳过跳步机制一旦存在，就必须明确：哪些步骤不允许跳过。通常以下类型的步骤不适合携带 skip 语义：全局初始化步骤（global_setup）写入最终结果的步骤（build_result）清理步骤（clean_all）审计写入步骤（dwh_action）这类步骤的本质是"生命周期保障"，不能因为业务条件判断而省略。 如果允许跳过，流程的完整性会被破坏。七、跳步记录与指标的设计跳步发生后，有两类信息需要被记录：1. 指标执行层应在判定跳过时明确记录一个"skipped"状态的 step 指标，而不是简单丢弃这次调用。 这样观测平台才能区分：step 从未被调度step 被调度但判定跳过step 被调度但执行失败这三种状态语义完全不同，混淆会严重干扰排障效率。2. 日志跳步时应输出明确的日志，携带 global_namespace、workflow_name、step_id，说明"判定跳过"。 这样在回放或复盘时，工程师能清楚地知道流程路径是否走了跳步分支。八、skip 函数与依赖上下文的关系skip 函数在访问 Context 时，通常需要知道自己能读取哪些命名空间的数据。 这意味着它的执行依赖于"前置步骤已写入了足够的中间态"。这引出一个重要约束：如果 skip 函数依赖某个步骤的输出来做判断，那么该步骤必须在当前步骤之前完成执行。这不是额外的人工约定，而应该通过步骤的 flow_dependency 来自然保证。 skip 函数的数据依赖与步骤的执行依赖是同一张图上的两面。九、常见反模式1. skip 写在业务节点内部这会让平台无法识别"合法跳步"与"执行失败"。2. skip 判断在 step_setup 之后这会产生已初始化但被放弃的步骤残态。3. skip 函数依赖不属于前置步骤的数据这会导致判断结果不稳定，在不同执行时序下得出不同结论。4. 没有指标记录，把跳步当作"什么都没发生"这会让观测体系看不到跳步路径，排障时出现"节点消失"的困惑。十、如何判断跳步机制设计是否成熟1. 平台能否区分"跳步"与"执行失败"如果不能，说明跳步尚未成为正式语义。2. skip 函数是否能感知前置步骤的上下文如果只能依赖静态配置，说明跳步逻辑还在外部配置层，没有接入执行运行时。3. 跳步行为是否被指标和日志完整覆盖如果没有，说明这条执行路径对观测系统是盲区。十一、总结：步骤条件跳过机制真正解决了什么步骤条件跳过不是一个微小的功能细节，而是流程平台从"总是执行"走向"按需执行"的关键语义设计。它真正解决的是：业务跳步条件与执行编排解耦跳步行为对平台可见、可解释、可度量流程路径不再只有"执行"和"失败"两种状态一旦这套机制就位，风控引擎的流程描述能力才算真正完整。
ElasticSearch 查询match all 查询匹配所有文档，_score 为 1GET policy_info/_search { "query": { "match_all": {} } }{ "took": 0, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 3556, "relation": "eq" }, "max_score": 1, "hits": [ { "_index": "policy_info", "_id": "5", "_score": 1, "_source": { ... ...boost参数可改变_score的值GET policy_info/_search { "query": { "match_all": {"boost": 1.2 } } }{ ... "max_score": 1.2, "hits": [ { "_index": "policy_info", "_id": "5", "_score": 1.2, "_source": { ... } } ... }match none查询，反转match all查询结果，一个文档也不匹配GET policy_info/_search { "query": { "match_none": {} } }{ "took": 0, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 0, "relation": "eq" }, "max_score": null, "hits": [] } }Term 查询term查询返回和提供词严格匹配的文档，不要对text字段使用term查询，使用GET policy_info/_mapping查看索引结构；参数：GET xxx/_search { "query": { "term": { "field": { "value": 12, "boost": 2, "case_insensitive": true } } } }field[Required, string]：查询字段value[Required， string]：提供的精准匹配的值，包括空格，大小写敏感；boost[Optional, float]：用于减少或增加查询的相关分数；默认为1；case_insensitive[Optional, bool]：是否大小写敏感，默认true；GET policy_info/_search { "query": { "term": { "id": { "value": 12, "boost": 2 } } } }{ "took": 5, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 1, "relation": "eq" }, "max_score": 2, "hits": [ { "_index": "policy_info", "_id": "12", "_score": 2, "_source": { "id": 12, "qdu_id": "bc58c81c7b2143819487053cb50f9acf", ... "create_time": "2024-09-06 16:15:41", "row_update_time": "2024-09-06 16:15:41" } } ] } }terms 查询返回和提供词严格匹配的文档，可提供多个值，类似SQL or；参数：GET xxx/_search { "query": { "terms": { "field": ['xxx', 'xxx'], "boost": 2 } } }field[Required, List[string]]：查询字段；boost[Optional, float]：用于减少或增加查询的相关分数；默认为1；GET policy_info/_search { "query": { "terms": { "id": [12, 32], "boost": 2 } } }{ "took": 1, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 2, "relation": "eq" }, "max_score": 2, "hits": [ { "_index": "policy_info", "_id": "12", "_score": 2, "_source": { "id": 12, "create_time": "2024-09-06 16:15:41", "row_update_time": "2024-09-06 16:15:41" } }, { "_index": "policy_info", "_id": "32", "_score": 2, "_source": { "id": 32, "create_time": "2024-09-06 16:15:47", "row_update_time": "2024-09-06 16:15:47" } } ] } }Range 查询查找指定范围内的文档；不能用来查text和keyword字段，查询日期时可用日期函数；参数：GET xxx/_search { "query": { "range": { "field": { "gte": "2024-09-06", "lte": "2024-09-06", "format": "yyyy-MM-dd", "time_zone": "+01:00", "relation": "CONTAINS" } } } }field[Required, object]：查询字段gt[Optional]：大于gte[Optional]：大于等于lt[Optional]：小于lte[Optional]：小于等于format[Optional, string]：在处理日期字段时，用来处理日期格式，不写默认使用mapping中的格式；relation[Optional, string]：范围查询如何匹配范围字段的值INTERSECTS[default]：只要查询的范围与文档的范围有任何重叠，文档就会被匹配；CONTAINS：当文档的范围完全包含查询的范围时，文档才会被匹配；WITHIN：当文档的范围完全被包含查询的范围内时，文档才会被匹配；time_zone[Optional, string]：国际协调时间（UTC）偏移量或者IANA时区；合法值+01:00，-09:00，Asia/Shanghaiboost[Optional, float]：用于减少或增加查询的相关分数；默认为1；GET policy_info/_search { "query": { "range": { "create_time": { "gte": "2024-09-06", "lte": "2024-09-06", "format": "yyyy-MM-dd", "time_zone": "+01:00", "relation": "CONTAINS" } } } }{ "took": 2, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 1407, "relation": "eq" }, "max_score": 1, "hits": [ { "_index": "policy_info", "_id": "5", "_score": 1, "_source": { "id": 5, ... "create_time": "2024-09-06 16:15:39", "row_update_time": "2024-09-06 16:15:39" } },IDs 查询主键查询参数：GET xxx/_search { "query": { "ids" : { "values" : ["1", "4", "100"] } } }value[Required， List[string]]：主键列表；GET policy_info/_search { "query": { "ids": { "values": [1] } } }{ "took": 1, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 1, "relation": "eq" }, "max_score": 1, "hits": [ { "_index": "policy_info", "_id": "1", "_score": 1, "_source": { "id": 1, ... "create_time": "2024-09-06 16:15:38", "row_update_time": "2024-09-06 16:15:38" } } ...Exists 查询exists 查询会匹配那些在指定字段上有值的文档；参数：GET xxx/_search { "query": { "exists": { "field": "xxx" } } }field[Required, string]：查询的字段如果json值是null或[]，则认为不存在；空字符串，包含null和另一个值的数组，自定义空值，则认为存在；Fuzzy 查询返回查询字段的值和提供值相似的文档，用莱文斯坦距离衡量；指两个字串之间，由一个转成另一个所需的最少编辑操作次数；允许的编辑操作包括：将一个字符替换成另一个字符；插入一个字符；删除一个字符；参数：GET xxx/_search { "query": { "fuzzy": { "field": { "value": "xxx", "fuzziness": 1, "max_expansions": 50, "prefix_length": 0, "transpositions": true, } }, } }field[Required, string]：查询字段value[Required，string]：提供的匹配的值；fuzziness[Optional，string]：最大可编辑距离；[Common options | Elasticsearch Guide [8.17] | Elastic](https://www.elastic.co/guide/en/elasticsearch/reference/current/common-options.html#fuzziness)明确指定允许的编辑距离；只能是 0，1，2；AUTO 是一种动态计算模糊度的方式，基于查询字符串的长度来确定允许的编辑距离；字符串长度为 0 到 2：不允许编辑（等同于 fuzziness: 0）；字符串长度为 3 到 5：允许一个编辑（等同于 fuzziness: 1）；字符串长度大于 5：允许两个编辑（等同于 fuzziness: 2）；max_expansions[Optional，integer]：模糊查询过程中生成的变体词语的最大数量；默认50；prefix_length[Optional，integer]：前缀长度，表示在进行模糊匹配之前，词语的前缀部分必须完全匹配，默认0；transpositions[Optional，Boolean]：是否允许字符换位，默认true；rewrite[Optional，string]： rewrite parameter.GET township_info_new/_search { "query": { "fuzzy": { "township_code": { "value": "5404242070000", "fuzziness": 2, "max_expansions": 50, "prefix_length": 3, "transpositions": true } } } }{ "took": 151, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 19, "relation": "eq" }, "max_score": 8.307324, "hits": [ { "_index": "township_info_new", "_id": "36004", "_score": 8.307324, "_source": { ... ...Prefix 查询返回字段匹配指定前缀的文档；参数：GET xxx/_search { "query": { "prefix": { "field": { "value": "xxx", "rewrite": "constant_score", "case_insensitive": true } } } }field[Required, string]：查询字段value[Required，string]：提供的匹配的值；rewrite[Optional，string]： rewrite parameter.case_insensitive[Optional, bool]：是否大小写敏感，默认true；GET policy_info/_search { "query": { "prefix": { "qdu_id": { "value": "552", "rewrite": "constant_score", "case_insensitive": true } } } }{ "took": 62, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 1, "relation": "eq" }, "max_score": 1, "hits": [ { "_index": "policy_info", "_id": "5", "_score": 1, "_source": { "id": 5, "qdu_id": "552d2db621224195b78a8868bff8f330", ... "create_time": "2024-09-06 16:15:39", "row_update_time": "2024-09-06 16:15:39" } } ] } }Regexp 查询返回与提供正则相匹配的文档；参数：GET xxx/_search { "query": { "regexp": { "field": { "value": "5.*0", "flags": "ALL", "case_insensitive": true, "max_determinized_states": 10000, "rewrite": "constant_score" } } } }field[Required, string]：查询字段；value[Required，string]：正则表达式；flags[Optional，string]：启用可选运算符；ALL：启用所有可选运算符；“”：ALL 的别名；COMPLEMENT：启用~运算符；EMPTY：启用#运算符；#运算符不匹配任何字符串，甚至不匹配空字符串；INTERVAL：启用<>运算符；使用<>来匹配数字范围；INTERSECTION：启用&运算符；类似AND运算符；如果左侧和右侧的模式都匹配，则匹配将成功；ANYSTRING：启用@运算符；匹配任何整个字符串NONE：禁用所有可选运算符；case_insensitive[Optional, bool]：是否大小写敏感，默认true；max_determinized_states[Optional，integer]：用于限制正则表达式查询在确定化过程中可以生成的状态数，以防止过于复杂的正则表达式导致性能问题或内存溢出；rewrite[Optional，string]： rewrite parameter；GET policy_info/_search { "query": { "regexp": { "qdu_id": { "value": "5.*0", "flags": "ALL", "case_insensitive": true, "max_determinized_states": 10000 } } } }{ "took": 6, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 10, "relation": "eq" },Wildcard 查询返回包含与通配符模式匹配的文档；参数：GET xxx/_search { "query": { "wildcard": { "field": { "value": "xxx", "boost": 1.0, "rewrite": "constant_score_blended" } } } }field[Required, string]：查询字段；value[Required，string]：通配符表达式；此参数支持两个通配符运算符：？，匹配任何单个字符；*，可以匹配零个或多个字符，包括一个空字符;boost[Optional, float]：用于减少或增加查询的相关分数；默认为1；case_insensitive[Optional, bool]：是否大小写敏感，默认true；rewrite[Optional，string]： rewrite parameter；wildcard[Optional，string]：value的别名，两个都提供，则使用最后一个；GET policy_info/_search { "query": { "wildcard": { "qdu_id": { "value": "*8888*", "boost": 1.0 } } } }{ "took": 2, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 2, "relation": "eq" }, "max_score": 1, "hits": [ ... Compound 查询Boolean 查询与其他查询的布尔组合匹配的文档匹配的查询；参数：get xxx/_search { "query":{ "bool": { "must": [ { "term": { "qds": { "value": "0941017" } } } ], "should": [ { "range": { "id": { "gte": 1, "lte": 200 } } } ], "filter": [ { "range": { "row_update_time": { "gte": "2024-09-19", "lte": "2024-09-20", "format": "yyyy-MM-dd" } } } ], "must_not": [ { "terms": { "qdu_id": [ "552d2db621224195b78a8868bff8f330", "8807bbf8db7b434ca8287209861962f3" ] } } ] } } }must：查询必须出现在文档中，会贡献得分；filter：查询必须出现在文档中，分数会被忽略；should：查询应该出现在匹配的文档中，出现会贡献得分；如果布尔查询中只有 should 子句，则至少有一个 should 子句必须匹配；否则，should 子句的数量必须至少为 minimum_should_match 参数指定的数量；must_not：查询不能出现在文档中；minimum_should_match：[minimum_should_match parameter | Elasticsearch Guide [8.17] | Elastic](https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl-minimum-should-match.html)boost：用于减少或增加查询的相关分数；默认为1；GET policy_info/_search { "query":{ "bool": { "must": [ { "term": { "qds": "0932009" } } ], "should": [ { "range": { "id": { "lte": 5 } } } ], "filter": [ { "range": { "row_update_time": { "gte": "2024-09-01", "lte": "2024-09-20", "format": "yyyy-MM-dd" } } } ], "must_not": [ { "range": { "id": { "gte": 100 } } } ] } } }{ "took": 1, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 65, "relation": "eq" }, "max_score": 5.050299, "hits": [ { ... } ...Boosting 查询指定一个正向查询和一个负向查询，并通过降低负向查询匹配的文档的分数来影响最终的相关性评分；参数：GET xxx/_search { "query": { "boosting": { "positive": {}, "negative": {} }, "negative_boost": 0.2 } } }positive[Required，query]：想要运行的查询，返回的文档必须和此查询匹配；negative[Required，query]：从返回的文档中进行匹配，减少匹配到的文档相关度评分；计算方式如下：获取positive查询中返回的相关度评分；评分和negative_boost相乘，得到最终得分；negative_boost[Required，float]：0-1中的值，用来减少相关度得分；GET policy_info/_search { "query": { "boosting": { "positive": { "term": { "qds": { "value": "0932009" } } }, "negative": { "range": { "id": { "gte": 2, "lte": 65 } } }, "negative_boost": 0.2 } } } { "took": 1, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 65, "relation": "eq" }, "max_score": 3.9069033, "hits": [ { "_index": "policy_info", "_id": "1", "_score": 3.9069033, "_source": { ... } }, { "_index": "policy_info", "_id": "5", "_score": 0.81005985, "_source": { ... } },Constant score 查询将查询的相关性评分设置为一个常量值；参数：GET xxx/_search { "query": { "constant_score": { "filter": {}, "boost": 1.2 } } }filter[Required，query]：返回匹配此查询的文档；boost：查询的相关分数；默认为1；GET policy_info/_search { "query": { "constant_score": { "filter": { "range": { "id": { "gte": 1, "lte": 2 } } }, "boost": 1.2 } } }{ "took": 0, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 2, "relation": "eq" }, "max_score": 1.2, "hits": [] ...Disjunction max 查询组合多个查询，并返回匹配文档中得分最高的那个查询的分数；参数：GET policy_info/_search { "query": { "dis_max": { "tie_breaker": 0.7, "boost": 1, "queries": [ {}, {} ] } } }tie_breaker[Optional，float]：$final\_score=maxscore + \sum_1^n score * tie\_breaker$ ，由于默认值为0，所以Elasticsearch会默认使用最高的相关性分数；querys[Required，query]：包含一个或多个查询子句；返回的文档必须与这些查询中的一个或多个匹配；如果文档匹配多个查询，Elasticsearch将使用最高的相关性分数；GET policy_info/_search { "query": { "dis_max": { "tie_breaker": 0.7, "boost": 1, "queries": [ { "range": { "id": { "gte": 1, "lte": 2 } } }, { "term": { "qdu_id": { "value": "b1e4ef6d8dba42b297c99c0b450eecab" } } } ] } } }{ "took": 10, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 3, "relation": "eq" }, "max_score": 6.6652584, "hits": [ ... ] ...Function score 查询 TODOFull Text 查询match 查询匹配搜索会比较搜索词和每个文档的相似度，只要搜索词能命中文档的分词就会被搜索到；参数：GET policy_info/_search { "query": { "match": { "field": { "query": "xxxxx", "analyzer": "standard", "auto_generate_synonyms_phrase_query": true, "boost": 1, "fuzziness": 1, "max_expansions": 10, "prefix_length": 2, "fuzzy_transpositions": true, "fuzzy_rewrite": "constant_score", "lenient": "true", "operator": "or", "minimum_should_match": 6, "zero_terms_query": "none" } } } }field[Required, object]：查询字段；query[Required, string]：查询analyzer[Optional, string]：分词器auto_generate_synonyms_phrase_query[Optional, Boolean]：用于控制在处理同义词时是否自动生成短语查询；boost[Optional, float]：用于减少或增加查询的相关分数；默认为1；fuzziness[Optional, float]：最大可编辑距离；[Common options | Elasticsearch Guide [8.17] | Elastic](https://www.elastic.co/guide/en/elasticsearch/reference/current/common-options.html#fuzziness)max_expansions[Optional, float]：模糊查询过程中生成的变体词语的最大数量；prefix_length[Optional, float]：前缀长度，表示在进行模糊匹配之前，词语的前缀部分必须完全匹配的长度fuzzy_transpositions[Optional, Boolean]：是否允许字符换位fuzzy_rewrite：rewrite parameter；lenient[Optional, Boolean]：是否忽略格式错误operator：指定查询词条之间的逻辑关系；OR：词条之间的关系是“或”，即只要有一个词条匹配，文档就会被认为是相关的；AND：词条之间的关系是“与”，即所有词条都必须匹配；minimum_should_match：至少应该匹配的数量，[minimum_should_match parameter | Elasticsearch Guide [8.17] | Elastic](https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl-minimum-should-match.html)zero_terms_query：analyzer 分析后如果没有词，例如停用词被删除：none：默认，返回空；all：全部返回；GET policy_info/_search { "query": { "match": { "standard_department": { "query": "南京，人民,政府", "analyzer": "standard", "auto_generate_synonyms_phrase_query": true, "boost": 1, "fuzziness": 1, "max_expansions": 10, "prefix_length": 2, "fuzzy_transpositions": true, "fuzzy_rewrite": "constant_score", "lenient": "true", "operator": "or", "minimum_should_match": 6, "zero_terms_query": "none" } } } }{ "took": 18, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 65, "relation": "eq" }, "max_score": 6, "hits": []Match boolean prefix 查询对输入的内容进行分词，对每个分词使用term查询（除最后一个分词），对最后一个分词使用prefix查询，把所有子查询放到bool查询的should列表中；参数：GET xxx/_search { "query": { "match_bool_prefix":{ "field": { "query": "xxxx xx", "analyzer": "standard", "operator": "OR", "minimum_should_match": 1, "fuzziness": "AUTO", "prefix_length": 2, "max_expansions": 50, "fuzzy_transpositions": true } } } }field[Required, object]：查询字段；query[Required, string]：查询analyzer[Optional, string]：分词器operator[Optional, string]：指定查询词条之间的逻辑关系；OR：词条之间的关系是“或”，即只要有一个词条匹配，文档就会被认为是相关的；AND：词条之间的关系是“与”，即所有词条都必须匹配；minimum_should_match[Optional, float]：至少应该匹配的数量，[minimum_should_match parameter | Elasticsearch Guide [8.17] | Elastic](https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl-minimum-should-match.html)fuzziness，prefix_length，max_expansions，fuzzy_transpositions， fuzzy_rewrite参数会被放入所有term查询中；GET policy_info/_search { "query": { "match_bool_prefix":{ "department_district": { "query": "320100 32020", "analyzer": "standard", "operator": "OR", "minimum_should_match": 1, "fuzziness": "AUTO", "prefix_length": 2, "max_expansions": 50, "fuzzy_transpositions": true } } } }{ "took": 3, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 1407, "relation": "eq" }, "max_score": 3.4872897, "hits": [] ... ...Match phrasematch_phrase会将检索关键词分词；match_phrase的分词结果必须在被检索字段的分词中都包含，而且顺序必须相同，而且默认必须都是连续的；参数GET xxx/_search { "query": { "match_phrase": { "field": { "query": "xxx", "slop": 5, "analyzer": "standard", "zero_terms_query": "none" } } } }field[Required, object]：查询字段；query[Required, string]：查询；analyzer[Optional, string]：分词器；slop[Optional, float]：允许短语查询在词条之间插入slop个数量的其他词条；zero_terms_query：analyzer 分析后如果没有词，例如停用词被删除：none：默认，返回空；all：全部返回；GET policy_info/_search { "query": { "match_phrase": { "title": { "query": "人民政府工作规则 通知", "slop": 2, "analyzer": "standard", "zero_terms_query": "none" } } } }{ "took": 1, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 6, "relation": "eq" }, "max_score": 8.821167, "hits": [ { "_index": "policy_info", "_id": "2884", "_score": 8.821167, "_source": { "title": "许昌市人民政府关于印发许昌市人民政府工作规则的通知", } },Match phrase prefix 查询用于查找包含提供的文本词条的文档，这些词条的顺序与提供的顺序相同。提供的文本的最后一个词条被视为前缀，匹配任何以该词条开头的单词。参数：GET policy_info/_search { "query": { "match_phrase_prefix": { "field": { "query": "xxx xxx", "analyzer": "standard", "max_expansions": 10, "slop": 1, "zero_terms_query": "none" } } } }field[Required, object]：查询字段；query[Required, string]：查询；analyzer[Optional, string]：分词器；max_expansions[Optional, integer]：查询过程中生成的变体词语的最大数量，默认50；slop[Optional, float]：允许短语查询在词条之间插入slop个数量的其他词条；zero_terms_query：analyzer 分析后如果没有词，例如停用词被删除：none：默认，返回空；all：全部返回；GET policy_info/_search { "query": { "match_phrase_prefix": { "title": { "query": "市人民政府关于印发", "analyzer": "standard", "max_expansions": 10, "slop": 1, "zero_terms_query": "none" } } } }{ "took": 13, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 295, "relation": "eq" }, "max_score": 5.0614715, "hits": [ { "_index": "policy_info", "_id": "1881", "_score": 5.0614715, "_source": { "id": 1881, "qdu_id": "0e286b12060f464eb893fb0a75dc66df", ... "title": "洛阳市人民政府关于印发洛阳动力谷发展规划（2011-2015..", } }, ...Query string 查询根据提供的查询字符串返回文档参数GET policy_info/_search { "query": { "query_string": { "default_field": "xxx", "query": "", "allow_leading_wildcard": true, "analyze_wildcard": false, "analyzer": "standard", "auto_generate_phrase_queries": false, "boost": 1, "default_operator": "OR", "enable_position_increments": true, "fields": ["*"], "fuzziness": 2, "fuzzy_max_expansions": 50, "fuzzy_prefix_length": 0, "fuzzy_transpositions": true, "lenient": false, "max_determinized_states": 10000, "minimum_should_match": 1, "quote_analyzer": "simple", "phrase_slop": 2, "quote_field_suffix": ".appName", "rewrite": "constant_score", "time_zone": "+00:00" } } }query[Required，string]：查询字符串，语法：field name：status:active：在fieldname为status的字段中查找active；title:(quick OR brown)：在field name为title的字段中查找quick或brown；author:"John Smith"：在field name为author的字段中查找精确短语John Smith；first\ name:Alice：在field name为first name的字段中查找Alice；book.\*:(quick OR brown)：在所有以 book. 开头的字段中查找quick或brown；_exists_:title：查title非空的数据；wildcards：用?代替单个字符，用*代替零个或多个字符；匹配符在开头可能导致性能问题。Regular expression：使用"/"包裹的字符串可被解析为正则表达式；例如：name:/joh?n(ath[oa]n)/Fuzziness：用~描述符进行模糊匹配；例如：quikc~；默认的编辑距离是2，使用quikc~1使编辑距离降为1；不要混合Fuzziness和wildcards语法；[Query string query | Elasticsearch Guide [8.17] | Elastic](https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl-query-string-query.html#query-string-fuzziness)Proximity：短语查询要求单词是完全相同的顺序且距离为0，但临近查询可以设置最大可编辑距离；"fox quick"~5，fox和quick间最多可有五个单词；Ranges：[]闭区间，{}开区间；date:[2012-01-01 TO 2012-12-31]：日期从2012-01-01至2012-12-31；count:[1 TO 5]：count 从1到5；tag:{alpha TO omega}：tag从alpha到omega，开区间；count:[10 TO *]：count 大于等于10；date:{* TO 2012-01-01}：date小于2012-01-01；count:[1 TO 5}：1至5，包括1，不包括5；age:>10;age:>=10;age:<10;age:<=10：这些也行；Boosting：用^符号增加相关度，可用在单词，短语和组上；例如quick^2 fox，quick的权重为2，fox的权重为1；Boolean operators：+单词必须被匹配，-单词必须不被匹配；例如：quick brown +fox -news；fox必须被匹配，news必须不被匹配；quick和brown可不被匹配，匹配到会增加相关度；Grouping：(quick OR brown) AND fox：匹配quick AND fox或者`brown AND fox；status:(active OR pending) title:(full text search)^2Reserved characters：+ - = && || > < ! ( ) { } [ ] ^ " ~ * ? : \ /这些字符用\转义； Whitespaces and empty queries：空格不被认为是运算符，如果查询字符串为空或者只包含空格，查询将返回空集； Avoid using the query_string query for nested documents使用nested query查询nested documents，不要使用query_string Search multiple fields：default_field[Optional，string]：查询字符串中没有明确指定字段时，默认搜索的字段；可以是*,，搜索所有的字段；不包括nested documents；allow_leading_wildcard[Optional，Boolean]：如果为真，则通配符*和?可在查询字符串第一位；默认为true；analyze_wildcard：是否对查询字符串中的通配符进行分析，默认为false， 这种情况下，Elasticsearch不会对这些通配符术语进行分析，这意味着它们会被直接用于匹配索引中的词项。analyzer：分词器；auto_generate_synonyms_phrase_query：控制是否自动生成同义词短语查询，默认是true。例如，如果你的同义词配置中包含 "quick, fast" 这样的同义词对，当用户搜索 "quick brown fox" 时，Elasticsearch 会自动生成一个短语查询来匹配"fast brown fox"。boost[Optional, float]：用于减少或增加查询的相关分数；默认为1；介于0和1.0之间的提升值会降低相关性分数。大于1.0的值会增加相关性分数。default_operator[Optional，string]：指定查询词条之间的逻辑关系；OR：词条之间的关系是“或”，即只要有一个词条匹配，文档就会被认为是相关的；AND：词条之间的关系是“与”，即所有词条都必须匹配；enable_position_increments[Optional，Boolean]：启用位置增量。位置增量是指在分析过程中，词项之间的距离（即位置差异）；如果你的分析器移除了停用词，那么词项之间的实际距离会被保留并在查询中考虑；fields[Optional，List[string]]：要搜索的field，以下两种查询等价。可以使用简单通配符；GET /_search { "query": { "query_string": { "fields": [ "content", "name", "city.*"], "query": "this AND that" } } }GET /_search { "query": { "query_string": { "query": "(content:this OR name:this) AND (content:that OR name:that) AND (city.\\*:that OR city.\\*:that)" } } }fuzziness[Optional, float]：最大可编辑距离；[Common options | Elasticsearch Guide [8.17] | Elastic](https://www.elastic.co/guide/en/elasticsearch/reference/current/common-options.html#fuzziness)fuzzy_max_expansions[Optional, integer]：查询过程中生成的变体词语的最大数量，默认50；fuzzy_prefix_length[Optional, integer]：前缀长度，表示在进行模糊匹配之前，词语的前缀部分必须完全匹配的长度；fuzzy_transpositions[Optional，Boolean]：是否允许字符换位，默认true；lenient[Optional，Boolean]：是否忽略格式错误；max_determinized_states[Optional，integer]：用于限制正则表达式查询在确定化过程中可以生成的状态数，以防止过于复杂的正则表达式导致性能问题或内存溢出；minimum_should_match[Optional, float]：至少应该匹配的数量，[minimum_should_match parameter | Elasticsearch Guide [8.17] | Elastic](https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl-minimum-should-match.html)quote_analyzer[Optional，string]：指定了用于将查询字符串中引用的文本（即用双引号括起来的文本）转换为令牌（tokens）的分析器；{ "query": { "query_string": { "query": "\"quick brown fox\" AND title:Elasticsearch", "fields": ["title", "content"], "quote_analyzer": "standard" } } }quote_analyzer 设置为 standard，这意味着 Elasticsearch 会使用 standard 分析器来处理引用的文本 "quick brown fox"。quote_analyzer 只影响引用的文本部分，不会影响查询字符串中其他部分的分析。phrase_slop[Optional，integer]：允许短语查询在词条之间插入slop个数量的其他词条；quote_field_suffix[Optional，string]： TODOrewrite[Optional，string]： rewrite parameter；time_zone：国际协调时间（UTC）偏移量或者IANA时区；合法值+01:00，-09:00，Asia/Shanghaitype[Optional, string]：多字段查询时可用：Query string query | Elasticsearch Guide 8.17 | ElasticGET policy_info/_search { "query": { "query_string": { "default_field": "title", "query": "+\"洛阳市\"*", "allow_leading_wildcard": true, "analyze_wildcard": false, "analyzer": "standard", "auto_generate_synonyms_phrase_query": false, "boost": 1, "default_operator": "OR", "enable_position_increments": true, // "fields": ["*"], "fuzziness": 2, "fuzzy_max_expansions": 50, "fuzzy_prefix_length": 0, "fuzzy_transpositions": true, "lenient": false, "max_determinized_states": 10000, "minimum_should_match": 1, "quote_analyzer": "standard", "phrase_slop": 2, "quote_field_suffix": ".appName", "rewrite": "constant_score", "time_zone": "+00:00" } } }{ "took": 25, "timed_out": false, "_shards": { "total": 3, "successful": 3, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 722, "relation": "eq" }, "max_score": 5.318619, "hits": [ { "_index": "policy_info", "_id": "2099", "_score": 5.318619, "title": "洛阳市人民政府办公室转发洛阳市司法局洛阳市卫生局洛阳市财政局..", } ...
redis 限流redis 令牌桶限流@dataclass(frozen=True) class RateLimitResult: allowed: bool remaining_tokens: float retry_after_seconds: float now_ms: int class RedisTokenBucketRateLimiter: _TOKEN_BUCKET_LUA = r""" local key = KEYS[1] -- 令牌桶状态存储的 Redis Key（Hash） local capacity = tonumber(ARGV[1]) -- 桶容量：最多可累积的令牌数 local rate = tonumber(ARGV[2]) -- 补充速率：每秒补充多少令牌（tokens/s） local requested = tonumber(ARGV[3]) -- 本次请求要消耗的令牌数 local ttl_ms = tonumber(ARGV[4]) -- 令牌桶 Key 的过期时间（毫秒），用于回收冷 Key local t = redis.call('TIME') -- 读取 Redis 服务器时间：{秒, 微秒} local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000) -- 服务器时间换算为毫秒时间戳 local data = redis.call('HMGET', key, 'tokens', 'ts') -- 读取当前令牌数(tokens)与上次更新时间(ts) local tokens = tonumber(data[1]) -- 当前令牌数（可能为 nil） local ts = tonumber(data[2]) -- 上次更新时间戳（毫秒，可能为 nil） if tokens == nil then tokens = capacity end -- 初始化：若不存在，则认为桶是满的 if ts == nil then ts = now end -- 初始化：若不存在，则将上次更新时间设为当前 if now < ts then ts = now end -- 防御：若时钟回拨导致 now < ts，则强制对齐 local delta_ms = now - ts -- 距离上次更新过去了多少毫秒 local refill = (delta_ms / 1000.0) * rate -- 这段时间应补充的令牌数量（可为小数） tokens = math.min(capacity, tokens + refill) -- 将令牌补充后并截断到容量上限 local allowed = 0 -- 是否允许：0/1（Lua 没有布尔返回给 Python 的统一类型） local retry_after_ms = 0 -- 如果不允许，建议等待多少毫秒再试 if tokens >= requested then -- 若令牌充足 allowed = 1 -- 标记允许 tokens = tokens - requested -- 扣除本次消耗 else -- 否则令牌不足 allowed = 0 -- 标记拒绝 if rate > 0 then -- 若补充速率 > 0，可计算需要等待的时间 local missing = requested - tokens -- 还差多少令牌 retry_after_ms = math.ceil((missing / rate) * 1000) -- 差额按速率换算为等待毫秒数（向上取整） else -- 若 rate=0，则永远补不回令牌 retry_after_ms = -1 -- 用 -1 表示无法通过等待获得令牌 end -- 结束 rate>0 分支 end -- 结束 tokens>=requested 分支 redis.call('HMSET', key, 'tokens', tokens, 'ts', now) -- 写回最新 tokens 与 ts（原子更新） if ttl_ms ~= nil and ttl_ms > 0 then -- 若配置了过期时间且 >0 redis.call('PEXPIRE', key, ttl_ms) -- 给令牌桶 Key 设置过期（毫秒） end -- 结束 ttl 分支 return {allowed, tokens, retry_after_ms, now} -- 返回：是否允许、剩余令牌、建议等待(ms)、当前时间(ms) """ def __init__( self, redis: Optional[AsyncRedis] = None, *, capacity: Union[int, float], refill_rate: Union[int, float], requested: Union[int, float] = 1, ttl_ms: Optional[int] = None, key_prefix: str = "rate_limit:token_bucket:", ) -> None: capacity_f = float(capacity) rate_f = float(refill_rate) req_f = float(requested) if capacity_f <= 0: raise ValueError("capacity 必须 > 0") if rate_f < 0: raise ValueError("refill_rate 必须 >= 0") if req_f <= 0: raise ValueError("requested 必须 > 0") if req_f > capacity_f: raise ValueError("requested 不能大于 capacity") self._redis = redis or async_redis_client self._key_prefix = key_prefix self._capacity = capacity_f self._refill_rate = rate_f self._requested = req_f self._ttl_ms = int(ttl_ms) if ttl_ms is not None else None def _full_key(self, key: str) -> str: return f"{self._key_prefix}{key}" @staticmethod def _default_ttl_ms( capacity: Union[int, float], refill_rate: Union[int, float] ) -> int: # 让桶在“完全补满所需时间”的 2 倍后过期，避免大量冷 key 常驻 # 至少 5 秒，防止极小桶/极大 rate 造成频繁抖动 if refill_rate <= 0: return 60_000 seconds = max(5.0, float(capacity) / float(refill_rate) * 2.0) return int(seconds * 1000) async def allow( self, key: str, *, requested: Optional[Union[int, float]] = None, ttl_ms: Optional[int] = None, ) -> RateLimitResult: """ 尝试拿令牌（令牌桶）。 - capacity/refill_rate/requested 默认来自 __init__ - requested/ttl_ms 允许在单次调用中覆盖 """ capacity_f = self._capacity rate_f = self._refill_rate req_f = float(requested) if requested is not None else self._requested if req_f <= 0: raise ValueError("requested 必须 > 0") if req_f > capacity_f: raise ValueError("requested 不能大于 capacity") ttl_ms_i = ( int(ttl_ms) if ttl_ms is not None else ( self._ttl_ms if self._ttl_ms is not None else self._default_ttl_ms(capacity_f, rate_f) ) ) full_key = self._full_key(key) res = await self._redis.eval( self._TOKEN_BUCKET_LUA, 1, full_key, capacity_f, rate_f, req_f, ttl_ms_i, ) allowed = bool(int(res[0])) remaining = float(res[1]) retry_after_ms = int(res[2]) now_ms = int(res[3]) retry_after_seconds = 0.0 if not allowed: retry_after_seconds = ( 0.0 if retry_after_ms <= 0 else retry_after_ms / 1000.0 ) return RateLimitResult( allowed=allowed, remaining_tokens=remaining, retry_after_seconds=retry_after_seconds, now_ms=now_ms, ) async def wait( self, key: str, *, requested: Optional[Union[int, float]] = None, ttl_ms: Optional[int] = None, max_wait_seconds: Optional[float] = None, poll_min_sleep: float = 0.01, ) -> RateLimitResult: """ 一直等到拿到令牌为止（适合 Activity 里对外部依赖做限流）。 max_wait_seconds: 超过则抛 TimeoutError """ start = time.perf_counter() while True: r = await self.allow( key, requested=requested, ttl_ms=ttl_ms, ) if r.allowed: return r sleep_s = max(poll_min_sleep, r.retry_after_seconds) if max_wait_seconds is not None: elapsed = time.perf_counter() - start if elapsed + sleep_s > max_wait_seconds: raise TimeoutError( f"rate limit wait timeout: key={key}, waited={elapsed:.3f}s, next_sleep={sleep_s:.3f}s" ) await asyncio.sleep(sleep_s) redis 滑动窗口限流@dataclass(frozen=True) class SlidingWindowResult: allowed: bool current_count: int retry_after_seconds: float now_ms: int class RedisSlidingWindowRateLimiter: """ Redis 滑动窗口限流（ZSET）。 语义：任意连续 window_seconds 秒内最多 limit 次通过。 """ _SLIDING_WINDOW_LUA = r""" local key = KEYS[1] -- 滑动窗口限流使用的 ZSET Key local window_ms = tonumber(ARGV[1]) -- 窗口长度（毫秒） local limit = tonumber(ARGV[2]) -- 窗口内允许的最大通过次数 local ttl_ms = tonumber(ARGV[3]) -- Key 过期时间（毫秒），用于回收冷 Key -- Redis server time (ms) -- 注释：以下用 Redis TIME 作为统一时钟，避免多机时间不一致 local t = redis.call('TIME') -- 读取 Redis 服务器时间：{秒, 微秒} local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000) -- 转为毫秒时间戳 -- window cleanup -- 注释：删除窗口之外的旧请求记录 local start = now - window_ms -- 窗口起点（毫秒） redis.call('ZREMRANGEBYSCORE', key, '-inf', start) -- 删除 score <= start 的成员（窗口外） local cnt = tonumber(redis.call('ZCARD', key)) -- 统计窗口内当前已有多少次请求 if cnt < limit then -- 若窗口内次数仍未达到上限（严格 <，避免多放 1 次） -- unique member to avoid overwrite under same ms -- 注释：member 唯一化，避免同毫秒覆盖导致少计数 local seq = redis.call('INCR', key .. ':seq') -- 递增序列号（辅助唯一 member） local member = tostring(now) .. ':' .. tostring(seq) -- member = now:seq redis.call('ZADD', key, now, member) -- 写入本次请求记录：score=now(member 的时间) if ttl_ms ~= nil and ttl_ms > 0 then -- 若配置了过期时间且 >0 redis.call('PEXPIRE', key, ttl_ms) -- 给 ZSET Key 设置过期时间 redis.call('PEXPIRE', key .. ':seq', ttl_ms) -- 给序列号 Key 也设置过期时间 end -- 结束 ttl 分支 return {1, cnt + 1, 0, now} -- 返回：允许(1)、窗口内计数(含本次)、建议等待(ms=0)、now(ms) else -- 否则窗口内次数已满，需要拒绝 -- compute retry_after_ms: when the oldest entry exits window -- 注释：计算何时最早记录滑出窗口 local oldest = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES') -- 取窗口内最早的一条（score 最小） local oldest_ts = nil -- 最早记录的时间戳（毫秒） if oldest ~= nil and #oldest >= 2 then -- 若存在成员且带 score oldest_ts = tonumber(oldest[2]) -- oldest[2] 为 score（ZSET 的时间戳） end -- 结束 oldest 解析 local retry_after = 0 -- 建议等待的毫秒数 if oldest_ts ~= nil then -- 若能拿到最早时间戳 retry_after = math.max(0, math.ceil((oldest_ts + window_ms) - now)) -- 等到 oldest_ts+window_ms 才会滑出 else -- 极端情况：ZSET 为空但 cnt>=limit（理论不应发生） retry_after = math.ceil(window_ms) -- 保守等待一个窗口长度 end -- 结束 oldest_ts 分支 if ttl_ms ~= nil and ttl_ms > 0 then -- 若配置了过期时间且 >0 redis.call('PEXPIRE', key, ttl_ms) -- 维持 key 的过期时间（避免热 key 被误删） redis.call('PEXPIRE', key .. ':seq', ttl_ms) -- 维持 seq key 的过期时间 end -- 结束 ttl 分支 return {0, cnt, retry_after, now} -- 返回：拒绝(0)、当前窗口计数、建议等待(ms)、now(ms) end -- 结束 if cnt < limit 分支 """ def __init__( self, redis: Optional[AsyncRedis] = None, *, limit: int, window_seconds: float, ttl_ms: Optional[int] = None, key_prefix: str = "rate_limit:sliding_window:", ) -> None: if limit <= 0: raise ValueError("limit 必须 > 0") if window_seconds <= 0: raise ValueError("window_seconds 必须 > 0") self._redis = redis or async_redis_client self._key_prefix = key_prefix self._limit = int(limit) self._window_seconds = float(window_seconds) self._ttl_ms = int(ttl_ms) if ttl_ms is not None else None def _full_key(self, key: str) -> str: return f"{self._key_prefix}{key}" @staticmethod def _default_ttl_ms(window_seconds: float) -> int: # key 存活时间略大于窗口，方便清理且避免冷 key 常驻 return int(max(5.0, window_seconds * 2.0) * 1000) async def allow( self, key: str, *, ttl_ms: Optional[int] = None, ) -> SlidingWindowResult: window_ms = int(self._window_seconds * 1000) ttl_ms_i = ( int(ttl_ms) if ttl_ms is not None else ( self._ttl_ms if self._ttl_ms is not None else self._default_ttl_ms(self._window_seconds) ) ) res = await self._redis.eval( self._SLIDING_WINDOW_LUA, 1, self._full_key(key), window_ms, self._limit, ttl_ms_i, ) allowed = bool(int(res[0])) current_count = int(res[1]) retry_after_ms = int(res[2]) now_ms = int(res[3]) return SlidingWindowResult( allowed=allowed, current_count=current_count, retry_after_seconds=retry_after_ms / 1000.0 if retry_after_ms > 0 else 0.0, now_ms=now_ms, ) async def wait( self, key: str, *, ttl_ms: Optional[int] = None, max_wait_seconds: Optional[float] = None, poll_min_sleep: float = 0.01, ) -> SlidingWindowResult: start = asyncio.get_running_loop().time() while True: r = await self.allow( key, ttl_ms=ttl_ms, ) if r.allowed: return r sleep_s = max(poll_min_sleep, r.retry_after_seconds) if max_wait_seconds is not None: elapsed = asyncio.get_running_loop().time() - start if elapsed + sleep_s > max_wait_seconds: raise TimeoutError( f"sliding window wait timeout: key={key}, waited={elapsed:.3f}s, next_sleep={sleep_s:.3f}s" ) await asyncio.sleep(sleep_s)
user nobody; # Nginx worker进程运行的用户及用户组 worker_processes 4; # 在master/worker运行方式下，定义worker进程的个数; 可以为auto worker_rlimit_nofile 65535; # 设置 Nginx worker 进程可以打开的最大文件描述符数量。这个指令通常用于确保 Nginx 能够处理大量并发连接，因为每个连接通常会占用一个文件描述符。 worker_rlimit_core 1m; # 通过worker_rlimit_core配置可以限制core文件的大小 working_directory /usr/local/lighthouse/softwares/nginx/coredump; # 设置coredump文件所放置的目录 daemon on; # 是否以守护进程方式运行Nginx master_process on; # 是否以 master/worker 方式工作 worker_cpu_affinity 1000 0100 0010 0001; # 绑定worker进程到指定的CPU内核 error_log logs/error.log notice; # error_log/path/file level; timer_resolution 100ms; # 每次内核的事件调用（如epoll、select、poll、kqueue等）返回时，都会执行一次gettimeofday，实现用内核的时钟来更新Nginx中的缓存时钟。 events { # 处理网络连接设置的核心部分 debug_connection 10.224.66.14; # 仅仅来自以上IP地址的请求才会输出debug级别的日志 debug_connection 10.224.57.0/24; # 其他请求仍然沿用error_log中配置的日志级别 accept_mutex on; # 负载均衡锁 accept锁默认是打开的，如果关闭它，那么建立TCP连接的耗时会更短，但worker进程之间的负载会非常不均衡 accept_mutex_delay 500ms; # 使用accept锁后，同一时间只有一个worker进程能够取到accept锁。这个accept锁不是阻塞锁，如果取不到会立刻返回。 如果有一个worker进程试图取accept锁而没有取到，它至少要等accept_mutex_delay定义的时间间隔后才能再次试图取锁。 multi_accept off; # 当事件模型通知有新连接时，尽可能地对本次调度中客户端发起的所有TCP请求都建立连接。 use epoll; # 事件驱动模型 worker_connections 8192; # 每个worker进程可以同时处理的最大连接数。 } http { include mime.types; default_type application/octet-stream; log_format combinedio '$remote_addr - $remote_user [$time_local] ' '"$request" $status $body_bytes_sent ' '"$http_referer" "$http_user_agent" $request_length $request_time $upstream_response_time'; access_log logs/access.log combinedio; sendfile on; # 硬盘 >> kernel buffer >> user buffer>> kernel socket buffer >>协议栈 | 硬盘 >> kernel buffer (快速拷贝到kernelsocket buffer) >>协议栈 gzip on; # 压缩响应内容 tcp_nopush on; # 数据量达到一定值后发送 tcp_nodelay on; # 对keepalive连接是否使用TCP_NODELAY选项。# 应用程序每产生一次操作就会发送一个包, delay 会等待数据，因此操作后它不会发送一个数据包，而是将这段时间内的数据打成一个大的包 keepalive_timeout 0; # 一个keepalive连接在闲置超过一定时间后,服务器关闭这个连接。 client_body_timeout 10; # second HTTP包体 client_header_timeout 10; # second 客户端与服务器建立连接后将开始接收HTTP头部，在这个过程中，如果在一个时间间隔（超时时间）内没有读取到客户端发来的字节，则认为超时，并向客户端返回408($$$"Request timed out")响应。 client_header_buffer_size 1k; # 正常情况下Nginx接收用户请求中HTTP header部分（包括HTTP行和HTTP头部）时分配的内存buffer大小。 large_client_header_buffers 4 4k; # Nginx接收一个超大HTTP头部请求的buffer个数和每个buffer的大小。如果HTTP请求行（如GET/index HTTP/1.1）的大小超过上面的单个buffer，则返回414 output_buffers 2 32k; client_max_body_size 64m; # 浏览器在发送含有较大HTTP包体的请求时，其头部会有一个Content-Length字段，client_max_body_size是用来限制Content-Length所示值的大小的。因此，这个限制包体的配置非常有用处，因为不用等Nginx接收完所有的HTTP包体——这有可能消耗很长时间——就可以告诉用户请求过大不被接受。 client_body_buffer_size 256k; # Nginx接收HTTP包体的内存缓冲区大小。也就是说，HTTP包体会先接收到指定的这块缓存中，之后才决定是否写入磁盘 client_body_in_file_only off; # 当值为非off时，用户请求中的HTTP包体一律存储到磁盘文件中，即使只有0字节也会存储为文件。当请求结束时，如果配置为on，则这个文件不会被删除（该配置一般用于调试、定位问题）​，但如果配置为clean，则会删除该文件。 client_body_in_single_buffer off; # 用户请求中的HTTP包体一律存储到内存buffer中。当然，如果HTTP包体的大小超过了下面client_body_buffer_size设置的值，包体还是会写入到磁盘文件中。 large_client_header_buffers 2k; # HTTP包体会先接收到指定的这块缓存中，之后才决定是否写入磁盘。如果用户请求中含有HTTP头部Content-Length，并且其标识的长度小于定义的buffer大小，那么Nginx会自动降低本次请求所使用的内存buffer，以降低内存消耗。 connection_pool_size 256; # Nginx对于每个建立成功的TCP连接会预先分配一个内存池，上面的size配置项将指定这个内存池的初始大小 request_pool_size 1k; # Nginx开始处理HTTP请求时，将会为每个请求都分配一个内存池，size配置项将指定这个内存池的初始大小 # client_body_temp_path # HTTP包体的临时存放目录 send_timeout 30; # second 发送响应的超时时间，即Nginx服务器向客户端发送了数据包，但客户端一直没有去接收这个数据包。如果某个连接超过send_timeout定义的超时时间，那么Nginx将会关闭这个连接。 reset_timeout_connection on; # 连接超时后将通过向客户端发送RST包来直接重置连接。这个选项打开后，Nginx会在某个连接超时后，不是使用正常情形下的四次握手关闭TCP连接，而是直接向用户发送RST重置包，不再等待用户的应答，直接释放Nginx服务器上关于这个套接字使用的所有缓存（如TCP滑动窗口 keepalive_requests 100; # 一个keepalive连接上默认最多只能发送100个请求。 limit_rate 4k; # 此配置是对客户端请求限制每秒传输的字节数。 limit_rate_after 1m; # 发送的响应长度超过limit_rate_after后才开始限速 ignore_invalid_headers off; # 忽略不合法的头部，默认on，如果将其设置为off，那么当出现不合法的HTTP头部时，Nginx会拒绝服务，并直接向用户发送400(Bad Request)错误。如果将其设置为on，则会忽略此HTTP头部。 underscores_in_headers off; # HTTP头部的名称中不允许带下划线 if_modified_since exact; # 浏览器在第一次请求资源后会缓存响应内容。在后续请求中，浏览器会在请求头中添加 If-Modified-Since，服务器根据该值判断是否需要返回最新的资源内容 # off：表示忽略用户请求中的If-Modified-Since头部。这时，如果获取一个文件，那么会正常地返回文件内容。HTTP响应码通常是200。 # exact：将If-Modified-Since头部包含的时间与将要返回的文件上次修改的时间做精确比较，如果没有匹配上，则返回200和文件的实际内容，如果匹配上，则返回 304 Not Modified # before：是比exact更宽松的比较。只要文件的上次修改时间等于或者早于用户请求中的If-Modified-Since头部的时间，就会向客户端返回304 Not Modified。 log_not_found on; # 此配置项表示当处理用户请求且需要访问文件时，如果没有找到文件，是否将错误日志记录到error.log文件中。 merge_slashes on; # 是否合并相邻的/ resolver 127.0.0.1 127.0.0.2; # DNS 解析地址 resolver_timeout 30; # DNS解析超时时间 server_tokens off; # 表示处理请求出错时是否在响应的Server头部中标明Nginx版本 server { listen 443 default deferred backlog=100 ssl; server_name work.ronie.ren; # 在开始处理一个HTTP请求时，Nginx会取出header头中的Host，与每个server中的server_name进行匹配，以此决定到底由哪一个server块来处理这个请求。有可能一个Host与多个server块中的server_name都匹配，这时就会根据匹配优先级来选择实际处理的server块。server_name与Host的匹配优先级如下：1)首先选择所有字符串完全匹配的server_name，如www.testweb.com。2)其次选择通配符在前面的server_name，如*.testweb.com。3)再次选择通配符在后面的server_name，如www.testweb.*。4)最后选择使用正则表达式才匹配的server_name，如~^\.testweb\.com$。 server_names_hash_bucket_size 1024 # 用于设置存储服务器名称的哈希表的桶大小。 server_names_hash_max_size 512 # 会影响散列表的冲突率。server_names_hash_max_size越大，消耗的内存就越多，但散列key的冲突率则会降低，检索速度也更快。 server_tokens off; keepalive_timeout 10; ssl_certificate /usr/local/lighthouse/softwares/nginx/certificates/work.ronie.ren_bundle.crt; ssl_certificate_key /usr/local/lighthouse/softwares/nginx/certificates/work.ronie.ren.key; ssl_session_timeout 5m; ssl_protocols TLSv1 TLSv1.1 TLSv1.2; ssl_ciphers ECDHE-RSA-AES128-GCM-SHA256:HIGH:!aNULL:!MD5:!RC4:!DHE; ssl_prefer_server_ciphers on; # 反向代理配置 location / { proxy_pass http://workservers; # 设置代理 proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } access_log "logs/work.ronie.ren.log"; error_log "logs/work.ronie.ren.error.log"; } upstream workservers{ ip_hash; # 据客户端的IP地址计算出一个key，将key按照upstream集群里的上游服务器数量进行取模，然后以取模后的结果把请求转发到相应的上游服务器中。这样就确保了同一个客户端的请求只会转发到指定的上游服务器中。 server 192.168.0.1:8080 weight=5 down; # down表示所在的上游服务器永久下线，只在使用ip_hash配置项时才有用。 server 192.168.0.2:8080 weight=5 max_fails=3 fail_timeout=30s; # 与fail_timeout配合使用，指在fail_timeout时间段内; server 192.168.0.3:8080 weight=5 max_fails=3 fail_timeout=30s; # 如果向当前的上游服务器转发失败次数超过number，则认为在当前的fail_timeout时间段内这台上游服务器不可用。 server 192.168.0.4:8080 weight=5 max_fails=3 fail_timeout=30s backup; # backup：在使用ip_hash配置项时它是无效的。 } }
docker kafka 配置文件version: '3.8' name: "kafka-service" # https://docs.docker.com/compose/how-tos/project-name/ services: kafka-server0: image: bitnami/kafka:latest hostname: kafka0 container_name: kafka0 ports: - "9092:9094" environment: KAFKA_CFG_NODE_ID: 0 KAFKA_CFG_BROKER_ID: 0 KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://0.0.0.0:9094 KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka0:9092,EXTERNAL://192.168.60.134:9092 KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@kafka1:9093,2@kafka2:9093,0@kafka0:9093 KAFKA_CFG_PROCESS_ROLES: broker,controller # KAFKA_CFG_LOG_DIRS: /var/lib/kafka/data KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: 3 KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3 KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR: 2 KAFKA_CFG_MIN_INSYNC_REPLICAS: 1 KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: true KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_KRAFT_CLUSTER_ID: d6c6ea9d1bfe423ebdc74579b10c61f0 networks: kafka: ipv4_address: 172.16.0.2 volumes: - ~/kafka-data/0:/bitnami/kafka kafka-server1: image: bitnami/kafka:latest hostname: kafka1 container_name: kafka1 ports: - "9094:9094" environment: KAFKA_CFG_NODE_ID: 1 KAFKA_CFG_BROKER_ID: 1 KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://0.0.0.0:9094 KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092,EXTERNAL://192.168.60.134:9094 KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@kafka1:9093,2@kafka2:9093,0@kafka0:9093 KAFKA_CFG_PROCESS_ROLES: broker,controller # KAFKA_CFG_LOG_DIRS: /var/lib/kafka/data KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: 3 KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3 KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR: 2 KAFKA_CFG_MIN_INSYNC_REPLICAS: 1 KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: true KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_KRAFT_CLUSTER_ID: d6c6ea9d1bfe423ebdc74579b10c61f0 networks: kafka: ipv4_address: 172.16.0.3 volumes: - ~/kafka-data/1:/bitnami/kafka kafka-server2: image: bitnami/kafka:latest hostname: kafka2 container_name: kafka2 ports: - "9096:9094" environment: KAFKA_CFG_NODE_ID: 2 KAFKA_CFG_BROKER_ID: 2 KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://0.0.0.0:9094 KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka2:9092,EXTERNAL://192.168.60.134:9096 KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@kafka1:9093,2@kafka2:9093,0@kafka0:9093 KAFKA_CFG_PROCESS_ROLES: broker,controller # KAFKA_CFG_LOG_DIRS: /var/lib/kafka/data KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: 3 KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3 KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR: 2 KAFKA_CFG_MIN_INSYNC_REPLICAS: 1 KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: true KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_KRAFT_CLUSTER_ID: d6c6ea9d1bfe423ebdc74579b10c61f0 networks: kafka: ipv4_address: 172.16.0.4 volumes: - ~/kafka-data/2:/bitnami/kafka networks: kafka: driver: bridge name: kafka-net ipam: config: - subnet: 172.16.0.0/24 gateway: 172.16.0.1 KAFKA_CFG_LISTENERS：配置的监听器将被用于监听网络请求，listener_name://host_name:port,listener_name2://host_name2:port2。可以同时配置多个，并且用逗号隔开，监听器的名称和端口必须是唯一的，host_name如果为空，则会绑定到默认的接口(网卡)。KAFKA_CFG_ADVERTISED_LISTENERS：ADVERTISED_LISTENERS 配置 Kafka broker 向客户端（生产者和消费者）报告的地址。这是客户端用来连接到 broker 的地址。例如：ADVERTISED_LISTENERS=PLAINTEXT://broker-hostname:9092 表示 broker 将向客户端报告 broker-hostname:9092 作为连接地址。KAFKA_CFG_PROCESS_ROLES：如果PROCESS_ROLES = Broker，服务器在KRaft模式中充当 Broker。如果PROCESS_ROLES = Controller，服务器在KRaft模式下充当 Controller。如果PROCESS_ROLES = Broker,Controller，服务器在KRaft模式中同时充当 Broker 和Controller。如果PROCESS_ROLES没有设置，那么集群就假定是运行在ZooKeeper模式下。KAFKA_CFG_CONTROLLER_QUORUM_VOTERS：标识有哪些节点是 Quorum 的投票者节点。所有想成为控制器的节点都需要包含在这个配置里面。需要包含每个节点的id。如KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@kafka1:9093,2@kafka2:9093,0@kafka0:9093KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR：用于指定内部主题 __consumer_offsets 的副本因子。__consumer_offsets 主题用于存储消费者组的偏移量信息。设置 KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: 3 表示 __consumer_offsets 主题将有 3 个副本。这意味着即使有两个 broker 发生故障，仍然可以从剩余的一个副本中读取消费者偏移量信息。KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR：指定内部主题 __transaction_state 的副本因子。__transaction_state 主题用于存储事务状态信息。KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR：指定了事务状态日志主题的最小同步副本数（ISR）。这个参数确保在提交事务时，至少有指定数量的副本是同步的。说一说你对 Kafka 中 ISR 的理解-腾讯云开发者社区-腾讯云KAFKA_CFG_MIN_INSYNC_REPLICAS：指定在写操作（例如生产消息）被认为成功之前，必须确认写入的最小同步副本（ISR）数量。问题：在外部网络能够正常使用KafkaProducer向kafka docker集群生产消息，但KafkaConsumer不能正常使用group_id消费消息（而不指定group_id，却能消费消息）。问题无法复现，在解决NotLeaderForPartitionError问题后，不在出现。挂载卷问题（待解决）：kafka0 | mkdir: cannot create directory '/bitnami/kafka/config': Permission denied kafka2 | mkdir: cannot create directory '/bitnami/kafka/config': Permission denied kafka1 | mkdir: cannot create directory '/bitnami/kafka/config': Permission deniedNOTE: As this is a non-root container, the mounted files and directories must have the proper permissions for the UID 1001.sudo chmod 776 -R kafka-data/：为其他用户组给出可读可写权限。Use non-root containersWhy non-root containers are important for securityWork With Non-Root ContainersRunning non-root containers on Openshift生产者 NotLeaderForPartitionError：KAFKA_CFG_ADVERTISED_LISTENERS配置错误：错误配置：kafka-server1: image: bitnami/kafka:latest hostname: kafka1 container_name: kafka1 ports: - "9092:9094" environment: KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094 KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092,EXTERNAL://localhost:9094正确配置：kafka-server1: image: bitnami/kafka:latest hostname: kafka1 container_name: kafka1 ports: - "9092:9094" environment: KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://0.0.0.0:9094 KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092,EXTERNAL://192.168.60.134:9092原因：Note: To connect from an external machine, change localhost above to your host's external IP/hostname and include EXTERNAL://0.0.0.0:9094 in KAFKA_CFG_LISTENERS to allow for remote connections.KAFKA_CFG_ADVERTISED_LISTENERS:PLAINTEXT://kafka1:9092,EXTERNAL://localhost:9094 修改为PLAINTEXT://kafka1:9092,EXTERNAL://192.168.60.134:9092192.168.60.134为docker宿主机IP，9092为docker宿主机映射端口(.venv) ronie@ronie:~/PycharmProjects/evilbox$ docker ps CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES f2509ff75469 bitnami/kafka:latest "/opt/bitnami/script…" 6 seconds ago Up 6 seconds 9092/tcp, 0.0.0.0:9092->9094/tcp, [::]:9092->9094/tcp kafka0 因为docker的网络拓扑是这样的graph TD subgraph Docker_Host direction TB br0[Bridge Network: kafka-net] kafka0[Kafka Broker 0] kafka1[Kafka Broker 1] kafka2[Kafka Broker 2] br0 --- kafka0 br0 --- kafka1 br0 --- kafka2 end subgraph External_Network direction TB ext[External Client ] end ext --- br0
RedisRedis-Server通过bitnami/redis:7.4.1实现 redis 主从集群version: '3.8' name: 'redis-service' services: redis-master: image: 'bitnami/redis:7.4.1' container_name: redis-master hostname: redis-master expose: - '6379' environment: - REDIS_REPLICATION_MODE=master - REDIS_PASSWORD=OeLFo2b3GVXKj0Ren1Dh - REDIS_MASTER_PORT_NUMBER=6379 networks: redis: ipv4_address: 192.168.0.2 redis-replica: image: 'bitnami/redis:7.4.1' deploy: replicas: 2 depends_on: - redis-master environment: - REDIS_REPLICATION_MODE=slave - REDIS_MASTER_HOST=192.168.0.2 - REDIS_MASTER_PORT_NUMBER=6379 - REDIS_MASTER_PASSWORD=OeLFo2b3GVXKj0Ren1Dh - REDIS_PASSWORD=OeLFo2b3GVXKj0Ren1Dh networks: redis: networks: redis: driver: bridge name: redis-net ipam: config: - subnet: 192.168.0.0/24 gateway: 192.168.0.1 Redis-Sentinel为保证redis集群高可用，添加redis sentinel监控master节点。version: '3.8' name: 'redis-sentinel-cluster' services: redis-sentinel: image: 'bitnami/redis-sentinel:6.2.16' deploy: replicas: 3 environment: REDIS_SENTINEL_PORT_NUMBER: '26379' REDIS_MASTER_HOST: '192.168.0.2' REDIS_MASTER_PORT_NUMBER: '6379' REDIS_SENTINEL_QUORUM: '2' # 在进行故障转移之前，必须同意主节点失效的 Sentinel 实例的最小数量。 REDIS_MASTER_PASSWORD: 'OeLFo2b3GVXKj0Ren1Dh' REDIS_SENTINEL_DOWN_AFTER_MILLISECONDS: '5000' REDIS_SENTINEL_FAILOVER_TIMEOUT: '5000' REDIS_MASTER_SET: 'mymaster' networks: - redis networks: redis: name: 'redis-net' external: true测试后发现，redis sentinel 能在master节点失效后选举新master节点，但是在docker-compose中只暴露了master节点的6379端口。于是在redis-net网络中添加nginx反代。nginx反代（不可用）nginx-reverse-proxy: image: 'bitnami/nginx:1.27' volumes: - ./nginx.conf:/opt/bitnami/nginx/conf/stream_server_blocks/my_server_block.conf environment: NGINX_ENABLE_STREAM: yes ports: - '6379:6379' networks: redis: depends_on: - redis-replica添加反代后，发现slave节点不能向master节点转发命令（记错了，这个是redis分片集群的功能，分片节点计算键的插槽，该插槽不归该节点管就转发给该管这个插槽的节点）。放弃反代，直接将测试代码接入redis-net网络。问题发现容器内无法访问外部网络(.venv) ronie@ronie:~/PycharmProjects/evilbox$ docker attach 81 / # ping www.baidu.com PING www.baidu.com (180.101.50.188): 56 data bytes ^C --- www.baidu.com ping statistics --- 2 packets transmitted, 0 packets received, 100% packet loss发现docker0网卡down且未被分配IP。(.venv) ronie@ronie:~/PycharmProjects/evilbox$ ip addr show docker0 3: docker0: <NO-CARRIER,BROADCAST,MULTICAST,UP> mtu 1500 qdisc noqueue state DOWN group default link/ether 02:42:e2:f4:ca:4c brd ff:ff:ff:ff:ff:ff 猜测可能是root mode docker 和 rootless mode docker 冲突。停止root mode docker。(.venv) ronie@ronie:~/PycharmProjects/evilbox$ ip addr show docker0 3: docker0: <NO-CARRIER,BROADCAST,MULTICAST,UP> mtu 1500 qdisc noqueue state DOWN group default link/ether 02:42:f3:75:a1:fd brd ff:ff:ff:ff:ff:ff inet 172.17.0.1/16 brd 172.17.255.255 scope global docker0 valid_lft forever preferred_lft forever 分配了IP，但仍然是down（应该是down，docker0在root mode下使用）。检查容器网络设置 docker inspect 89，发现容器被设置了代理？？？不知道什么时候修改的，该回来。Proxy configuration | Docker Docs"Env": [ "NO_PROXY=localhost, 127.0.0.1, 10.96.0.0/12, 192.168.59.0/24, 192.168.49.0/24, 192.168.39.0/24", "no_proxy=localhost, 127.0.0.1, 10.96.0.0/12, 192.168.59.0/24, 192.168.49.0/24, 192.168.39.0/24", "HTTP_PROXY=http://127.0.0.1:2081", "http_proxy=http://127.0.0.1:2081", "HTTPS_PROXY=http://127.0.0.1:2081", "https_proxy=http://127.0.0.1:2081" ],注意到4404 /proc/self/exe --state-dir=/run/user/1000/dockerd-rootless --net=slirp4netns ，发现docker在rootless模式下是不用docker0这个网卡的。rootlesskit/docs/network.md at v2.0.0 · rootless-containers/rootlesskitRootless mode | Docker DocsPing does not work when /proc/sys/net/ipv4/ping_group_range is set to 1 0:(.venv) ronie@ronie:/proc/sys/net/ipv4$ pwd /proc/sys/net/ipv4 (.venv) ronie@ronie:/proc/sys/net/ipv4$ cat ping_group_range 1 0Rootless mode | Docker DocsOn some distributions, ping does not work by default.Add net.ipv4.ping_group_range = 0 2147483647 to /etc/sysctl.conf (or /etc/sysctl.d) and run sudo sysctl --system to allow using ping.rootless docker 网络 | 今天我学了什么总的来说是代理问题。clientimport typing from redis.sentinel import Sentinel from redis import Redis import redis import string import random class RedisClient(Redis): def __init__( self, sentinels: typing.Union[typing.List[typing.Tuple[str, int]], Sentinel], **kwargs ): if isinstance(sentinels, Sentinel): self.sentinels = sentinels else: self.sentinels = Sentinel(sentinels, socket_timeout=0.1) self.client_kwargs = kwargs host, port = self.sentinels.discover_master("mymaster") super().__init__(host=host, port=port, **kwargs) def renew(self): host, port = self.sentinels.discover_master("mymaster") super().__init__(host=host, port=port, **self.client_kwargs) return self def __getattribute__(self, item): attr = super().__getattribute__(item) if item != 'ensure_client' and callable(attr): attr = self.ensure_client(attr) return attr @staticmethod def ensure_client(func): def wrap(*args, **kwargs): self = args[0] if args else None if self is None: return func(*args, **kwargs) for i in range(3): try: ret = func(*args, **kwargs) break except redis.exceptions.ConnectionError: self.renew() else: raise redis.exceptions.ConnectionError return ret return wrap if __name__ == '__main__': r = RedisClient( sentinels=[ ('redis-sentinel-cluster-sentinel-1', 26379), ('redis-sentinel-cluster-sentinel-2', 26379), ('redis-sentinel-cluster-sentinel-3', 26379), ], password='OeLFo2b3GVXKj0Ren1Dh' ) r.lpush("test", "".join(random.choices(string.printable, k=20))) print(r.rpop("test"))
1711. 大餐计数1711. 大餐计数from typing import List from collections import Counter import math class Solution: def countPairs(self, deliciousness: List[int]) -> int: min_deliciousness = min(deliciousness) if min_deliciousness <= 0: min_deliciousness = 0.5 max_deliciousness = max(deliciousness) count = Counter(deliciousness) ans = 0 wait_list = [2 ** i for i in range(int(math.log(min_deliciousness * 2, 2)), int(math.log(max_deliciousness * 2, 2)) + 1)] used = set() for k, v in count.items(): for i in range(len(wait_list)): if wait_list[i] - k in count: if wait_list[i] - k == k and v > 1: ans += (1 + v - 1) * (v - 1) // 2 ans = ans % (10 ** 9 + 7) elif wait_list[i] - k == k: pass else: if wait_list[i] - k not in used: ans += v * count[wait_list[i] - k] ans = ans % (10 ** 9 + 7) used.add(k) return ans % (10 ** 9 + 7)1702. 修改后的最大二进制字符串1702. 修改后的最大二进制字符串class Solution: def maximumBinaryString(self, binary: str) -> str: left = 0 binary = list(binary) while left < len(binary) and binary[left] == '1': left += 1 right = left + 1 while right < len(binary): if binary[right] == '1': right += 1 continue if right - left == 1: binary[left] = '1' else: binary[left] = '1' binary[right] = '1' binary[left + 1] = '0' left += 1 right += 1 return ''.join(binary)133. 克隆图133. 克隆图class Node: def __init__(self, val=0, neighbors=None): self.val = val self.neighbors = neighbors if neighbors is not None else [] from typing import Optional class Solution: def __init__(self): self.created = {} self.used = set() def cloneGraph(self, node: Optional['Node']) -> Optional['Node']: if node is None: return None val = node.val self.dfs(node) return self.created[val] def dfs(self, node: Optional['Node']): if node.val in self.used: return if node.val not in self.created: new_node = Node(node.val) self.created[node.val] = new_node else: new_node = self.created[node.val] self.used.add(node.val) for i in range(len(node.neighbors)): if node.neighbors[i].val in self.created: new_node.neighbors.append(self.created[node.neighbors[i].val]) else: tmp_node = Node(node.neighbors[i].val) self.created[tmp_node.val] = tmp_node new_node.neighbors.append(tmp_node) self.dfs(node.neighbors[i]) 1991. 找到数组的中间位置1991. 找到数组的中间位置class Solution: def findMiddleIndex(self, nums: List[int]) -> int: right = sum(nums) left = 0 for i in range(len(nums)): right -= nums[i] if left == right: return i left += nums[i] return -1994. 腐烂的橘子994. 腐烂的橘子class Solution: def __init__(self): self.time = 0 def orangesRotting(self, grid: List[List[int]]) -> int: rotting_oranges = set() normal_oranges = set() for i in range(len(grid)): for j in range(len(grid[i])): if grid[i][j] == 2: rotting_oranges.add((i, j)) if grid[i][j] == 1: normal_oranges.add((i, j)) if not normal_oranges: return 0 while rotting_oranges: rotting_oranges_bak = set(rotting_oranges) for (x, y) in rotting_oranges_bak: if x - 1 >= 0 and grid[x - 1][y] == 1: grid[x - 1][y] = 2 rotting_oranges.add((x - 1, y)) if x + 1 < len(grid) and grid[x + 1][y] == 1: grid[x + 1][y] = 2 rotting_oranges.add((x + 1, y)) if y - 1 >= 0 and grid[x][y - 1] == 1: grid[x][y - 1] = 2 rotting_oranges.add((x, y - 1)) if y + 1 < len(grid[x]) and grid[x][y + 1] == 1: grid[x][y + 1] = 2 rotting_oranges.add((x, y + 1)) rotting_oranges = rotting_oranges - rotting_oranges_bak normal_oranges = normal_oranges - rotting_oranges self.time += 1 return self.time - 1 if not normal_oranges else -1
2260. 必须拿起的最小连续卡牌数2260. 必须拿起的最小连续卡牌数class Solution: def minimumCardPickup(self, cards: List[int]) -> int: res = {} mark = {cards[k]: False for k in range(len(cards))} for i in range(len(cards)): if cards[i] in res: if res[cards[i]][1] - res[cards[i]][0] > i - res[cards[i]][2] or res[cards[i]][1] == res[cards[i]][0]: res[cards[i]] = [res[cards[i]][2], i, i] mark[cards[i]] = True res[cards[i]][2] = i else: res[cards[i]] = [i, i, i] ans = float('inf') for k, v in res.items(): if v[0] == v[1]: continue if mark[k] and ans > v[1] - v[0] + 1: ans = v[1] - v[0] + 1 return ans if ans != float('inf') else -11893. 检查是否区域内所有整数都被覆盖1893. 检查是否区域内所有整数都被覆盖class Solution: def isCovered(self, ranges: List[List[int]], left: int, right: int) -> bool: ranges.sort(key=lambda x: x[0]) father = [ranges[i][0] for i in range(len(ranges))] for i in range(1, len(ranges)): if ranges[i][0] < ranges[i - 1][1] and ranges[i][1] < ranges[i - 1][1]: ranges[i][1] = ranges[i - 1][1] for i in range(1, len(ranges)): if ranges[i][0] - ranges[i - 1][1] <= 1: father[i] = father[i - 1] left_father = None right_father = None for i in range(len(ranges)): if ranges[i][0] <= left <= ranges[i][1] and not left_father: left_father = father[i] if ranges[i][0] <= right <= ranges[i][1] and not right_father: right_father = father[i] if left_father and right_father: break if left_father is None and right_father is None: return False else: return left_father == right_father2269. 找到一个数字的 K 美丽值2269. 找到一个数字的 K 美丽值class Solution: def divisorSubstrings(self, num: int, k: int) -> int: num = str(num) ans = 0 for i in range(len(num) - k + 1): sub_num = int(num[i:i+k]) if sub_num == 0: continue if int(num) % sub_num == 0: ans += 1 return ans1779. 找到最近的有相同 X 或 Y 坐标的点1779. 找到最近的有相同 X 或 Y 坐标的点class Solution: def nearestValidPoint(self, x: int, y: int, points: List[List[int]]) -> int: ans = 0 distance = float('inf') for i in range(len(points)): if points[i][0] == x or points[i][1] == y: dis = abs(points[i][0] - x) + abs(points[i][1] - y) if dis < distance: distance = dis ans = i return ans if distance != float('inf') else -12233. K 次增加后的最大乘积2233. K 次增加后的最大乘积贪心import heapq class Solution: def maximumProduct(self, nums: List[int], k: int) -> int: heapq.heapify(nums) for i in range(k): heapq.heapreplace(nums, nums[0] + 1) ans = 1 for i in range(len(nums)): ans *= nums[i] ans %= (10 ** 9 + 7) return ans