那个让Claude直接操控我电脑的MCP协议，到底是个什么鬼东西？
昨晚熬夜折腾了一宿，头发又掉了几十根，总算把Anthropic搞的那个MCP（Model Context Protocol）给摸透了。说实话，刚听到这个词的时候，我内心是拒绝的。这年头，概念造得比母猪下崽还快，今天Agent，明天RAG，后天又来个什么Chain。咱们干运维的，最怕这种听着高大上、落地一地鸡毛的玩意儿。但是！这次好像真不一样。当我第一次在Claude Desktop里，直接让它“读取我本地的Nginx日志并分析502错误原因”，然后它真的直接读了我硬盘里的文件并且给出分析结果时，我承认，我有点起鸡皮疙瘩了。这玩意儿不是单纯的插件，它更像是一个给大模型开的“USB接口”。以前我们怎么用AI？复制粘贴代码，或者把日志脱敏了再喂给它，要么就是费劲巴拉地写个脚本调用API。现在好了，MCP直接把这个过程标准化了。废话不多说，今天咱们就来扒一扒这个MCP到底是个啥，以及作为一个手痒的IT人，怎么自己动手撸一个MCP服务，让AI变成你的私人定制运维专家。啥是MCP？别给我整虚的官方文档写得那叫一个晦涩，什么“开放标准”、“上下文交互”，看着就困。其实你把它想象成电脑的USB协议就完事了。以前（没有MCP的时候），你想让ChatGPT连你的数据库，你得自己写个特殊的插件；你想让Claude连你的Jira，又得写另一套代码。这就好比以前的手机充电口，诺基亚是圆孔，三星是扁口，苹果是宽口，乱得一塌糊涂。MCP就是Type-C接口。只要你的数据源（比如你的运维平台、你的本地文件夹、你的Postgres数据库）实现了MCP协议，不管是Claude，还是最近很火的Cursor编辑器，亦或是Zed，都能直接插上用。这里面有两个核心角色：MCP Host（主机）：比如Claude Desktop客户端，或者Cursor编辑器。它是大脑，负责提问和处理。MCP Server（服务端）：这就是我们要开发的东西。它负责干活，比如去查数据库、去执行个ls -l、去读个文件。这一打通，事情就变得有意思了。你可以写个Server，专门查你们公司的服务器负载。然后你在Claude里问：“现在哪台机器CPU飙高了？”Claude就通过MCP协议，调你的Server查数据，然后告诉你结果。听起来是不是有点像API？是，也不是。它比API更“智能”，因为它把工具的定义、资源的模板都标准化了，AI一看就知道怎么用，不需要你写那种几百行的Prompt去教它。准备动手：环境这块得支棱起来咱们是搞技术的，光说不练假把式。只要你会点Python，这事儿就不难。虽说官方也支持TypeScript，但咱们运维圈子，Python才是亲爹。先把环境弄好。我假设你电脑上已经有Python 3.10以上了。没有的话自己去面壁。我最近特别喜欢用 uv 这个包管理工具，速度快得离谱，比pip那种老牛拉破车强多了。当然，你习惯用pip也行。先建个目录，假装我们要搞一个“运维百宝箱”：mkdir ops-mcp-server cd ops-mcp-server python -m venv .venv source .venv/bin/activate # Windows下是 .venv\Scripts\activate接下来安装核心库。Anthropic官方出了个SDK，叫 mcp。pip install mcp[cli]装完之后，咱们别急着写代码。很多教程一上来就让你写类、写继承，那太累了。咱们用个偷懒的方法——FastMCP。这玩意儿就像FastAPI一样，装饰器一挂，代码跑起来，贼爽。对了，还得装个 uvicorn 之类的吗？其实MCP通常是走标准输入输出（Stdio）通信的，特别是在本地跑的时候，这样最安全，不需要暴露端口。这也是我觉得它设计得精妙的地方——它默认不是为了做成HTTP微服务，而是做成一个本地的“插件进程”。第一个吃螃蟹：写个“查进程”的工具咱们的目标很简单：让Claude能查我电脑上运行的进程。新建个文件 server.py。代码大概长这样，逻辑我都写注释里了，仔细看：from mcp.server.fastmcp import FastMCP import psutil # 初始化一个Server，名字随便起，就叫 "OpsTools" mcp = FastMCP("OpsTools") @mcp.tool() def list_top_processes(limit: int = 5) -> str: """ 列出当前系统占用CPU最高的几个进程。 Args: limit: 要显示的进程数量，默认为5个 """ # 既然是实战，就得真去读系统信息 # psutil这库搞运维的应该都熟 procs = [] for proc in psutil.process_iter(['pid', 'name', 'username', 'cpu_percent']): try: # 这是一个坑，psutil获取cpu_percent有时候第一次是0，但这只是个demo，无所谓了 procs.append(proc.info) except (psutil.NoSuchProcess, psutil.AccessDenied): pass # 按CPU使用率降序排序 procs.sort(key=lambda x: x['cpu_percent'], reverse=True) # 咱们搞得人性化一点，格式化一下输出 result = "TOP CPU 进程列表:\n" result += f"{'PID':<10} {'名称':<20} {'用户':<15} {'CPU%':<10}\n" result += "-" * 60 + "\n" for p in procs[:limit]: result += f"{p['pid']:<10} {p['name']:<20} {p['username']:<15} {p['cpu_percent']:<10}\n" return result @mcp.tool() def kill_process_by_pid(pid: int) -> str: """ 通过PID杀掉某个进程（慎用！）。 """ try: p = psutil.Process(pid) p_name = p.name() p.terminate() # 温柔一点，先terminate return f"成功终止进程: {p_name} (PID: {pid})" except psutil.NoSuchProcess: return f"找不到PID为 {pid} 的进程，是不是已经挂了？" except psutil.AccessDenied: return f"权限不足！你是不是没sudo？杀不掉 PID {pid}" except Exception as e: return f"出事了，杀进程失败: {str(e)}" # 这一行是启动的关键 if __name__ == "__main__": mcp.run()代码很简单对吧？这里面有个特别关键的点，就是 @mcp.tool() 这个装饰器。你不需要写任何Prompt告诉AI“我有这个工具，参数是什么什么”。FastMCP会自动读取你的函数签名（Type Hints）和文档字符串（Docstring），自动生成MCP协议所需的Schema。也就是说，你注释写得越清楚，AI用得越顺手。这简直是“代码即文档”的最高境界啊。怎么连上Claude？这才是坑最大的地方代码写好了，你直接运行 python server.py，你会发现...什么反应都没有，或者程序直接挂在那不动。别慌，这是正常的。因为它是设计成通过Stdio（标准输入输出）交互的，它在等指令呢。要让Claude Desktop用上它，你得去改配置文件。这个配置文件通常藏得很深。Mac用户在：~/Library/Application Support/Claude/claude_desktop_config.jsonWindows用户在：%APPDATA%\Claude\claude_desktop_config.json你要是找不到，就在Claude Desktop的菜单里找找“Settings” -> “Developer”之类的选项，通常有个按钮能直接打开这个配置目录。打开这个JSON文件，加上这一段：{ "mcpServers": { "my-ops-tools": { "command": "/你的路径/ops-mcp-server/.venv/bin/python", "args": ["/你的路径/ops-mcp-server/server.py"] } } }注意！敲黑板了！这里千万别直接写 python。一定要写你那个虚拟环境里的绝对路径！因为Claude启动的时候，它不知道你的环境变量是啥，如果用系统的python，很可能找不到 mcp 或者 psutil 库，然后就默默失败，连个报错都没有，急死你。Windows用户记得要把路径里的反斜杠 \ 写成双斜杠 \\，不然JSON解析会报错。这种低级错误我当年刚入行的时候没少犯，现在想起来还觉得蠢。改完配置文件，重启Claude Desktop。这时候，你会看到输入框旁边多了一个小锤子的图标（或者是类似连接的图标）。点开它，如果你看到了 list_top_processes 和 kill_process_by_pid，恭喜你，你的第一个MCP Server跑通了！见证奇迹的时刻现在，你在对话框里输入：“帮我看看现在电脑上哪个进程最吃CPU？”你会看到Claude稍微停顿一下（它在思考用哪个工具），然后它会显示“Used list_top_processes”。接着，它会把你代码里返回的那个格式化好的字符串展示出来，甚至还会基于这个结果给你一点建议。比如它可能会说：“兄弟，你的 chrome 占用太高了，要不要关几个标签页？”接着你回：“把那个PID为 12345 的进程给我杀了。”它就会调用 kill_process_by_pid。这种感觉，怎么形容呢？就像是你给AI装了一双手。以前它只能动嘴皮子，现在它能干活了。进阶：Resource和Prompts又是啥？光有Tool（工具）还不够。MCP协议里还有两个重要的概念：Resources（资源）和Prompts（提示词）。Resources 理解起来就是“数据源”。比如你有一个经常变动的日志文件，或者一个API返回的实时JSON数据。你可以把它定义成一个Resource。@mcp.resource("file://logs/app.log") def read_log() -> str: # 实际上这里可以写得很复杂，比如只读最后100行 with open("app.log", "r") as f: return f.read()这样，AI就可以像订阅RSS一样，随时获取这个资源的内容，而不需要通过“调用函数”这种动作。它更像是一种被动的上下文注入。Prompts 则是给人用的模板。比如你经常让AI帮你写SQL语句。你可以定义一个Prompt叫 sql-expert，里面预设好你们公司的表结构描述。在MCP里定义好Prompt后，你在Claude的界面里就能直接选这个Prompt，省去了每次都要复制粘贴一大堆背景信息的麻烦。说实话，对于咱们个人开发者或者小团队运维来说，Tool 是最有用的。Resource和Prompt在大规模协作的时候才显威力。遇到的那些坑（血泪史）别看我写得挺顺，实际开发的时候全是坑。调试极其困难：因为Server是Claude作为子进程启动的，你看不到它的控制台输出。如果你的代码里有 print("hello")，完了，这会破坏Stdio的通信协议，导致连接直接断开。解决办法：用 sys.stderr.write() 来打印日志，或者专门配个logging写文件。千万别往stdout打杂七杂八的东西。超时问题：如果你的工具执行时间太长（比如去爬个网页），Claude可能会觉得Server挂了。解决办法：搞异步。FastMCP支持 async def。尽量把耗时的操作做成异步的。依赖地狱：刚才说了，路径一定要写对。而且，如果你用了像 pandas 这种重型库，加载时间过长，有时候也会导致连接超时。安全隐患：兄弟们，这玩意儿是直接在你的环境里跑代码啊！你要是写了个 rm -rf / 的工具，然后AI抽风了给你执行了，那画面太美我不敢看。忠告：所有的写操作、删除操作，一定要在代码里做二次确认，或者干脆只给AI只读权限。别太相信大模型的节操，它们有时候会幻觉，明明你没让它删，它觉得删了比较好，就给你动手了。这东西对运维意味着什么？MCP这东西，可能会彻底改变我们做运维工具的方式。以前我们做个运维平台，得画前端页面，得写后端API，得做鉴权，得写文档教同事怎么用。以后呢？我可能只需要写几个Python函数，把它们暴露成MCP Server。同事们想查数据、想重启服务，直接对着他们的Claude或者IDE说话就行了。“帮我把测试环境的数据库回滚到昨天的快照。”“帮我查一下昨天晚上8点到9点的错误日志。”根本不需要GUI界面了！Chat Interface（聊天界面）就是新的GUI。而且，这东西还能联动。比如你用Cursor写代码，你可以搞一个Gitlab的MCP Server。你在写代码的时候，直接问Cursor：“这个函数是谁在哪次提交里改的？”Cursor直接通过MCP调Gitlab API查出来告诉你。这效率，提升的不是一星半点。当然，现在MCP还在早期阶段，生态还比较乱。但我敢打赌，2026年这玩意儿会遍地开花。各种SaaS软件都会自带MCP接口。总结这篇博客写得有点散，主要是想把那种探索过程中的兴奋和踩坑的痛感分享给你们。简单回顾一下：MCP是AI连接数据的USB接口。用Python的 mcp 库配合 FastMCP 可以快速开发。通过 claude_desktop_config.json 进行配置。切记别用 print 调试，路径要写绝对路径。安全第一，别给AI太大权限。这就是我眼里的MCP。它不神秘，它就是个干活的工具。但是把它用好了，你的AI就不再是个只会陪聊的傻白甜，而是一个手里拿着扳手和螺丝刀的真正助手。如果你也想试试，赶紧动手吧。哪怕写个简单的“查询本地天气”或者“读取每日记事本”的工具，那种成就感也是爆棚的。行了，今天就唠到这。代码要是跑不通，或者遇到了奇葩报错，欢迎在后台留言，咱们一起脱发，一起进步。最后，如果你觉得这篇文章对你有那么一点点启发，或者只是觉得我吐槽得比较在理，麻烦动动发财的小手：公众号：运维躬行录个人博客：躬行笔记
什么是虚拟机？虚拟机（VM）是由软件创建和运行的物理计算机的虚拟化版本。本质上，它就是一台“云端计算机”。创建步骤创建虚拟机：在 Azure 门户中搜索虚拟机。点击创建虚拟机，在下拉菜单中选择虚拟机。在“基本信息”选项卡上，选择现有资源组或为虚拟机创建一个新的资源组。为虚拟机创建一个实例名称。选择一个地区。如有必要，请选择可用区；否则，请选择“不需要基础架构冗余”。选择图像和兼容的安全类型“标准”。创建管理员帐户。选择用户名和密码。创建入站端口：HTTP (80) 和 RDP。确认资格。转到监控选项卡，禁用系统启动诊断。转到标签选项卡，为虚拟机创建一个标签。然后进行审核和创建。点击创建。部署完成后，转到资源，然后转到网络，增加虚拟机的屏幕使用时间。保存更改。进入虚拟机概览并连接虚拟机。请前往下载 RDP 文件。下载完成后，进入电脑的下载文件夹，通过远程桌面协议进行部署，然后输入你在管理员部分创建的密码。连接上创建的win10如果这篇文章对你有帮助，别忘了点赞转发支持一下！想了解更多运维实战经验和技术干货，记得关注微信公众号@运维躬行录，领取学习大礼包！！！我会持续分享更多接地气的运维知识和踩坑经验。让我们一起在运维这条路上互相学习，共同进步！公众号：运维躬行录个人博客：躬行笔记
最近有接触到nodejs，今天就来讲讲在 Windows 上安装 Node.js ，直接通过命令行安装不再需要点击安装程序——这要归功于微软官方的 Windows 程序包管理器 winget 。以下是如何在几秒钟（三秒真男人！！！）内完全通过终端使用 npm 安装 Node.js 的方法。1. 打开 PowerShell按 Win + X → 选择 Windows PowerShell（管理员） 或终端（管理员） 。2. 查找可用的 Node.js 版本首先，列出所有可用的 Node.js 软件包：winget search OpenJS.NodeJS你会看到类似这样的结果：名称 ID 版本 源 ----------------------------------------------- Node.js OpenJS.NodeJS 25.1.0 winget Node.js 14 OpenJS.NodeJS.14 14.21.3 winget Node.js 16 OpenJS.NodeJS.16 16.20.2 winget Node.js 18 OpenJS.NodeJS.18 18.20.8 winget Node.js 20 OpenJS.NodeJS.20 20.19.5 winget Node.js 21 OpenJS.NodeJS.21 21.7.2 winget Node.js 22 OpenJS.NodeJS.22 22.21.0 winget Node.js 23 OpenJS.NodeJS.23 23.11.0 winget Node.js (LTS) OpenJS.NodeJS.LTS 24.11.0 winget选择 LTS 版本（推荐）或最新版本并安装：winget install -e --id OpenJS.NodeJS.LTS或者安装最新版本：winget install -e --id OpenJS.NodeJS⚠️ 建议大多数用户使用 LTS（长期支持）版本——它更稳定，并且会获得更长时间的安全更新。3. 重新打开命令行界面（重要！）安装完成后， 关闭并重新打开 PowerShell 或终端 。这将刷新 PATH 环境变量，以便识别 node 和 npm 。然后验证安装情况：node --version npm --version你应该能看到 Node.js 和 npm 的版本号（npm 与 Node.js 捆绑在一起）。4. 故障排除：PowerShell 中无法识别 npm。如果看到类似 npm : File C:\Program Files\nodejs\npm.ps1 cannot be loaded because running scripts is disabled on this system 的错误，则表示 PowerShell 的执行策略阻止了 npm 脚本。以下三种方法可以解决这个问题：选项 A：更改执行策略（推荐）以管理员身份运行 PowerShell 并设置执行策略：Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser然后关闭并重新打开 PowerShell。这样既可以运行本地脚本（例如 npm），又能防止未经签名的远程脚本运行。选项 B：改用命令提示符如果您不想更改 PowerShell 设置，请改用命令提示符 (cmd) 或带有命令提示符配置文件的 Windows 终端 。npm 在 cmd 中无需任何策略更改即可正常工作。选项 C：绕过当前会话暂时绕过当前 PowerShell 会话的策略：Set-ExecutionPolicy -ExecutionPolicy Bypass -Scope Process这只会影响当前会话，关闭 PowerShell 时会重置。5.（可选）升级 npm保持 npm 更新：npm install -g npm@latest为什么要使用 winget？✅ 官方（由微软和 OpenJS 基金会维护）✅ 自动将 Node.js 和 npm 添加到 PATH 环境变量中（只需重新打开终端即可生效）✅ 完全可脚本化（非常适合自动化和持续集成）✅ 适用于全新机器的可复现设置就这样——你只需一条命令就在 Windows 上安装了 Node.js 和 npm，重新打开终端后即可使用。如果这篇文章对你有帮助，别忘了点赞转发支持一下！想了解更多运维实战经验和技术干货，记得关注微信公众号@运维躬行录，领取学习大礼包！！！我会持续分享更多接地气的运维知识和踩坑经验。让我们一起在运维这条路上互相学习，共同进步！公众号：运维躬行录个人博客：躬行笔记
这个开源知识库太香了！PandaWiki搭建全过程，踩坑经验全分享
最近公司要搭建一个内部知识库，领导让我调研几个方案。说实话，市面上的知识库产品真的是五花八门，什么Confluence、Notion、语雀啥的，但要么收费贵得离谱，要么功能不够用。偶然间发现了PandaWiki这个开源项目，试用了一下发现还真不错。界面简洁，功能够用，关键是免费！今天就把我搭建的整个过程分享给大家，希望能帮到有需要的朋友。什么是PandaWiki项目地址：https://github.com/chaitin/PandaWiki演示体验demo：https://192.168.198.133:2443/doc/editor/019a131f-12df-759c-a1a9-194d3c5ae47cPandaWiki是一个基于Go语言（好多项目都是go开发的，看来确实得学go了！！！）开发的轻量级知识库系统，支持Markdown编辑，有着不错的用户体验。相比其他重量级的知识库系统，它的部署和维护都相对简单。我之前用过MediaWiki，那个配置起来真的是要命，光是数据库配置就能折腾半天。PandaWiki就不一样了，基本上几个命令就能跑起来。环境准备在开始之前，我们需要准备一台Linux服务器。我这里用的是Debian 13（老演员了！），配置是2核4G，对于小团队使用完全够了。首先更新一下系统：apt update && apt upgrade -y然后安装一些必要的工具：apt install -y wget curl git vim开始部署PandaWiki一条命令搞定安装！！舒服！bash -c "$(curl -fsSLk https://release.baizhi.cloud/panda-wiki/manager.sh)"选择安装还是升级！会让你确认安装目录，无脑回车就行，也是通过docker部署(最喜欢看这种安装界面了哈哈哈哈)还有交流群二维码！！哈哈哈初始化配置打开浏览器访问 http://你的服务器IP:8080，应该能看到PandaWiki的初始化页面。第一次访问会要求你创建管理员账户。我建议用户名不要用admin这种太常见的，密码也要设置复杂一点。初始化完成后，你就能看到一个干净的知识库界面了。界面还是挺清爽的，左侧是目录树，右侧是内容区域。让你配置大模型很丰富注册靠谱白嫖一个（不是广告！）基本使用新手文档：https://192.168.198.133:2443/doc/editor/019a131f-12da-76b3-bafe-3cfbe1069e69创建文档点击"新建文档"按钮，可以选择创建普通文档或者文件夹。PandaWiki支持Markdown语法，写起来还是很顺手的。我试着创建了几个测试文档，发现它的Markdown渲染效果还不错，代码高亮、表格、图片都支持得挺好。我这里直接通过rss来导入自己博客的文档！通过rss订阅可以直接拉取很多地方的操作文档！这是拉取到的博客文章：去发布导入本地文本这就是财富！！！发布！！！使用统计！常用配置使用体验用了一段时间下来，感觉PandaWiki还是挺不错的。界面简洁，功能够用，性能也还可以。我们团队大概20多个人在用，响应速度还是很快的。当然也有一些不足的地方，比如权限管理相对简单，插件生态也不够丰富。但对于大多数中小团队来说，这些功能已经够用了。最重要的是它是开源免费的，不用担心突然收费或者服务停止。而且代码托管在GitHub上，如果有需要还可以自己修改。我感觉很nice！！会给出具体的引用，我准备用这个来替代我的博客搜索了哈哈哈写在最后PandaWiki作为一个轻量级的知识库解决方案，确实有它的优势。部署简单，维护方便，功能够用，关键是免费开源。虽然在一些高级功能上可能不如商业产品，但对于大多数团队来说已经足够了。特别是对于预算有限的小团队，这绝对是个不错的选择。如果你也在寻找知识库解决方案，不妨试试PandaWiki。反正是免费的，试试又不会怀孕。部署过程中如果遇到问题，可以查看官方文档或者GitHub上的issue。社区虽然不大，但还是比较活跃的。好了，今天的分享就到这里。如果这篇文章对你有帮助，记得点个赞转发一下。有问题的话也可以在评论区留言，我看到会回复的。如果这篇文章对你有帮助，别忘了点赞转发支持一下！想了解更多运维实战经验和技术干货，记得关注微信公众号@运维躬行录，领取学习大礼包！！！我会持续分享更多接地气的运维知识和踩坑经验。让我们一起在运维这条路上互相学习，共同进步！公众号：运维躬行录个人博客：躬行笔记
Python sh库：让运维脚本告别subprocess的痛苦，一行代码搞定系统命令
最近在整理服务器监控脚本的时候，发现自己写的代码里到处都是subprocess.Popen()，看着就头疼。想起刚入行那会儿，为了执行个简单的ls命令，愣是写了十几行代码处理各种异常和输出，现在想想真是太蠢了。直到遇到了Python的sh库，我才发现原来执行系统命令可以这么优雅。今天就来聊聊这个让我爱不释手的库，相信看完之后你也会和我一样，再也不想碰subprocess了。为什么我会选择sh库说实话，Python标准库里的subprocess功能很强大，但用起来真的很繁琐。每次想执行个简单命令都要考虑一堆问题：怎么获取输出？怎么处理错误？怎么传参数？import subprocess # 传统的subprocess写法，光看着就累 result = subprocess.run(['ls', '-la'], capture_output=True, text=True, check=True) print(result.stdout)而用sh库呢？import sh print(sh.ls('-la'))就这么简单！一行代码搞定，看着就舒服多了。sh库的安装和基本使用安装很简单，pip一下就行：pip install sh基本使用更简单，直接把系统命令当成Python函数调用就可以了：import sh # 执行ls命令 print(sh.ls()) # 带参数的命令 print(sh.ls('-la', '/home')) # 管道操作也很简单 print(sh.grep(sh.ps('aux'), 'python'))我第一次看到这种写法的时候，真的被震撼到了。这不就是我一直想要的那种简洁的方式吗？实际项目中的应用场景在我日常运维中，sh库真的帮了大忙。比如说服务器健康检查脚本：import sh from datetime import datetime def check_server_status(): # 检查磁盘使用情况 disk_usage = sh.df('-h') print(f"磁盘使用情况：\n{disk_usage}") # 检查内存使用 memory_info = sh.free('-h') print(f"内存使用情况：\n{memory_info}") # 检查CPU负载 load_avg = sh.uptime() print(f"系统负载：{load_avg}") # 检查特定进程 try: nginx_processes = sh.pgrep('nginx') print(f"Nginx进程数：{len(nginx_processes.strip().split())}") except sh.ErrorReturnCode_1: print("警告：Nginx进程未运行！")这种写法比subprocess清爽太多了，而且代码可读性特别好。处理命令输出和错误sh库在处理命令输出方面也很贴心：import sh # 获取命令输出 output = sh.ls('/home') print(type(output)) # <class 'sh.RunningCommand'> print(str(output)) # 转换为字符串 # 实时处理输出 def process_line(line): print(f"处理行: {line.strip()}") sh.tail('-f', '/var/log/syslog', _out=process_line)错误处理也很直观：import sh try: # 执行可能失败的命令 result = sh.ls('/nonexistent') except sh.ErrorReturnCode_2 as e: print(f"命令执行失败: {e}") print(f"错误输出: {e.stderr}") print(f"标准输出: {e.stdout}") print(f"退出码: {e.exit_code}")高级功能：管道和重定向sh库对管道的支持让我印象深刻，写起来就像在shell里一样自然：import sh # 管道操作 result = sh.ps('aux') | sh.grep('python') | sh.wc('-l') print(f"Python进程数量: {result}") # 更复杂的管道 log_errors = sh.tail('-n', '1000', '/var/log/app.log') | sh.grep('ERROR') print(log_errors) # 重定向到文件 sh.dmesg(_out='/tmp/dmesg.log') # 从文件读取输入 sh.sort(_in='/tmp/unsorted.txt', _out='/tmp/sorted.txt')这种写法真的很接近shell脚本的感觉，但又保持了Python的灵活性。在自动化部署中的应用我们团队的自动化部署脚本大量使用了sh库，效果很不错：import sh import os def deploy_application(app_name, version): print(f"开始部署 {app_name} 版本 {version}") # 停止服务 try: sh.systemctl('stop', app_name) print("服务已停止") except sh.ErrorReturnCode: print("服务停止失败，继续部署...") # 备份当前版本 backup_dir = f"/backup/{app_name}-{datetime.now().strftime('%Y%m%d_%H%M%S')}" sh.mkdir('-p', backup_dir) sh.cp('-r', f'/opt/{app_name}', backup_dir) print(f"已备份到 {backup_dir}") # 下载新版本 download_url = f"https://releases.example.com/{app_name}-{version}.tar.gz" sh.wget(download_url, '-O', f'/tmp/{app_name}-{version}.tar.gz') # 解压和部署 sh.tar('-xzf', f'/tmp/{app_name}-{version}.tar.gz', '-C', '/opt/') # 更新配置文件权限 sh.chown('-R', 'app:app', f'/opt/{app_name}') sh.chmod('-R', '755', f'/opt/{app_name}') # 启动服务 sh.systemctl('start', app_name) sh.systemctl('enable', app_name) # 验证部署 try: status = sh.systemctl('is-active', app_name) if 'active' in str(status): print("部署成功！") else: print("部署可能有问题，请检查服务状态") except sh.ErrorReturnCode: print("服务启动失败！")日志分析的实际应用在日志分析方面，sh库也表现得很出色（看看就行，都用python了，谁还这样分析呀！！！）：import sh from collections import defaultdict import re def analyze_nginx_logs(log_file): # 分析访问最多的IP top_ips = sh.awk('{print $1}', log_file) | sh.sort() | sh.uniq('-c') | sh.sort('-nr') | sh.head('-10') print("访问最多的IP地址：") print(top_ips) # 分析状态码分布 status_codes = sh.awk('{print $9}', log_file) | sh.sort() | sh.uniq('-c') print("\n状态码分布：") print(status_codes) # 分析最耗时的请求 slow_requests = sh.awk('$NF > 1.0 {print $0}', log_file) | sh.sort('-k11', '-nr') | sh.head('-20') print("\n响应时间超过1秒的请求：") print(slow_requests) # 统计每小时的请求量 hourly_stats = sh.awk('{print $4}', log_file) | sh.cut('-c14-15') | sh.sort() | sh.uniq('-c') print("\n每小时请求量统计：") print(hourly_stats) # 使用示例 analyze_nginx_logs('/var/log/nginx/access.log')需要注意的坑参数传递的问题import sh # 错误的写法 # sh.find('/home -name "*.log"') # 这样会报错 # 正确的写法 sh.find('/home', '-name', '*.log') # 或者使用关键字参数 sh.find('/home', name='*.log')处理包含空格的路径import sh # 包含空格的路径需要特别处理 path_with_spaces = "/home/user/my documents" sh.ls(path_with_spaces) # sh库会自动处理空格长时间运行的命令import sh # 对于长时间运行的命令，建议设置超时 try: result = sh.rsync('-av', '/source/', '/destination/', _timeout=3600) except sh.TimeoutException: print("同步超时！")与其他库的对比说实话，除了sh库，我也用过其他一些执行系统命令的库：subprocess vs shsubprocess功能更全面，但代码冗长sh库更简洁，但在某些复杂场景下可能不够灵活os.system vs shos.system太原始，不推荐在生产环境使用sh库提供了更好的错误处理和输出控制pexpect vs shpexpect适合需要交互的场景sh库更适合执行简单的系统命令在容器化环境中的应用现在容器化部署很流行，sh库在这方面也很有用：import sh def manage_containers(): # 列出所有运行中的容器 running_containers = sh.docker('ps', '--format', 'table {{.Names}}\t{{.Status}}') print("运行中的容器：") print(running_containers) # 检查容器资源使用情况 container_stats = sh.docker('stats', '--no-stream', '--format', 'table {{.Container}}\t{{.CPUPerc}}\t{{.MemUsage}}') print("\n容器资源使用情况：") print(container_stats) # 清理无用的镜像 sh.docker('image', 'prune', '-f') print("\n已清理无用镜像") manage_containers()总结说了这么多，其实sh库最大的优势就是让Python代码更接近shell脚本的直觉性，同时又保持了Python的强大功能。在我的日常工作中，它已经成为了不可或缺的工具。当然，任何工具都有其适用场景。对于简单的系统命令执行，sh库绝对是首选；但如果需要复杂的进程控制或者跨平台兼容性，可能还是需要考虑其他方案。最重要的是，在使用任何执行系统命令的库时，都要时刻注意安全性。特别是在处理用户输入或者在生产环境中运行时，一定要做好输入验证和权限控制。希望这篇文章能帮到正在为subprocess复杂语法而头疼的朋友们。如果你有什么好的使用技巧或者踩过的坑，也欢迎在评论区分享交流。毕竟，工具是为了提高效率而存在的，选择最适合自己项目需求的工具才是最重要的。sh库只是众多选择中的一个，但对于大部分运维场景来说，它确实是个很不错的选择。如果这篇文章对你有帮助，别忘了点赞转发支持一下！想了解更多运维实战经验和技术干货，记得关注微信公众号@运维躬行录，领取学习大礼包！！！我会持续分享更多接地气的运维知识和踩坑经验。让我们一起在运维这条路上互相学习，共同进步！公众号：运维躬行录个人博客：躬行笔记
CDN优化实战：从命中率30%到95%的血泪史，AWS CloudFront踩坑全记录
最近客户cdn命中率异常，让我优化下缓存策略。经过一番折腾，总算把命中率提上来了！！！今天就把这些年踩过的坑和一些实用的经验分享给大家。其实CDN优化这事儿，说简单也简单，说复杂也复杂。很多人以为接入个CDN就万事大吉了，结果发现命中率低得可怜，钱花了不少效果却不明显。我之前就遇到过这种情况，CDN命中率只有30%多，简直是在烧钱。先说说CDN的那些关键指标做CDN优化，你得先知道怎么看数据。就像开车要看仪表盘一样，不然怎么知道车子跑得怎么样？命中率（Hit Ratio）这个是最重要的指标，没有之一。命中率就是CDN直接返回缓存内容的请求占总请求的比例。一般来说，静态资源的命中率应该在90%以上，如果你的命中率只有50-60%，那肯定有问题。在AWS CloudFront里查看命中率很简单，进入CloudFront控制台，选择你的分发，然后点击"Monitoring"标签页。这里会显示Cache hit rate的图表，可以看到不同时间段的命中率变化。回源率（Origin Request Rate）这个和命中率是相对的，回源率高说明CDN经常要去源站拿数据，这样就失去了CDN的意义。正常情况下回源率应该控制在10%以下。响应时间包括CDN响应时间和回源响应时间。CDN响应时间一般应该在100ms以内，如果超过这个数值就要检查是不是节点选择有问题。我记得有次优化一个电商网站，发现图片加载特别慢。一查监控发现CDN响应时间居然有2秒多，后来才发现是缓存策略配置错误，每次都要回源验证。带宽使用情况这个主要是看成本，CDN的费用很大一部分就是带宽费用。通过监控带宽使用情况，可以了解流量分布，优化成本。在AWS CloudFront的监控面板里，你还能看到：Origin latency（回源延迟）Error rate（错误率）Requests（请求数）Bytes downloaded（下载字节数）这些数据都很有用，特别是错误率，如果突然飙升可能是源站出问题了。命中率低的常见原因和解决方案说到命中率低，我遇到过各种奇葩的情况。缓存策略配置不当这是最常见的问题。很多人图省事，直接用默认配置，结果发现很多本该缓存的内容都没有缓存。比如说，有些动态参数会导致URL不同，CDN就认为是不同的资源。像这种情况：https://example.com/image.jpg?timestamp=1234567890 https://example.com/image.jpg?timestamp=1234567891CDN会认为这是两个不同的文件，实际上内容是一样的。解决办法是在CloudFront的Behavior设置里配置Query String参数处理，可以选择忽略某些参数或者只转发特定参数。TTL设置太短Time To Live设置太短会导致缓存频繁过期。我见过有人把图片的TTL设置成5分钟，这不是瞎搞吗？静态资源的TTL至少要设置几个小时，甚至几天。在CloudFront里可以设置三种TTL：Minimum TTL：最小缓存时间Maximum TTL：最大缓存时间Default TTL：默认缓存时间一般我会这样设置：图片、CSS、JS等静态资源：24小时到7天HTML页面：1-6小时API接口：根据业务需求，可能几分钟到几小时源站响应头配置问题有时候源站返回了不合适的Cache-Control头，比如no-cache或者max-age=0，这会导致CDN不缓存内容。我之前遇到过一个案例，开发在Nginx配置里给所有静态资源都加了no-cache，结果CDN命中率惨不忍睹。后来改成这样：location ~* \.(jpg|jpeg|png|gif|css|js)$ { expires 7d; add_header Cache-Control "public, immutable"; }预热不充分新上线的CDN或者清除缓存后，需要一段时间来"预热"。这期间命中率会比较低，这是正常现象。但如果长时间命中率都上不去，就要检查其他配置了。CloudFront支持手动预热，可以通过Invalidation功能清除缓存，然后用脚本批量访问重要资源来预热缓存。CDN优化策略详解现在说说aws托管的缓存策略AWS CDN托管的缓存策略详细说明策略名称缓存内容不缓存内容使用场景Amplify缓存大多数静态资源和部分动态内容不缓存包含特定身份验证头的请求Amplify应用程序的通用缓存策略Amplify-Default缓存静态资源和基于常用头部的内容，保留某些Cookie不缓存带有特定认证头的请求，某些动态API响应Amplify应用的标准配置，平衡缓存与个性化Amplify-Default-V2与Amplify-Default类似，但有优化的TTL设置与Amplify-Default类似需要改进的缓存性能的Amplify应用Amplify-DefaultNoCookies缓存静态资源和页面内容不缓存任何Cookie信息，不缓存认证请求不需要基于Cookie区分用户的Amplify应用Amplify-DefaultNoCookies-V2与DefaultNoCookies类似，但有更长的TTL与DefaultNoCookies类似追求更高缓存命中率的静态Amplify应用Amplify-ImageOptimization缓存图像文件(jpg, png, gif等)，支持不同的图像格式不缓存非图像内容，不保留大多数请求头图像密集型网站，图片库，电商产品展示Amplify-ImageOptimization-V2与ImageOptimization类似，增加了WebP等现代格式支持与ImageOptimization类似需要现代图像格式支持的应用Amplify-StaticContent缓存JS, CSS, 字体, 图像等静态资源，长TTL不缓存HTML或动态内容，不保留Cookie静态资源的长期缓存，提高重复访问性能Amplify-StaticContent-V2与StaticContent类似，但有更优化的压缩设置与StaticContent类似需要最佳静态资源性能的应用CachingDisabled不缓存任何内容所有内容都直接从源站获取实时数据，个性化内容，需要实时源站响应的场景CachingOptimized缓存大多数内容，启用压缩，优化TTL不缓存POST/PUT请求，特定动态API路径通用Web应用，需要平衡缓存与新鲜度的场景CachingOptimizedForUncompressedObjects缓存未压缩的对象，自动应用适当的TTL不缓存已经压缩的内容，不处理某些动态请求源站不提供压缩的情况，CDN负责压缩Elemental-MediaPackage缓存视频分段，流媒体清单文件不缓存用户特定的媒体请求，认证令牌视频流媒体，直播，点播视频服务UseOriginCacheControlHeaders根据源站的Cache-Control头决定缓存行为如果源站指示不缓存，则不缓存；忽略查询参数源站已有完善缓存策略的情况UseOriginCacheControlHeaders-QueryStrings根据源站Cache-Control头和URL查询参数缓存如源站指示不缓存则不缓存基于查询参数提供不同内容且源站控制缓存的场景AWS CloudFront 源请求策略详细说明策略名称转发内容不转发内容使用场景AllViewer转发所有查看者请求参数，包括查询字符串、标头和Cookie不限制任何参数的转发需要将所有客户端请求信息传递给源站的场景，适用于高度依赖客户端信息的动态内容AllViewerAndCloudFrontHeaders-2022-06转发所有查看者请求参数和截至2022年6月的所有CloudFront标头不限制任何参数或CloudFront标头的转发需要完整客户端信息以及CloudFront添加的信息(如设备类型、地理位置)的场景AllViewerExceptHostHeader转发除Host标头外的所有查看者请求参数不转发Host标头当源站需要使用不同的主机名处理请求，但仍需其他所有客户端信息的场景CORS-CustomOrigin转发与CORS(跨源资源共享)相关的标头到自定义源不转发与CORS无关的其他标头需要支持跨域请求的自定义源站(非S3)场景，如API或Web应用CORS-S3Origin转发与CORS相关的标头到S3源不转发与CORS无关的其他标头需要支持跨域请求的S3存储桶场景，如从浏览器直接访问S3资源Elemental-MediaTailor-PersonalizedManifests转发Elemental MediaTailor个性化清单所需的标头不转发与媒体个性化无关的标头视频流媒体场景，特别是需要个性化内容或广告插入的流媒体服务HostHeaderOnly仅转发Host标头到源站不转发任何其他查看者请求参数简单的静态内容分发场景，源站仅需要知道请求的主机名UserAgentReferrerHeaders转发User-Agent(用户代理)和Referer(引用页)标头到源站不转发其他查看者请求参数源站需要了解访问者浏览器类型和来源网站的场景，如内容适配或引用统计这些源请求策略控制CloudFront将哪些信息从查看者(客户端)请求中转发到源站。选择合适的策略可以优化源站请求处理，减少不必要的信息传递，同时确保源站获得所需的上下文信息来正确响应请求。AWS CloudFront 响应标头策略详细说明策略名称添加/修改内容不添加/不修改内容使用场景CORS-and-SecurityHeadersPolicy添加CORS相关标头和安全标头到响应中不修改源站提供的其他标头需要同时支持跨域请求和增强安全性的Web应用，如需要跨域访问的API且需要防XSS、点击劫持等安全保护CORS-With-Preflight添加CORS相关标头并支持预检(OPTIONS)请求不添加安全标头，不修改源站提供的其他标头需要支持复杂跨域请求的场景，特别是涉及非简单请求(如带自定义标头或使用PUT/DELETE方法)的APICORS-with-preflight-and-SecurityHeadersPolicy添加CORS相关标头(含预检支持)和安全标头不修改源站提供的其他标头需要全面CORS支持(包括复杂请求)和安全增强的Web应用，如SPA(单页应用)与后端API交互SecurityHeadersPolicy仅添加安全相关标头到每个响应不添加CORS标头，不修改源站提供的其他标头注重安全性但不需要跨域支持的网站，如企业内部应用或对安全有高要求的公共网站SimpleCORS仅添加基本CORS标头，支持简单跨域请求不添加安全标头，不支持预检请求，不修改源站提供的其他标头只需基本跨域支持的简单场景，如只允许GET请求的公共API或静态资源这些响应标头策略控制CloudFront如何修改从源站返回给客户端的HTTP响应标头。适当的策略可以增强Web应用的安全性，启用跨域资源共享功能，而无需修改源站配置。安全标头通常包括内容安全策略(CSP)、X-XSS-Protection、X-Frame-Options等，用于防止常见的Web安全威胁。当然要是这些不满足还可以自定义！！其他一些缓存策略分层缓存策略不同类型的资源用不同的缓存策略。我一般会创建多个Behavior，针对不同的路径模式设置不同的缓存规则：/api/* - TTL: 5分钟，转发所有参数 /images/* - TTL: 7天，忽略查询参数 /css/* - TTL: 1天，启用压缩 /js/* - TTL: 1天，启用压缩 /* - TTL: 1小时，默认行为启用压缩CloudFront支持自动压缩，可以显著减少传输数据量。在Behavior设置里开启"Compress Objects Automatically"就行了。不过要注意，压缩会消耗一些CPU资源，对于已经压缩过的文件（如图片）就没必要再压缩了。合理使用Origin Shield这是CloudFront的一个高级功能，相当于在CDN和源站之间加了一层缓存。对于回源比较频繁的场景很有用，可以减少源站压力。我在一个视频网站项目中用过Origin Shield，效果很明显。因为视频文件比较大，用户分布又比较集中，开启Origin Shield后回源请求减少了60%多。HTTP/2和HTTP/3支持现在的CDN基本都支持HTTP/2了，CloudFront也不例外。HTTP/2的多路复用特性可以显著提升页面加载速度，特别是对于有很多小文件的网站。智能路由CloudFront会自动选择最优的边缘节点，但有时候你可能需要手动干预。比如某些地区的节点质量不好，可以通过地理位置限制来避免使用这些节点。AWS CloudFront的一些实用技巧用了这么久CloudFront，发现了一些比较实用的技巧。Lambda@Edge这个功能很强大，可以在CDN边缘节点运行代码。我用过几个场景：动态修改响应头根据用户地理位置返回不同内容简单的A/B测试图片格式转换（WebP支持检测）比如这个检测WebP支持的例子：exports.handler = (event, context, callback) => { const request = event.Records[0].cf.request; const headers = request.headers; const accept = headers.accept && headers.accept[0] && headers.accept[0].value; if (accept && accept.includes('image/webp')) { request.uri = request.uri.replace(/\.(jpg|jpeg|png)$/i, '.webp'); } callback(null, request); };实时日志分析CloudFront可以把访问日志实时推送到Kinesis Data Streams，然后用Lambda处理。我用这个功能做过实时监控，当错误率超过阈值时自动发送告警。多源站配置CloudFront支持配置多个源站，可以实现负载均衡和故障转移。我一般会配置主源站和备用源站，当主源站出问题时自动切换到备用源站。自定义错误页面通过配置Custom Error Pages，可以为不同的HTTP错误码返回自定义页面。这样用户体验会好很多，而且可以减少无效的回源请求。监控和告警设置CDN优化不是一次性的工作，需要持续监控和调优。CloudWatch监控CloudFront的所有指标都会推送到CloudWatch，可以设置各种告警：命中率低于阈值告警错误率超过阈值告警回源延迟过高告警带宽使用异常告警我一般会设置这些告警规则：命中率 < 85% 持续10分钟 -> 发送告警 4xx错误率 > 5% 持续5分钟 -> 发送告警 5xx错误率 > 1% 持续5分钟 -> 发送告警 回源延迟 > 2秒 持续5分钟 -> 发送告警第三方监控工具除了CloudWatch，我还会用一些第三方工具来监控CDN性能，比如Pingdom、GTmetrix等。这些工具可以从用户角度测试网站性能，发现一些CloudWatch监控不到的问题。成本优化CDN的费用不便宜，特别是流量大的网站。Price Class选择CloudFront有三个价格等级，包含的边缘节点数量不同。如果你的用户主要在北美和欧洲，选择Price Class 100就够了，没必要选择包含全球所有节点的Price Class All。缓存策略优化提高命中率不仅能改善性能，还能降低成本。因为CDN的费用主要是按流量计算的，命中率高意味着回源流量少，总体费用就低。如果命中率低就会出现流量放大效应：• 正常命中率80%：1GB实际需求 = 1.25GB CDN流量• 当前命中率可能<20%：1GB实际需求 = 5GB+ CDN流量压缩和格式优化启用压缩可以减少传输数据量，WebP格式的图片比JPEG小30-50%。这些优化都能直接降低CDN费用。我之前优化过一个图片网站，通过启用压缩和WebP转换，CDN费用降低了40%多。常见问题排查做CDN优化经常会遇到各种问题，分享几个排查思路。缓存不生效首先检查响应头，看看有没有Cache-Control、Expires等缓存相关的头。然后检查CloudFront的Behavior配置，确认TTL设置正确。可以用curl命令测试：curl -I https://your-domain.com/test.jpg看响应头里的X-Cache字段，Hit表示命中缓存，Miss表示没有命中。某些地区访问慢可能是CDN节点选择有问题。用不同地区的VPS测试访问速度，确定是哪些地区有问题。然后检查是不是需要调整Price Class或者配置地理位置限制。缓存穿透如果某个资源一直无法缓存，可能是URL参数或者响应头有问题。检查是不是有随机参数，或者源站返回了no-cache头。我遇到过一个奇葩问题，开发在图片URL后面加了随机数防止缓存，结果CDN命中率为0。后来改成用版本号，问题就解决了。总结CDN优化是个系统工程，需要从多个维度来考虑。命中率是最重要的指标，但不是唯一的指标。要结合业务特点，制定合适的缓存策略。AWS CloudFront功能很强大，但配置也比较复杂。建议先从基础配置开始，逐步优化。不要一开始就搞得很复杂，容易出问题。最重要的是要持续监控和优化，CDN不是配置好就不管了。用户行为会变化，业务需求也会变化，CDN策略也要跟着调整。我现在维护的几个网站，CDN命中率都在95%以上，用户访问速度提升了3-5倍。虽然前期折腾了不少时间，但效果还是很明显的。如果你也在做CDN优化，遇到问题可以多交流。这块水还是挺深的，大家一起学习进步。如果这篇文章对你有帮助，别忘了点赞转发支持一下！想了解更多运维实战经验和技术干货，记得关注微信公众号@运维躬行录，领取学习大礼包！！！我会持续分享更多接地气的运维知识和踩坑经验。让我们一起在运维这条路上互相学习，共同进步！公众号：运维躬行录个人博客：躬行笔记
博客终于大更新啦！之前那个默认主题比较简单！除了博客文章不能给粉丝盆友带来什么价值。所以我在修改后的页面上加了一个导航页。希望能够激你大家收藏的欲望（大家有什么更好的建议也欢迎在评论区提哈，感谢支持！！！）主要变动：1、添加白天黑夜主题切换2、修改logo3、添加导航页，这个我会持续更新，欢迎大家收藏，有事没事来逛逛4、添加个人介绍页面5、优化页面样式，响应式布局采用DaisyUI + Tailwind CSS构建，提供简洁优雅的设计风格和丰富的功能特性。主题注重用户体验和响应式设计。DaisyUI官网地址：https://daisyui.com/Tailwind CSS官网地址：https://tailwindcss.com/
云计算专业术语大揭秘！这些"黑话"你都懂吗？
最近在公司开会，听到同事们各种云计算术语满天飞，什么"弹性伸缩"、"容器编排"、"微服务架构"...说实话，刚入行那会儿我也是一脸懵，感觉大家说的都是外星语。做了这么多年运维，我发现很多朋友都有这样的困扰：技术文档看不懂，会议听不明白，跟开发沟通像是在打哑谜。其实云计算的这些专业术语并没有想象中那么高深，今天我就来给大家捋一捋，用最接地气的方式解释这些"黑话"。基础设施即服务，听起来很高大上IaaS（Infrastructure as a Service），翻译过来就是基础设施即服务。简单说就是云服务商把服务器、存储、网络这些硬件资源虚拟化之后租给你用。我记得以前公司要上个新项目，得先采购服务器，等设备到货，然后机房上架，网络配置...整个流程下来至少要一个月。现在用云服务，几分钟就能开出一台虚拟机，这就是IaaS的魅力。PaaS（Platform as a Service）更进一步，不仅提供基础设施，还提供了开发和运行环境。比如你要部署一个Java应用，传统方式需要自己安装JDK、配置Tomcat、设置数据库连接等等。用PaaS的话，这些环境都给你准备好了，你只需要把代码扔上去就行。SaaS（Software as a Service）就更直接了，软件即服务。像我们平时用的钉钉、腾讯会议、Office 365都属于SaaS。用户不需要安装任何软件，打开浏览器就能使用。说到这里想起个笑话，有次我跟一个做传统IT的朋友聊天，他问我："你们云计算到底是干什么的？"我说："就是把别人的电脑当自己的电脑用。"他恍然大悟："那不就是网吧模式吗？"...虽然这个比喻不太准确，但某种程度上还真有点像。虚拟化技术，一台机器变多台虚拟化（Virtualization）是云计算的基石。通过虚拟化技术，一台物理服务器可以运行多个虚拟机，每个虚拟机都像是一台独立的计算机。Hypervisor是虚拟化的核心组件，负责管理和分配物理资源给各个虚拟机。常见的有VMware vSphere、Microsoft Hyper-V、开源的KVM等。容器（Container）是另一种虚拟化技术，比传统虚拟机更轻量级。如果说虚拟机是把一栋楼分成若干套房子，那容器就是把一套房子分成若干个房间。Docker是目前最流行的容器技术。我刚接触Docker的时候觉得特别神奇，一个应用连同它的运行环境可以打包成一个镜像，在任何支持Docker的机器上都能运行。这解决的是运行环境标准化的问题，避免了“一杯茶一包烟，一个环境配一天”的悲剧发生。容器编排（Container Orchestration）就是管理大量容器的技术。想象一下，如果你有成百上千个容器需要管理，手工操作肯定要疯掉。Kubernetes（简称K8s）是目前最主流的容器编排平台，虽然学习曲线比较陡峭，但掌握之后确实很强大。弹性和可扩展性，云计算的杀手锏弹性伸缩（Auto Scaling）是云计算的一大优势。系统可以根据负载情况自动增加或减少资源。比如电商网站在促销期间访问量暴增，系统会自动启动更多服务器来应对；活动结束后又会自动释放多余的资源。传统电商公司活动页面，平时访问量很少，但每次做活动就会有大量用户涌入。以前总是要提前评估流量，手动扩容，经常出现要么资源不够导致页面卡顿，要么准备过多造成浪费。现在配置了弹性伸缩，系统会根据CPU使用率和响应时间自动调整，省心多了。负载均衡（Load Balancing）是把请求分发到多个服务器上，避免某台服务器过载。就像银行有多个窗口，客户排队时会被引导到空闲的窗口一样。CDN（Content Delivery Network）内容分发网络，把静态资源缓存到离用户最近的节点上，加快访问速度。比如你在北京访问一个网站，如果服务器在深圳，数据要跑很远；但如果在北京有CDN节点，就能就近获取内容。按量计费，云计算的经济学革命说到云计算，不得不提它最吸引人的特点之一——按量计费。这个模式彻底改变了企业的IT成本结构。传统IT模式下，你要上个项目，先得买服务器、买软件许可证、租机房位置...这些都是一次性的大笔投入。以前公司要上个ERP系统，光硬件采购就花了几十万，结果项目上线后发现性能过剩，大部分资源都闲置着。更要命的是，三年后硬件过保了，又得花钱升级。云计算的按量计费就像用水用电一样，用多少付多少。你的应用今天访问量大，系统自动扩容，多用了资源就多付费；明天访问量小，资源自动释放，费用也跟着降下来。比如某电商公司，平时一天可能就几百个访问，但做促销活动的时候一小时就有几万个访问。以前为了应对峰值流量，得准备足够的服务器，但99%的时间这些服务器都在"睡觉"。现在用云服务，平时只需要一台小配置的虚拟机，活动期间自动扩展到十几台，活动结束又缩回去。算下来一年能省好几万块。按需付费（Pay-as-you-go）的好处不仅仅是省钱，更重要的是降低了试错成本。想试个新技术？开个虚拟机测试一下，用完就删掉，可能就花几块钱。以前要试个新东西，得写申请、走流程、采购设备，等折腾完了热情都没了。资源池化（Resource Pooling）是按量计费能够实现的技术基础。云服务商把大量的物理资源放在一起，通过虚拟化技术分配给不同的用户。就像共享单车一样，一辆车可以被多个人在不同时间使用，提高了利用率，降低了成本。不过按量计费也有个坑，就是如果不注意控制，费用可能会超出预期。我听说过有个创业公司，因为代码bug导致无限循环调用云服务API，一夜之间产生了几万块的费用...所以现在各大云服务商都提供了费用预警（Cost Alert）功能，超过设定金额会自动通知。预付费（Prepaid）和后付费（Postpaid）是两种常见的计费模式。预付费就是先充值再使用，用完了再充；后付费是先使用再结算，每月出账单。对于用量比较稳定的业务，预付费通常有折扣；对于用量波动大的业务，后付费更灵活。现在很多云服务商还推出了Savings Plans这种新的计费模式，承诺在一定时间内使用一定金额的服务，就能享受更大的折扣。这对于有长期规划的企业来说是个不错的选择。微服务架构，化整为零的艺术微服务（Microservices）是把一个大的应用拆分成多个小的、独立的服务。每个服务负责特定的业务功能，可以独立开发、部署和扩展。传统的单体应用（Monolithic Application）就像一个大杂烩，所有功能都打包在一起。要修改一个小功能，整个应用都要重新部署。微服务架构下，各个服务相对独立，修改用户服务不会影响订单服务。不过微服务也不是银弹，它带来了新的复杂性。服务之间的通信、数据一致性、分布式事务等问题都需要考虑。我们公司从单体应用迁移到微服务架构的过程中就踩了不少坑，比如服务调用链路过长导致性能下降，服务间依赖关系复杂导致故障传播等。API网关（API Gateway）是微服务架构中的重要组件，作为所有服务的统一入口，负责请求路由、认证授权、限流熔断等功能。服务网格（Service Mesh）是处理服务间通信的基础设施层，通过代理的方式为服务间通信提供安全、快速、可靠的连接。Istio是目前比较流行的服务网格实现。数据存储的各种花样云计算环境下的数据存储也有很多种类。对象存储（Object Storage）适合存储非结构化数据，比如图片、视频、文档等。AWS S3、阿里云OSS都属于对象存储。它的特点是容量几乎无限，但不能像文件系统那样直接修改文件内容好处就是按量付费用多少付多少。块存储（Block Storage）提供原始的块级别存储，可以挂载到虚拟机上当作硬盘使用。性能比较好，适合数据库等对IO要求较高的应用。文件存储（File Storage）提供标准的文件系统接口，多个服务器可以同时挂载和访问。适合需要共享文件的场景。NoSQL数据库在云环境中也很流行，包括文档数据库（如MongoDB）、键值数据库（如Redis）、列族数据库（如Cassandra）、图数据库（如Neo4j）等。它们各有特点，适合不同的应用场景。常见的用户画像系统用的是MongoDB，因为用户属性比较灵活，关系型数据库的固定表结构不太适合。缓存层用的是Redis，因为它的性能特别好。不同的数据用不同的存储方案，这也是云计算时代的一个趋势。安全和合规，不能忽视的重要话题身份和访问管理（IAM）是云安全的基础，控制谁可以访问什么资源。通过角色、策略等机制实现细粒度的权限控制。虚拟私有云（VPC）为用户提供逻辑隔离的网络环境，就像在公共的云环境中划出一块私人领域。安全组（Security Group）类似于虚拟防火墙，控制进出虚拟机的网络流量。数据加密包括传输加密和存储加密。传输加密确保数据在网络传输过程中不被窃听，存储加密确保数据在磁盘上是加密保存的。合规性（Compliance）对很多企业来说都很重要，特别是金融、医疗等行业。云服务商通常会提供各种合规认证，比如ISO 27001、SOC 2、PCI DSS等。监控和运维，保障系统稳定运行可观测性（Observability）是最近比较热门的概念，包括监控（Monitoring）、日志（Logging）、链路追踪（Tracing）三个方面。监控主要关注系统的各种指标，比如CPU使用率、内存使用率、网络流量、响应时间等。Prometheus、Grafana是目前比较流行的监控组合。日志记录了系统运行过程中的详细信息，是排查问题的重要依据。ELK（Elasticsearch、Logstash、Kibana）是经典的日志处理方案，不过现在也有很多新的选择，比如Fluentd、Loki等。链路追踪可以跟踪一个请求在分布式系统中的完整调用路径，对于排查微服务架构下的性能问题特别有用。Jaeger、Zipkin是这个领域的代表产品。SLA（Service Level Agreement）服务等级协议，定义了服务的可用性标准。比如99.9%的可用性意味着一年中服务不可用的时间不能超过8.76小时。SLO（Service Level Objective）服务等级目标，是SLA的具体实现目标。SLI（Service Level Indicator）服务等级指标，是衡量服务性能的具体指标。我们团队现在每周都会开SLA回顾会议，分析上周的服务可用性情况，如果有故障会进行根因分析。刚开始觉得这个流程有点繁琐，但坚持下来确实对提升服务质量很有帮助。DevOps和CI/CD，开发运维一体化DevOps强调开发和运维的协作，通过自动化工具和流程提高软件交付的效率和质量。这不仅仅是技术问题，更是文化和组织结构的变革。CI/CD（Continuous Integration/Continuous Deployment）持续集成和持续部署是DevOps的重要实践。开发人员提交代码后，系统会自动进行构建、测试、部署等操作。Jenkins是最老牌的CI/CD工具，虽然界面比较古老，但功能很强大。GitLab CI、GitHub Actions是比较新的选择，与代码仓库集成得更紧密。基础设施即代码（Infrastructure as Code）把基础设施的配置用代码来描述和管理，提高了基础设施管理的一致性和可重复性。AWS CloudFormation、Terraform、Ansible、是这个领域的代表工具。我记得以前部署一个新环境，需要手工配置各种服务器、网络、数据库等，不仅耗时而且容易出错。现在用AWS CloudFormation写好配置文件，一个命令就能创建整套环境，而且每次创建的环境都是一模一样的。蓝绿部署（Blue-Green Deployment）是一种零停机部署方式，通过维护两套相同的生产环境来实现无缝切换。金丝雀部署（Canary Deployment）是另一种部署策略，先把新版本部署到少量服务器上，观察运行情况，没问题再逐步扩大范围。云原生，面向云环境设计的应用云原生（Cloud Native）是一种构建和运行应用的方法，充分利用云计算的优势。云原生应用通常具有微服务架构、容器化、动态编排、DevOps等特征。十二要素应用（The Twelve-Factor App）是构建云原生应用的方法论，包括代码库、依赖、配置、后端服务等12个方面的最佳实践。无服务器计算（Serverless）是云原生的一种极致形式，开发者只需要关注业务逻辑，不需要管理服务器。AWS Lambda、阿里云函数计算都属于Serverless服务。比如你公司有个数据处理任务，每天只运行几分钟，但需要很大的计算资源。以前要专门准备服务器，大部分时间都是闲置的。现在用AWS Lambda，任务来了就自动启动，处理完就自动释放，成本降低了很多。事件驱动架构（Event-Driven Architecture）通过事件来触发和连接各个服务，实现松耦合的系统设计，如AWS EventBridge。边缘计算和混合云边缘计算（Edge Computing）把计算能力部署到靠近数据源的地方，减少延迟，提高响应速度。比如自动驾驶汽车需要实时处理传感器数据，不可能把数据传到云端处理再返回结果。混合云（Hybrid Cloud）结合了公有云和私有云的优势，敏感数据放在私有云，其他业务使用公有云。多云（Multi-Cloud）是指同时使用多个云服务商的服务，避免厂商锁定，提高系统的可靠性。我们公司就是典型的混合云架构，核心业务系统部署在私有云上，确保数据安全和合规；一些边缘业务和临时需求使用公有云，灵活性更好。不过这样也带来了管理复杂性，需要统一的管理平台。人工智能和机器学习服务现在各大云服务商都提供了AI和ML相关的服务，降低了人工智能应用的门槛。机器学习平台（ML Platform）提供了从数据准备、模型训练、模型部署到模型管理的全流程服务。AWS SageMaker、Google AI Platform都是这类服务。预训练模型让没有AI背景的开发者也能快速集成AI能力，比如语音识别、图像识别、自然语言处理等。MLOps是机器学习运维，类似于DevOps，强调机器学习模型的自动化部署、监控和管理。说实话，AI这块我也在学习中，感觉变化特别快。前几天还在研究某个框架，过几天就出了新版本，功能完全不一样了。不过这也是技术发展的常态，保持学习的心态很重要。成本优化，云计算的必修课成本优化（Cost Optimization）是云计算使用中的重要话题。云服务按使用量付费的模式虽然灵活，但如果不注意控制，成本可能会失控。预留实例（Reserved Instance）通过预付费的方式获得更低的价格，适合长期稳定的工作负载。竞价实例（Spot Instance）使用云服务商的空闲资源，价格很便宜，但可能会被随时回收，适合容错性强的批处理任务。资源标签（Resource Tagging）通过给资源打标签来进行分类管理，便于成本分析和控制。总结云计算的专业术语确实很多，但掌握了这些基本概念，再看技术文档或者参加技术讨论就不会那么吃力了。其实很多概念并不复杂，关键是要理解它们解决了什么问题。比如微服务解决了单体应用难以扩展的问题，容器解决了环境一致性的问题，DevOps解决了开发和运维协作的问题。云计算技术发展很快，新的概念和工具层出不穷。我的建议是不要被这些术语吓到，也不要盲目追新。理解基本原理，根据实际需求选择合适的技术方案，这才是最重要的。另外，理论知识固然重要，但实践经验更宝贵。有机会的话多动手试试，搭个测试环境玩玩，踩踩坑，这样理解会更深刻。目前各大云厂商都有给测试套餐，大家可以免费注册来学习下。AWS 更是可以免费使用 40 余种核心云服务产品，长达 12 个月，这羊毛不薅白不薅！我写这篇文章的时候也在想，云计算发展到今天，已经不再是什么高深的技术，而是像水电一样的基础设施。掌握这些基本概念，对每个IT从业者来说都是必要的。希望这篇文章能帮到大家，如果觉得有用的话，记得点赞转发哦！有什么问题也欢迎在评论区讨论。想了解更多运维相关的干货内容，记得关注@运维躬行录，我会持续分享实用的技术经验和心得体会！
前言最近一个盆友提了一个需求，想要通过币安公开接口实时关注指定币种类价格，然后达到指定阈值后通过飞书发送消息。既然是盆友那忙肯定得帮。具体步骤币安接口分析https://developers.binance.com/docs/zh-CN/binance-spot-api-docs/web-socket-streams，这是api文档。有其他需求的朋友可以根据文档来对接指定api。因为高频要求，本次直接使用WebSocket调用，不使用rest api。关于python WebSocket库的使用可以参考这篇文章，https://blog.csdn.net/liaoqingjian/article/details/146908212核心功能分析高频数据采集：通过币安REST API获取所有现货交易对信息，这样避免手动更新币种通过WebSocket实时订阅交易对的ticker数据按批次订阅，每批100个交易对，避免单个连接订阅过多价格监控：记录每个交易对最近1分钟的价格历史计算1分钟内的价格涨跌幅当涨跌幅超过设定阈值(3%)时触发警报警报机制：通过飞书Webhook发送警报消息实现冷却期机制，避免同一交易对短时间内重复发送警报警报包含交易对、涨跌幅、价格变化、成交量等信息程序源码import json import asyncio import websockets import requests from datetime import datetime import time class BinanceSpotMonitor: def __init__(self, feishu_webhook_url): self.ws_url = "wss://stream.binance.com:9443/ws" self.rest_api_url = "https://api.binance.com/api" self.feishu_webhook_url = feishu_webhook_url self.market_data = {} self.batch_size = 100 # 每批订阅的交易对数量 self.price_history = {} # 存储价格历史 self.alert_threshold = 3.0 # 涨跌幅阈值，超过此值将触发推送 self.alert_cooldown = {} # 用于控制推送频率，避免同一币种短时间内多次推送 self.cooldown_period = 60 # 冷却时间(秒)，同一币种在此时间内只推送一次 async def get_spot_symbols(self): """获取所有现货交易对，仅保留USDT交易对""" try: response = requests.get(f"{self.rest_api_url}/v3/exchangeInfo") response.raise_for_status() symbols = [] for symbol in response.json()["symbols"]: # 只保留以USDT结尾的交易对 if (symbol["status"] == "TRADING" and symbol["isSpotTradingAllowed"] and symbol["quoteAsset"] == "USDT"): symbols.append({ "symbol": symbol["symbol"], "baseAsset": symbol["baseAsset"], "quoteAsset": symbol["quoteAsset"] }) return symbols except Exception as e: print(f"获取交易对列表失败: {e}") return [] async def connect_and_subscribe_batch(self, symbols_batch): """连接到币安WebSocket并订阅一批市场的ticker数据""" try: async with websockets.connect(self.ws_url) as websocket: # 创建订阅请求 subscribe_request = { "method": "SUBSCRIBE", "params": [f"{symbol['symbol'].lower()}@ticker" for symbol in symbols_batch], "id": 1 } # 发送订阅请求 await websocket.send(json.dumps(subscribe_request)) # 接收订阅确认 response = await websocket.recv() print(f"订阅确认批次 ({len(symbols_batch)}个交易对): {response}") # 持续接收数据 while True: try: message = await websocket.recv() await self.process_message(message, symbols_batch) except websockets.exceptions.ConnectionClosedError as e: print(f"连接关闭: {e}") # 尝试重新连接 await asyncio.sleep(5) break except Exception as e: print(f"处理消息时出错: {e}") except Exception as e: print(f"WebSocket连接错误: {e}") # 尝试重新连接 await asyncio.sleep(5) asyncio.create_task(self.connect_and_subscribe_batch(symbols_batch)) async def process_message(self, message, symbols_batch): """处理接收到的WebSocket消息""" data = json.loads(message) # 确保是ticker数据 if 's' in data: # 's' 是symbol的字段 symbol = data['s'] # 查找对应的baseAsset和quoteAsset symbol_info = next((s for s in symbols_batch if s["symbol"] == symbol), None) if not symbol_info: return base_asset = symbol_info["baseAsset"] quote_asset = symbol_info["quoteAsset"] # 确保只处理USDT交易对 if quote_asset != "USDT": return # 提取所需数据 price = float(data['c']) # 最新价格 volume_24h = float(data['q']) # 24小时成交量(报价资产) current_time = datetime.now() formatted_time = current_time.strftime('%Y-%m-%d %H:%M:%S') short_time = current_time.strftime('%H:%M:%S') # 处理价格历史 if symbol not in self.price_history: self.price_history[symbol] = { "prices": [(price, current_time)], "last_alert": None } else: # 添加当前价格到历史记录 self.price_history[symbol]["prices"].append((price, current_time)) # 清理超过1分钟的历史数据 self.price_history[symbol]["prices"] = [ (p, t) for p, t in self.price_history[symbol]["prices"] if (current_time - t).total_seconds() <= 60 ] # 如果有足够的历史数据，计算1分钟涨跌幅 if len(self.price_history[symbol]["prices"]) >= 2: oldest_price, oldest_time = self.price_history[symbol]["prices"][0] time_diff = (current_time - oldest_time).total_seconds() # 确保有足够的时间间隔(至少30秒)来计算有意义的涨跌幅 if time_diff >= 30: price_change_percent = ((price - oldest_price) / oldest_price) * 100 # 存储市场数据 self.market_data[symbol] = { 'symbol': symbol, 'base_asset': base_asset, 'quote_asset': quote_asset, 'price': price, 'old_price': oldest_price, 'price_change_percent_1min': price_change_percent, 'volume_24h': volume_24h, 'last_update': formatted_time } # 检查是否需要发送警报 await self.check_and_send_alert(symbol, price_change_percent, price, oldest_price, volume_24h, short_time) async def check_and_send_alert(self, symbol, price_change_percent, current_price, old_price, volume_24h, time_str): """检查是否需要发送警报并发送""" # 检查涨跌幅是否超过阈值 if abs(price_change_percent) >= self.alert_threshold: # 检查冷却期 current_time = time.time() if symbol in self.alert_cooldown and current_time - self.alert_cooldown[symbol] < self.cooldown_period: return # 在冷却期内，不发送警报 # 更新冷却时间 self.alert_cooldown[symbol] = current_time # 准备警报消息 direction = "上涨" if price_change_percent > 0 else "下跌" formatted_volume = f"{volume_24h:,.2f}" # 构建飞书消息 current_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S') base_asset = self.market_data[symbol]['base_asset'] quote_asset = self.market_data[symbol]['quote_asset'] pair_name = f"{base_asset}/{quote_asset}" message = f"【test 现货推送】\n{current_date}\n币安:{base_asset}/1m内{direction}/\n{abs(price_change_percent):.2f}%\n价格:{old_price:.8f}/{current_price:.8f}\n成交:5m${formatted_volume}\n时间:{time_str}" # 发送到飞书 await self.send_to_feishu(message) print(f"已发送警报: {symbol} 1分钟{direction} {abs(price_change_percent):.2f}%") async def send_to_feishu(self, message): """发送消息到飞书机器人""" try: headers = {'Content-Type': 'application/json'} payload = { "msg_type": "text", "content": { "text": message } } response = requests.post(self.feishu_webhook_url, headers=headers, data=json.dumps(payload)) if response.status_code != 200: print(f"发送飞书消息失败: {response.status_code}, {response.text}") return response.status_code == 200 except Exception as e: print(f"发送飞书消息出错: {e}") return False async def start(self): """启动市场数据监控""" # 获取所有交易对 symbols = await self.get_spot_symbols() print(f"获取到 {len(symbols)} 个USDT交易对") # 将交易对分批处理 for i in range(0, len(symbols), self.batch_size): batch = symbols[i:i + self.batch_size] asyncio.create_task(self.connect_and_subscribe_batch(batch)) # 避免同时创建太多连接 await asyncio.sleep(1) async def main(): # 替换为你的飞书机器人webhook URL feishu_webhook_url = "" monitor = BinanceSpotMonitor(feishu_webhook_url) await monitor.start() # 保持程序运行 while True: await asyncio.sleep(3600) if __name__ == "__main__": asyncio.run(main())仅供学习参考，切勿用于非法用途
前言有几个微信需要多开登录，每次都要按回车点击微信多开。把握不住，老是会开很多个。用python实现多开，精准控制需要打开的数量，但是打开前先要关闭所有微信。运行图片：源码import tkinter as tk import subprocess import os import time from tkinter import messagebox, ttk class WeChatMultiOpener: def __init__(self, root): self.root = root self.root.title("微信多开器") self.root.geometry("300x200") self.root.resizable(False, False) # 设置界面 self.setup_ui() def setup_ui(self): # 主框架 main_frame = tk.Frame(self.root, padx=20, pady=20) main_frame.pack(fill=tk.BOTH, expand=True) # 标题标签 title_label = tk.Label(main_frame, text="微信多开工具", font=("Arial", 14, "bold")) title_label.pack(pady=10) # 数量选择框架 count_frame = tk.Frame(main_frame) count_frame.pack(pady=10, fill=tk.X) # 数量标签 count_label = tk.Label(count_frame, text="打开数量:") count_label.pack(side=tk.LEFT, padx=(0, 5)) # 数量输入框 self.count_var = tk.StringVar(value="2") count_spinbox = ttk.Spinbox( count_frame, from_=1, to=10, width=5, textvariable=self.count_var ) count_spinbox.pack(side=tk.LEFT) # 按钮框架 button_frame = tk.Frame(main_frame) button_frame.pack(pady=10) # 打开微信按钮 open_button = tk.Button(button_frame, text="打开微信", width=12, command=self.open_multiple_wechat, bg="#1aad19", fg="white") open_button.pack(side=tk.LEFT, padx=5) # 关闭所有微信按钮 close_button = tk.Button(button_frame, text="关闭所有微信", width=12, command=self.close_all_wechat) close_button.pack(side=tk.LEFT, padx=5) # 状态标签 self.status_var = tk.StringVar(value="就绪") status_label = tk.Label(main_frame, textvariable=self.status_var, fg="gray") status_label.pack(pady=(10, 0)) def open_multiple_wechat(self): try: # 获取用户输入的数量 count_text = self.count_var.get().strip() try: count = int(count_text) if count < 1 or count > 10: messagebox.showwarning("警告", "请输入1-10之间的数字") return except ValueError: messagebox.showwarning("警告", "请输入有效的数字") return wechat_path = r"C:\Program Files\Tencent\WeChat\WeChat.exe" # 检查路径是否存在 if not os.path.exists(wechat_path): print("未找到微信程序，请检查安装路径") return # 更新状态 self.status_var.set(f"正在打开 {count} 个微信实例...") self.root.update() # 打开指定数量的微信 for i in range(0,count): # 使用命令行参数启动第二个微信实例 self. open_multiple() # 稍微延迟，避免同时启动太多实例 self.status_var.set(f"已成功打开 {count} 个微信实例") except Exception as e: self.status_var.set("操作失败") messagebox.showerror("错误", f"启动微信失败: {str(e)}") def open_multiple(self): # 微信安装路径，请根据实际情况修改 wechat_path = r"C:\Program Files\Tencent\WeChat\WeChat.exe" # 检查路径是否存在 if not os.path.exists(wechat_path): print("未找到微信程序，请检查安装路径") return # 使用命令行参数启动第二个微信实例 subprocess.Popen([wechat_path, "/multi"]) def close_all_wechat(self): try: self.status_var.set("正在关闭所有微信...") self.root.update() subprocess.run("taskkill /f /im WeChat.exe", shell=True) self.status_var.set("已关闭所有微信实例") except Exception as e: self.status_var.set("操作失败") messagebox.showerror("错误", f"关闭微信失败: {str(e)}") if __name__ == "__main__": root = tk.Tk() app = WeChatMultiOpener(root) root.mainloop()写在最后希望能够帮到你们，有需要的朋友也可以私信后台，找我要打包好的程序直接运行。