Posts

前言 开发调试阶段，运行 Flask 的方式多直接使用 app.run()，但 Flask 内置的 WSGI Server 的性能并不高。对于生产环境，一般使用 gunicorn。如果老项目并不需要多高的性能，而且用了很多单进程内的共享变量，使用 gunicorn 会影响不同会话间的通信，那么也可以试试直接用 gevent。 ...

前言 在 FastAPI 应用中执行定时任务，通常可以选择 celery，但 Celery 相对重量级，且需要依赖 Redis 或 RabbitMQ 等消息队列。当服务规模较小，且原本未使用 Redis 或 RabbitMQ 时，仅为定时任务功能引入这些依赖会增加额外的运维成本。 ...

前言 SQLite 和 PostgreSQL 都支持 UPSERT 操作，即"有则更新，无则新增"。冲突列必须有唯一约束。 语法： PostgreSQL: INSERT ... ON CONFLICT (column) DO UPDATE/NOTHING SQLite: INSERT ... ON CONFLICT(column) DO UPDATE/NOTHING。注意括号位置 场景 PostgreSQL SQLite 说明 基本 UPSERT ON CONFLICT (col) DO UPDATE SET ... ON CONFLICT(col) DO UPDATE SET ... 括号位置略有不同 冲突忽略 ON CONFLICT (col) DO NOTHING ON CONFLICT(col) DO NOTHING 相同 引用新值 EXCLUDED.col excluded.col PostgreSQL 大写，SQLite 小写 返回结果 RETURNING * RETURNING * 相同 条件更新 WHERE condition 不支持 WHERE SQLite 限制 注意事项 冲突列必须有唯一约束 PostgreSQL 和 SQLite 的语法相似，但仍有细微差别。使用原生 SQL 时需要注意。 SQLite 在 UPSERT 时不支持 WHERE 子句，需要改用 CASE 表达式或应用层过滤。 SQLite 3.35+ 版本才支持 RETURNING EXCLUDED 和 RETURNING EXCLUDED EXCLUDED 表示冲突时被拦截的新值。 ...

前言 此前在 MCP 官网就注意到官方提供了 Go SDK，近期由于在 Python 环境下开发 MCP Server 有点"审美疲劳"，因此决定使用 Go 语言尝尝鲜。 从个人实际体验来看，Go 语言在并发处理方面确实具有显著优势：无需纠结于同步阻塞、异步事件循环、多进程多线程通信等复杂的并发问题，goroutine 一把梭哈。同时，Go 语言的部署也非常便捷，编译后生成的静态二进制文件具有良好的可移植性，可以在不同环境中直接运行。 然而，这种便利性也伴随着一定的代价。相较于 Python，使用 Go 语言实现 MCP 功能相对复杂一些，开发效率略低。这就是软件工程中的经典权衡了：运行成本与开发成本往往难以兼得，需要根据具体场景进行取舍。 MCP 协议简介 可能都耳熟能详了，但以防还有不熟悉的朋友，先简单介绍下MCP Model Context Protocol (MCP) 是一种标准化的协议，旨在为 AI 模型提供统一的工具调用接口。通过 MCP，开发者可以将各种工具、服务和数据源暴露给 AI 模型，使其能够执行超出基础语言模型能力范围的操作。MCP 支持多种传输协议，包括 HTTP 和 Stdio，为不同场景下的集成提供了灵活性。 一个简单的 MCP Server 示例 MCP 官方 Go SDK 在定义工具（Tool）时，要求明确指定输入参数和输出结果的数据结构。对于功能较为简单的工具，也可以直接使用 any 类型。以下是一个完整的 MCP Server 示例，提供了三个实用工具： getCurrentDatetime：获取当前时间，返回 RFC3339 格式（2006-01-02T15:04:05Z07:00）的时间戳字符串。由于不需要输入参数，因此参数类型定义为 any，输出同样使用 any 类型。 getComputerStatus：获取当前系统的关键信息，包括 CPU 使用率、内存使用情况、系统版本等。该工具接受一个 CPUSampleTime 参数，对应的输入结构体为 GetComputerStatusIn，输出结构体为 GetComputerStatusOut（Go SDK 的示例中通常采用 xxxIn 和 xxxOut 的命名约定来区分工具的输入输出结构体）。 ...

前言 在本地虚拟机环境中使用 kubeadm 搭建 Kubernetes 集群是学习和实验的理想选择。考虑到实际应用场景中可能存在的网络限制以及镜像构建需求，本文详细记录了在完全离线环境下部署单节点 Kubernetes 集群的完整过程。通过集成 Harbor 私有镜像仓库，所有 Kubernetes 组件镜像均从本地 Harbor 实例拉取，确保部署过程的可靠性和可重复性。 ...

前言 在实际业务中，根据 tracking_id 追溯一条请求的完整处理路径是比较常见的需求。借助 Flask 自带的全局对象 g 以及钩子函数，可以很容易地为每条请求添加 tracking_id，并在日志中自动记录。 ...

前言 在异步方法中直接调用同步方法会阻塞整个事件循环，导致应用在执行同步方法期间无法处理任何其他并发请求，严重影响服务的整体性能和响应能力。 为了解决这个问题，核心思路是将同步方法交给外部线程池或进程池执行，避免阻塞主事件循环。 方法 1：使用 asyncio.to_thread Python 3.9 及以后版本可以使用 asyncio.to_thread 方法，将同步函数运行在独立的线程中，并返回一个可供 await 的协程对象 import asyncio import time from fastapi import FastAPI app = FastAPI() def sync_task(name: str): time.sleep(2) return f"Hello {name}, sync task done!" @app.get("/async-call") async def async_endpoint(): result = await asyncio.to_thread(sync_task, "World") return {"message": result} 方法 2：直接定义同步路由 FastAPI 支持定义同步路由，FastAPI 会自动在一个外部线程池中运行该函数。不过出于代码整体设计和一致性的考虑，不建议在异步项目中混用同步路由。 方法 3：使用 run_in_threadpool FastAPI 基于 Starlette，而 Starlette 提供了一个工具函数 run_in_threadpool，这种方式类似于 asyncio.to_thread，在某些老版本的 FastAPI 或特定的 contextvars 传递场景下更常用。 from fastapi.concurrency import run_in_threadpool @app.get("/method3") async def starlette_endpoint(): result = await run_in_threadpool(sync_task, "Starlette") return {"message": result} 方法 4：使用进程池 对于 CPU 密集型任务，应该使用多进程 ProcessPoolExecutor 来处理 import concurrent.futures import math from fastapi import FastAPI app = FastAPI() # 创建一个全局进程池 executor = concurrent.futures.ProcessPoolExecutor() def cpu_intensive_calculation(n: int): # 模拟重度 CPU 计算 return sum(math.isqrt(i) for i in range(n)) @app.get("/cpu-bound-task") async def cpu_task(): loop = asyncio.get_running_loop() result = await loop.run_in_executor(executor, cpu_intensive_calculation, 10**7) return {"result": result}