尧图网页设计 
简介面向 API 服务运营与开发者的全新二开版接口管理系统源码基于 NginxPHP7.4MySQL5.7 环境测试运行访问域名即可完成安装项目重点修复原版 API 鉴权漏洞与源代码暴露风险并升级为现代化响应式前后端设计适配 PC 与移动多终端管理场景帮助使用者搭建安全、易用的接口管理后台。资源包共 446 个文件以 zip 压缩包形式提供主要文件包括 84 个 PHP 后台逻辑与接口文件、217 个 JavaScript 交互脚本、48 个 CSS 样式表另附 SQL 数据库、字体及图标资源整体约 20.17MB目前已有 145 人学习。二开内容覆盖完整新增 API 分类功能实现结构化接口管理API 详情页集成在线调试工具已登录用户自动获取密钥后台提供 QPS 限速开关为后续流量控制预留扩展同时修复了邮件标题显示问题并美化支付成功页与通知邮件。整套源码目录规范、前端组件选择成熟既适合开发者深入研读接口平台设计思路也可快速二次开发构建自有 API 计费管理系统。1. 全新二开版API管理系统源码为什么卡住大家的总是计费而不是转发做过API平台的人都有个体会把一个接口开放出去很简单但让它按调用量收钱难度会直接跳到另一个量级。标题里这三个词——二开版、API计费、全开源——其实对应三件事一套能对外售卖API能力的业务系统、一套能算清账的计量链路、以及一套可以随意改动的完整代码。这类源码的真正价值从不在网关转发那一层而在计费闭环谁调了多少次、余额扣到哪、套餐怎么算、密钥怎么发、对账怎么对。这篇文章要聊清楚的就是这件事读者是手上有API服务想对外变现的团队或者需要把API能力分发出去、向内部部门摊派调用成本的技术负责人。2. 二开版API管理系统源码的架构与部署先用三条线拆清源码再启动2.1 拆源码的三条业务线管理后台、API网关、计费Worker二开版API管理系统源码拿到手第一件事不是急着配环境而是把代码按业务线拆开。常见做法是分成管理后台、API网关、计费服务三块少数实现会把计费直接塞进网关进程里但那种结构后面对账会很难受。管理后台负责的是“卖”的动作创建商户、签发密钥、配置套餐、查看调用报表。API网关负责“转发”的动作校验签名、检查限流、把请求转发到真正的上游服务再把响应原样返回。计费服务负责“算账”的动作从Redis里实时扣减余额、把流水异步落进MySQL、触发余额不足回调。三条线的关系是请求先进网关网关在转发前调用计费检查转发完成后计费Worker异步写流水管理后台只读库和缓存不参与在线链路。模块职责数据落点管理后台商户、套餐、密钥、报表MySQL 业务库API网关签名校验、限流、转发Redis 实时状态计费 Worker扣费、流水、回调、对账Redis 扣减MySQL 落账拆清楚这条线之后碰到问题就知道该看哪个进程的日志。比如调用方报“余额扣了但请求失败”问题大概率不在网关而在计费Worker的补偿逻辑或者转发之后没有把失败状态传回计费模块。2.2 本地跑通最小部署从环境依赖到三个进程同时拉起来假设这套二开版源码是PythonFastAPI MySQL Redis 的技术栈这是目前这类系统里很常见的一种选型下面以它为例讲部署。环境要求是 Python 3.10、MySQL 8.0、Redis 6.x内存至少预留 1GB 给Redis存计费状态。# 1. 创建虚拟环境并安装依赖requirements.txt 由源码包自带 python3 -m venv .venv source .venv/bin/activate pip install -r requirements.txt # 2. 复制环境配置模板改成你自己的连接信息 cp .env.example .env # 编辑 .envDB_HOST、DB_PORT、DB_USER、DB_PASSWORD、REDIS_HOST、REDIS_PORT # 这三个参数必改DB_PASSWORD、ADMIN_PASSWORD、SECRET_KEY # 3. 初始化数据库表结构执行后 MySQL 里会生成 merchant/app_key/plan/billing_log 等表 python scripts/init_db.py # 4. 启动三个进程管理后台监听 8000API网关监听 9000计费Worker常驻后台 uvicorn admin.app:app --host 0.0.0.0 --port 8000 uvicorn gateway.app:app --host 0.0.0.0 --port 9000 python worker/billing_worker.py 第一步和第二步决定了后面所有账能不能算对。DB_PASSWORD 和 SECRET_KEY 必须改SECRET_KEY 会参与回调通知的签名生成如果沿用源码包里的默认值下游调用方拿到源码就能伪造回调。Admin 后台监听 8000 端口网关监听 9000 端口两者物理隔离是为了后续给网关单独做水平扩容计费Worker独立进程则是为了避免在线请求阻塞时扣费延迟。启动完三个进程后先确认 Redis 里能读到套餐缓存、MySQL 里能看到四张业务表再进下一步初始化数据。这里最容易翻车的地方是 MySQL 字符集没设成 utf8mb4导致下游传 emoji 进去时直接 500我一般会在 init_db.py 里强制指定DEFAULT CHARSETutf8mb4。2.3 初始化最小业务数据一个商户、一把密钥、一个按次套餐部署跑通之后离第一笔计费还差最关键的一步数据初始化。二开版源码一般会自带初始化脚本但如果脚本只建表不造业务数据你就得手动插入。下面是一套最精简的初始化SQL覆盖商户、密钥、套餐、流水四张核心表这也是计费链路能跑通的最少数据。-- 商户假设这是你的第一个客户 INSERT INTO merchant (name, status) VALUES (测试商户, 1); -- 套餐按次计费每次 0.01 元余额 10000 次每分钟限 120 次 INSERT INTO plan (name, price, quota, rate_limit, window_sec) VALUES (按次体验包, 0.0100, 10000, 120, 60); -- 密钥app_id 发给下游secret 只在服务端保存 INSERT INTO app_key (app_id, secret, merchant_id, plan_id, status) VALUES (app_test_001, sk_live_9f8e7d6c5b4a, 1, 1, 1); -- 流水表本身不需要初始化但这个自增主键和幂等唯一键很关键 -- request_id 是幂等键同一条请求扣费两次会直接报错 CREATE TABLE IF NOT EXISTS billing_log ( id BIGINT PRIMARY KEY AUTO_INCREMENT, request_id VARCHAR(32) NOT NULL UNIQUE, app_id VARCHAR(32) NOT NULL, cost INT NOT NULL DEFAULT 1, status TINYINT DEFAULT 0, created_at DATETIME(3) );商户表只存身份信息计费不看它套餐表里的 price 是每次调用的单价quota 是剩余可用次数rate_limit 参与限流判定app_key 表把下游调用方和套餐绑定在一起一个商户可以开多个应用丢给不同项目用。四张表里最重要的是 billing_logcost 字段我建议用整数存“分”或“厘”不要用浮点数哪怕价格表里写的是 0.0100 元落账时也要转成 1 分不然月底对账时浮点误差会让你怀疑人生。初始化完成后拿 app_id 和一个任意签名去调网关的健康检查接口能看到 200 就说明这一整套最小闭环已经跑通可以进入计费模块的实现细节了。3. 计费模块的实现逻辑把API调用量变成余额扣减需要算清四笔账3.1 先定计费模型按次、包周期还是阶梯决定了表的字段怎么设计计费模型是整个系统的地基选错后面全是坑。最常见的三种模型是按次计费、包周期套餐、阶梯计费。按次计费最简单每次调用扣固定金额适合刚起步的小商户包周期是按月/按年买固定额度超额后拒绝服务或转按次阶梯计费则是在一个周期内累计调用量跨过某个阈值后单价自动下降适合中大型API服务。模型优点缺点适合场景按次实现简单对账直观大客户嫌贵小商户、测试期包周期收入可预期超额对账麻烦大部分SaaS API阶梯大客户愿意用需要周期累计调用量大的老客户二开版源码里这三个模型一般会同时存在。实现上按次计费只需要一个 quota 字段递减包周期要记录周期开始时间和重置时间阶梯计费麻烦得多需要一张独立的累计表按自然日或自然月汇总调用量。我的建议是先只开放按次和包周期两种阶梯计费等客户真到了那个量级再二开否则前期光是统计口径就能耗掉你两周。3.2 用Redis做实时扣减用MySQL做最终流水双写才能既快又稳计费模块的核心矛盾是“快”和“稳”。一次API调用会在几十毫秒内结束扣费必须同步完成用MySQL在在线链路里做行锁扣减并发一高就死锁但如果只扣Redis进程一挂账就丢了。所以常见做法是两层配合Redis负责实时扣减MySQL流水负责最终对账。# billing_middleware.py import uuid import time import redis from fastapi import Request, HTTPException # db1 单独给计费用不让业务缓存干扰计费数据 r redis.Redis(host127.0.0.1, port6379, db1, decode_responsesTrue) async def check_billing(request: Request): # 下游调用方在请求头里带上平台签发的 app_id app_id request.headers.get(X-App-Id) if not app_id: raise HTTPException(status_code401, detailmissing app_id) plan_key fplan:{app_id} # 套餐信息在 Redis 里用 hash 缓存quota 剩余次数、price_val 单价、rate_limit 限流阈值 # 示例hset plan:app_test_001 quota 10000 price_val 1 rate_limit 120 quota int(r.hget(plan_key, quota) or 0) if quota 0: # 402 是 HTTP 里专门表示“余额不足”的语义比 500 好排查 raise HTTPException(status_code402, detailinsufficient quota) # 生成请求ID幂等键流水表里靠它去重 request_id uuid.uuid4().hex # 先扣后记INCRBY 是 Redis 原子操作不用 Lua 也能防超卖 r.hincrby(plan_key, quota, -1) # 把扣费动作追加到本应用的临时流水队列Worker 异步刷 MySQL r.lpush(fbilling:queue:{app_id}, f{request_id}:{int(time.time())}) # 上游转发成功后网关把 request_id 一并带到响应头方便调用方排查 return {request_id: request_id, app_id: app_id}这里有两个关键参数必须解释清楚。db1是给Redis单独划了一个逻辑库计费相关的 key 全部放在 db 1管理后台的缓存和会话放 db 0这样清理缓存时不会误删计费数据。hincrby加-1是原子操作两个并发请求同时进来时不会都读到同一个 quota 值这是避免超卖的基本功。扣费时机的选择值得多说一句先扣后记即先扣余额再转发。如果先转发再扣费上游响应已经返回了但扣费失败调用方等于免费调了一次先扣费后转发虽然可能出现“扣了钱但上游超时”的情况但这个问题可以在避坑章节里用补偿机制解决比免费调用可控得多。计费中间件返回的 request_id 还会放在响应头里调用方工单排查时凭这个ID就能在两分钟内定位到流水。3.3 余额不足与回调通知别让调用方糊里糊涂断了服实时扣减之外二开版API管理系统还必须有主动通知能力调用方余额见底时平台不只返回402还要主动推一个回调给调用方让他去充值。回调通知最怕两件事没推到、重复推。所以回调接口必须支持重试并且用 request_id 做幂等。# callback.py import hashlib import hmac import time import requests def send_balance_callback(app_id: str, merchant_callback_url: str, secret: str): # 回调体只有四个字段app_id、额度剩多少、时间戳、签名 payload { app_id: app_id, quota_left: 0, timestamp: int(time.time()), } # 签名用 HMAC-SHA256secret 是 app_key 表里那个签名防伪造 raw f{app_id}{payload[timestamp]}{payload[quota_left]} payload[sign] hmac.new( secret.encode(), raw.encode(), hashlib.sha256 ).hexdigest() # 重试 3 次间隔 5 秒、30 秒、300 秒不阻塞在线请求 for interval in (5, 30, 300): try: resp requests.post(merchant_callback_url, jsonpayload, timeout10) # 调用方回调接口要返回 HTTP 200 才算成功否则进入下一次重试 if resp.status_code 200: return True except requests.RequestException: pass time.sleep(interval) return False回调带了时间戳所以调用方收到回调后必须校验时间偏差超过五分钟直接丢弃。签名算法与网关签名保持一致调用方可以用同一套验签逻辑处理。回调场景里最容易踩的坑是回调接口地址本身配置错误二开版里这个地址通常在商户表里但有些源码包会把它写死在配置文件里导致所有商户共用一个回调地址测试时能收到、上线后全部推给了同一家排查时要先去确认回调地址是商户级别还是平台级别的配置。4. 网关鉴权与路由转发签名校验、限流和计费如何联动4.1 基于HMAC-SHA256的签名校验app_id公开secret只在服务端API管理系统的第一道门是鉴权。平台给每个调用方签发一个 app_id 加 secretapp_id 放在请求头里明文传输secret 不能出现在请求里否则抓包的人直接就能伪造调用。所以要做签名调用方把请求参数、时间戳、随机数一起算一个 HMAC-SHA256 签名服务端用同样的 secret 重算比对。# auth.py import hashlib import hmac import time from fastapi import Header, HTTPException def verify_sign( app_id: str Header(..., aliasX-App-Id), timestamp: str Header(..., aliasX-Timestamp), nonce: str Header(..., aliasX-Nonce), sign: str Header(..., aliasX-Sign), secret: str , ): # 时间窗 300 秒超过就拒绝。防止旧的合法请求被无限重放 if abs(int(time.time()) - int(timestamp)) 300: raise HTTPException(status_code401, detailtimestamp expired) # 拼接顺序必须固定两端用的规则要一模一样 raw f{app_id}{timestamp}{nonce} expect hmac.new(secret.encode(), raw.encode(), hashlib.sha256).hexdigest() # compare_digest 防时序攻击不能用 直接比 if not hmac.compare_digest(expect, sign): raise HTTPException(status_code401, detailsign mismatch) # 生产环境里nonce 要放进 Redis 做去重同一个 nonce 只能用一次 # 这里省略 Redis 部分避免与计费中间件抢连接影响性能 return app_id签名校验里最容易漏的是 nonce 去重。时间戳窗口 300 秒意味着一个合法签名在 5 分钟内都有效如果不去重攻击者抓包后能在这 5 分钟内无限重放。常见的做法是把 nonce 写进 Redis 的 set 结构窗口时间设置为与签名时间窗一致。secret 不落请求不代表可以明文存数据库二开版里 app_key 表的 secret 字段至少要做哈希处理我见过不少源码把 secret 以明文存一旦数据库被拖走全部调用方密钥直接暴露。4.2 限流不只看QPS把套餐剩余次数与滑动窗口绑在一起网关的第二道关卡是限流。传统限流只关心每秒请求数但在API管理系统里限流必须和套餐联动同样的 120 次每分钟按次计费和包周期套餐的触发条件不一样。二开版常见做法是把套餐表里的 rate_limit 和 window_sec 读出来作为限流阈值传给网关。# rate_limit.py import time import redis r redis.Redis(host127.0.0.1, port6379, db1, decode_responsesTrue) def sliding_window_check(app_id: str, limit: int 120, window_sec: int 60): # 用有序集合存每次调用的时间戳天然实现滑动窗口 key frate:{app_id} now time.time() # 先清掉窗口外的老记录避免 zset 无限膨胀 r.zremrangebyscore(key, 0, now - window_sec) count r.zcard(key) if count limit: raise PermissionError(rate limited) # 当前请求的时间戳入集合zadd 的 score 和 member 都用时间戳 r.zadd(key, {str(now): now}) # 窗口结束 60 秒后整把 key 过期防止死数据占内存 r.expire(key, window_sec 60) return count 1这里不是用 Redis 的 INCR 加过期时间做固定窗口而是用 zset 做滑动窗口差别在于固定窗口在窗口切换瞬间会出现 2 倍突发流量。zset 方案每次请求都要执行一次 zremrangebyscore窗口越大成本越高所以 window_sec 设定为 60 秒、limit 在 1000 以下时性能没有问题。如果套餐有更高限额我一般会把限流窗口改成 1 秒粒度加 60 秒聚合计数两种方案配合而不是只用滑动窗口。限流与计费的联动体现在顺序上先做限流检查再做计费扣减。限流失败不扣费计费失败不转发。两个中间件用同一个 app_id 做 key但 key 前缀不同一个是rate:一个是plan:避免互相覆盖。4.3 转发时替换上游密钥对接大模型API服务的统一出口网关的最后一步是转发到上游真实API服务。这一步是二开版系统和普通反向代理的关键区别普通代理只改 URLAPI计费系统还要在转发时替换认证信息。下游调用方用的是平台签发的 app_id而上游服务只认平台自己采购的 key两个 key 不能混。# forwarder.py import httpx async def forward_to_upstream(api_config: dict, headers: dict, body: bytes): # api_config 里是上游地址、请求方法、超时、以及平台集中管理的上游key # 常见做法是在管理后台维护一张上游API配置表对接DeepSeek、智谱这类大模型API服务 async with httpx.AsyncClient(timeoutapi_config[timeout]) as client: resp await client.request( methodapi_config[method], urlapi_config[url], headers{**headers, Authorization: fBearer {api_config[upstream_key]}}, contentbody, ) # 网关原样返回上游响应不解析body这样对下游调用方来说是透明的 return resp.status_code, resp.content转发层的超时参数必须比网关本身超时短不然网关等上游、下游等网关会出现两层超时叠加导致的 504。比如网关整体超时设 15 秒上游转发超时就要设 13 秒留 2 秒给签名校验和计费扣减。二开版里上游 key 应该集中放在配置表里由平台统一采购和续费不能把上游 key 下发给调用方否则调用方直接绕过平台调上游这个系统就变成纯转发工具而不是计费平台了。5. 二开版源码避坑部署、鉴权、计费最容易翻车的五个地方5.1 鉴权与路由方向上常踩的坑上游key配错、超时重试、时间窗过期现象一下游调用方明明拿了合法的 app_id转发时报llm-deepseek: no api key for provider route deepseek-official。原因二开版把上游 key 放在配置表后运维把平台自己的上游 key 误配到了下游应用的 app_key 上网关转发时没有做替换上游收到的是不存在的 key。解决转发前确认 api_config 表里 upstream_key 字段有值并且生产环境不要用 init 脚本里的测试 key。我这个坑实际踩过两次第一次是以为上游服务的问题排查了半小时才发现是 key 配错了桶。现象二一次请求因为网络抖动超时调用方重试后账单里出现两条扣费记录。原因计费中间件在转发前先扣费超时后客户端重试网关又走了一遍扣费逻辑同一个业务请求被扣了两次。这是“先扣后记”方案最典型的副作用。解决扣费之前先查 billing_log 的 request_id如果客户端重试时带上了原始 request_id直接返回第一次的响应不算钱。所以网关的响应头一定要返回 request_id并且引导调用方在重试时带上这个ID。现象三上游返回 400报this models maximum context length is 1048576 tokens这类参数超长错误。原因签名校验成功后请求转发给上游大模型API但请求体太大或参数超限上游拒绝了。这个不算计费系统自身缺陷但会让调用方误以为网关有问题。解决网关只转发不解析但可以在管理后台配置“上游错误码白名单”把这类 400 错误透传给调用方的同时标记为“不扣费”。是否对 4xx 扣费要提前定好规矩我一般遵循 5xx 不扣、4xx 扣一半、2xx必扣具体在二开版里用 status 字段区分。5.2 计费与对账方向上常踩的坑Redis丢了额度、流水丢尾部、统计口径打架现象四调用量远没到套餐上限却突然全部返回 402 余额不足。原因Redis 里的 plan: 前缀 key 因为内存淘汰策略被回收了计费中间件读到的 quota 是 0直接拒绝服务。二开版源码很多默认使用 Redisallkeys-lru淘汰策略计费 key 和普通缓存一起竞争内存一旦内存压力上来最先被淘汰的就是这些长尾 key。解决给计费 key 单独提一个 Redis 实例或者至少把淘汰策略改成volatile-lru并且写一个启动预热脚本把所有 enabled 状态套餐的 quota 从 MySQL 刷进 Redis。现象五计费 Worker 重启后Redis 流水队列里最后十秒的数据丢了账单少记了调用量。原因Worker 用 lpop 从billing:queue:{app_id}取数据进程在 lpop 之后、写 MySQL 之前崩溃记录就丢了。解决把 Redis list 换成 Redis Streams用 XREADGROUP XACK 做消费确认Worker 重启后先 XAUTOCLAIM 捞回未确认消息再继续处理。这个改造大概要花半天但能避免月底对账时账单少几十万的尴尬。计费统计口径还有一个长期存在的坑限流按滑动窗口算账单按自然日聚合两边数字永远对不上。原因60 秒滑动窗口在 23:59:30 到 00:00:30 这个区间内跨了两个自然日限流统计被切成了两段而账单按自然日切分后就少计或重计了一部分。解决在报表里同时标注“按自然日”和“按调用窗口”两种口径别强行让它们相等如果客户要求严格对账把计费周期改成从每日零点开始的固定 60 秒窗口牺牲一点滑动窗口的平滑性换取对账一致性。这个选择没有绝对正确关键是要在文档里写清楚。6. 上线前验证用并发脚本核对“扣费次数成功调用次数”再谈卖API6.1 200并发压测脚本计费上限要压出来上线前一天至少要做一次并发压测不是压 QPS而是压计费一致性。我常用的办法是起 200 个线程同时调同一个接口然后核对三个数字成功响应数、Redis 里扣减的额度、MySQL 流水条数。# perf_check.py import threading import requests URL http://127.0.0.1:9000/api/v1/demo HEADERS {X-App-Id: app_test_001, X-Sign: test_sign} ok fail 0 lock threading.Lock() def call(): global ok, fail try: r requests.get(URL, headersHEADERS, timeout5) with lock: if r.status_code 200: ok 1 else: fail 1 except Exception: with lock: fail 1 threads [threading.Thread(targetcall) for _ in range(200)] for t in threads: t.start() for t in threads: t.join() print(fok{ok} fail{fail})跑完这个脚本如果 ok fail 200 但 MySQL 流水条数小于 ok 数说明计费 Worker 消费存在延迟等 30 秒再查大概率能追上如果 Redis 扣减数与 ok 数对不上说明计费中间件有请求没走到扣费逻辑这时候要去查是不是限流中间件提前拦截了。6.2 一键核对脚本对Redis余额和MySQL流水压测通过后再做一次静态核对把 Redis 里的剩余额度、MySQL 流水总量、初始配额三者放在同一个脚本里对比对不上就说明“先扣后记”链路里有缺口。# reconcile.py import redis import pymysql r redis.Redis(host127.0.0.1, port6379, db1, decode_responsesTrue) conn pymysql.connect(host127.0.0.1, userroot, password***, databaseapi_platform) init_quota 10000 quota_left int(r.hget(plan:app_test_001, quota)) with conn.cursor() as cur: cur.execute( SELECT SUM(cost), COUNT(*) FROM billing_log WHERE app_id%s AND status1, (app_test_001,), ) row cur.fetchone() consumed row[0] or 0 print(f初始配额: {init_quota}) print(fRedis剩余: {quota_left}) print(f流水扣费: {consumed})这可能是整个系统上线前最有价值的一段脚本它验证的恰恰是“计费”和“调用量”这两件事是否真的等同。如果 Redis 剩余额度与 init_quota - consumed 不相等说明存在漏记或多扣的流水绝不带着这个差异上线。我自己的习惯是核对脚本一直保留每周跑一次计费系统最怕的不是没有日志而是日志和余额各说各话——黑匣子状态比明确报错更让人紧张。这套系统上线后我也多次调整过量价策略后来意识到二开版源码和维护者自己的工程习惯同样重要真正的成本不在部署那两天而在后续每次新套餐、新客户、上游接口变更时计费链路能不能跟着改对。希望这篇能帮你在动手前就把账算明白少踩几个我已经替你踩过的坑。本文还有配套的精品资源点击获取 二开版API管理系统源码：从计费链路到网关转发的实战解析
简介面向 API 服务运营与开发者的全新二开版接口管理系统源码基于 NginxPHP7.4MySQL5.7 环境测试运行访问域名即可完成安装项目重点修复原版 API 鉴权漏洞与源代码暴露风险并升级为现代化响应式前后端设计适配 PC 与移动多终端管理场景帮助使用者搭建安全、易用的接口管理后台。资源包共 446 个文件以 zip 压缩包形式提供主要文件包括 84 个 PHP 后台逻辑与接口文件、217 个 JavaScript 交互脚本、48 个 CSS 样式表另附 SQL 数据库、字体及图标资源整体约 20.17MB目前已有 145 人学习。二开内容覆盖完整新增 API 分类功能实现结构化接口管理API 详情页集成在线调试工具已登录用户自动获取密钥后台提供 QPS 限速开关为后续流量控制预留扩展同时修复了邮件标题显示问题并美化支付成功页与通知邮件。整套源码目录规范、前端组件选择成熟既适合开发者深入研读接口平台设计思路也可快速二次开发构建自有 API 计费管理系统。1. 全新二开版API管理系统源码为什么卡住大家的总是计费而不是转发做过API平台的人都有个体会把一个接口开放出去很简单但让它按调用量收钱难度会直接跳到另一个量级。标题里这三个词——二开版、API计费、全开源——其实对应三件事一套能对外售卖API能力的业务系统、一套能算清账的计量链路、以及一套可以随意改动的完整代码。这类源码的真正价值从不在网关转发那一层而在计费闭环谁调了多少次、余额扣到哪、套餐怎么算、密钥怎么发、对账怎么对。这篇文章要聊清楚的就是这件事读者是手上有API服务想对外变现的团队或者需要把API能力分发出去、向内部部门摊派调用成本的技术负责人。2. 二开版API管理系统源码的架构与部署先用三条线拆清源码再启动2.1 拆源码的三条业务线管理后台、API网关、计费Worker二开版API管理系统源码拿到手第一件事不是急着配环境而是把代码按业务线拆开。常见做法是分成管理后台、API网关、计费服务三块少数实现会把计费直接塞进网关进程里但那种结构后面对账会很难受。管理后台负责的是“卖”的动作创建商户、签发密钥、配置套餐、查看调用报表。API网关负责“转发”的动作校验签名、检查限流、把请求转发到真正的上游服务再把响应原样返回。计费服务负责“算账”的动作从Redis里实时扣减余额、把流水异步落进MySQL、触发余额不足回调。三条线的关系是请求先进网关网关在转发前调用计费检查转发完成后计费Worker异步写流水管理后台只读库和缓存不参与在线链路。模块职责数据落点管理后台商户、套餐、密钥、报表MySQL 业务库API网关签名校验、限流、转发Redis 实时状态计费 Worker扣费、流水、回调、对账Redis 扣减MySQL 落账拆清楚这条线之后碰到问题就知道该看哪个进程的日志。比如调用方报“余额扣了但请求失败”问题大概率不在网关而在计费Worker的补偿逻辑或者转发之后没有把失败状态传回计费模块。2.2 本地跑通最小部署从环境依赖到三个进程同时拉起来假设这套二开版源码是PythonFastAPI MySQL Redis 的技术栈这是目前这类系统里很常见的一种选型下面以它为例讲部署。环境要求是 Python 3.10、MySQL 8.0、Redis 6.x内存至少预留 1GB 给Redis存计费状态。# 1. 创建虚拟环境并安装依赖requirements.txt 由源码包自带 python3 -m venv .venv source .venv/bin/activate pip install -r requirements.txt # 2. 复制环境配置模板改成你自己的连接信息 cp .env.example .env # 编辑 .envDB_HOST、DB_PORT、DB_USER、DB_PASSWORD、REDIS_HOST、REDIS_PORT # 这三个参数必改DB_PASSWORD、ADMIN_PASSWORD、SECRET_KEY # 3. 初始化数据库表结构执行后 MySQL 里会生成 merchant/app_key/plan/billing_log 等表 python scripts/init_db.py # 4. 启动三个进程管理后台监听 8000API网关监听 9000计费Worker常驻后台 uvicorn admin.app:app --host 0.0.0.0 --port 8000 uvicorn gateway.app:app --host 0.0.0.0 --port 9000 python worker/billing_worker.py 第一步和第二步决定了后面所有账能不能算对。DB_PASSWORD 和 SECRET_KEY 必须改SECRET_KEY 会参与回调通知的签名生成如果沿用源码包里的默认值下游调用方拿到源码就能伪造回调。Admin 后台监听 8000 端口网关监听 9000 端口两者物理隔离是为了后续给网关单独做水平扩容计费Worker独立进程则是为了避免在线请求阻塞时扣费延迟。启动完三个进程后先确认 Redis 里能读到套餐缓存、MySQL 里能看到四张业务表再进下一步初始化数据。这里最容易翻车的地方是 MySQL 字符集没设成 utf8mb4导致下游传 emoji 进去时直接 500我一般会在 init_db.py 里强制指定DEFAULT CHARSETutf8mb4。2.3 初始化最小业务数据一个商户、一把密钥、一个按次套餐部署跑通之后离第一笔计费还差最关键的一步数据初始化。二开版源码一般会自带初始化脚本但如果脚本只建表不造业务数据你就得手动插入。下面是一套最精简的初始化SQL覆盖商户、密钥、套餐、流水四张核心表这也是计费链路能跑通的最少数据。-- 商户假设这是你的第一个客户 INSERT INTO merchant (name, status) VALUES (测试商户, 1); -- 套餐按次计费每次 0.01 元余额 10000 次每分钟限 120 次 INSERT INTO plan (name, price, quota, rate_limit, window_sec) VALUES (按次体验包, 0.0100, 10000, 120, 60); -- 密钥app_id 发给下游secret 只在服务端保存 INSERT INTO app_key (app_id, secret, merchant_id, plan_id, status) VALUES (app_test_001, sk_live_9f8e7d6c5b4a, 1, 1, 1); -- 流水表本身不需要初始化但这个自增主键和幂等唯一键很关键 -- request_id 是幂等键同一条请求扣费两次会直接报错 CREATE TABLE IF NOT EXISTS billing_log ( id BIGINT PRIMARY KEY AUTO_INCREMENT, request_id VARCHAR(32) NOT NULL UNIQUE, app_id VARCHAR(32) NOT NULL, cost INT NOT NULL DEFAULT 1, status TINYINT DEFAULT 0, created_at DATETIME(3) );商户表只存身份信息计费不看它套餐表里的 price 是每次调用的单价quota 是剩余可用次数rate_limit 参与限流判定app_key 表把下游调用方和套餐绑定在一起一个商户可以开多个应用丢给不同项目用。四张表里最重要的是 billing_logcost 字段我建议用整数存“分”或“厘”不要用浮点数哪怕价格表里写的是 0.0100 元落账时也要转成 1 分不然月底对账时浮点误差会让你怀疑人生。初始化完成后拿 app_id 和一个任意签名去调网关的健康检查接口能看到 200 就说明这一整套最小闭环已经跑通可以进入计费模块的实现细节了。3. 计费模块的实现逻辑把API调用量变成余额扣减需要算清四笔账3.1 先定计费模型按次、包周期还是阶梯决定了表的字段怎么设计计费模型是整个系统的地基选错后面全是坑。最常见的三种模型是按次计费、包周期套餐、阶梯计费。按次计费最简单每次调用扣固定金额适合刚起步的小商户包周期是按月/按年买固定额度超额后拒绝服务或转按次阶梯计费则是在一个周期内累计调用量跨过某个阈值后单价自动下降适合中大型API服务。模型优点缺点适合场景按次实现简单对账直观大客户嫌贵小商户、测试期包周期收入可预期超额对账麻烦大部分SaaS API阶梯大客户愿意用需要周期累计调用量大的老客户二开版源码里这三个模型一般会同时存在。实现上按次计费只需要一个 quota 字段递减包周期要记录周期开始时间和重置时间阶梯计费麻烦得多需要一张独立的累计表按自然日或自然月汇总调用量。我的建议是先只开放按次和包周期两种阶梯计费等客户真到了那个量级再二开否则前期光是统计口径就能耗掉你两周。3.2 用Redis做实时扣减用MySQL做最终流水双写才能既快又稳计费模块的核心矛盾是“快”和“稳”。一次API调用会在几十毫秒内结束扣费必须同步完成用MySQL在在线链路里做行锁扣减并发一高就死锁但如果只扣Redis进程一挂账就丢了。所以常见做法是两层配合Redis负责实时扣减MySQL流水负责最终对账。# billing_middleware.py import uuid import time import redis from fastapi import Request, HTTPException # db1 单独给计费用不让业务缓存干扰计费数据 r redis.Redis(host127.0.0.1, port6379, db1, decode_responsesTrue) async def check_billing(request: Request): # 下游调用方在请求头里带上平台签发的 app_id app_id request.headers.get(X-App-Id) if not app_id: raise HTTPException(status_code401, detailmissing app_id) plan_key fplan:{app_id} # 套餐信息在 Redis 里用 hash 缓存quota 剩余次数、price_val 单价、rate_limit 限流阈值 # 示例hset plan:app_test_001 quota 10000 price_val 1 rate_limit 120 quota int(r.hget(plan_key, quota) or 0) if quota 0: # 402 是 HTTP 里专门表示“余额不足”的语义比 500 好排查 raise HTTPException(status_code402, detailinsufficient quota) # 生成请求ID幂等键流水表里靠它去重 request_id uuid.uuid4().hex # 先扣后记INCRBY 是 Redis 原子操作不用 Lua 也能防超卖 r.hincrby(plan_key, quota, -1) # 把扣费动作追加到本应用的临时流水队列Worker 异步刷 MySQL r.lpush(fbilling:queue:{app_id}, f{request_id}:{int(time.time())}) # 上游转发成功后网关把 request_id 一并带到响应头方便调用方排查 return {request_id: request_id, app_id: app_id}这里有两个关键参数必须解释清楚。db1是给Redis单独划了一个逻辑库计费相关的 key 全部放在 db 1管理后台的缓存和会话放 db 0这样清理缓存时不会误删计费数据。hincrby加-1是原子操作两个并发请求同时进来时不会都读到同一个 quota 值这是避免超卖的基本功。扣费时机的选择值得多说一句先扣后记即先扣余额再转发。如果先转发再扣费上游响应已经返回了但扣费失败调用方等于免费调了一次先扣费后转发虽然可能出现“扣了钱但上游超时”的情况但这个问题可以在避坑章节里用补偿机制解决比免费调用可控得多。计费中间件返回的 request_id 还会放在响应头里调用方工单排查时凭这个ID就能在两分钟内定位到流水。3.3 余额不足与回调通知别让调用方糊里糊涂断了服实时扣减之外二开版API管理系统还必须有主动通知能力调用方余额见底时平台不只返回402还要主动推一个回调给调用方让他去充值。回调通知最怕两件事没推到、重复推。所以回调接口必须支持重试并且用 request_id 做幂等。# callback.py import hashlib import hmac import time import requests def send_balance_callback(app_id: str, merchant_callback_url: str, secret: str): # 回调体只有四个字段app_id、额度剩多少、时间戳、签名 payload { app_id: app_id, quota_left: 0, timestamp: int(time.time()), } # 签名用 HMAC-SHA256secret 是 app_key 表里那个签名防伪造 raw f{app_id}{payload[timestamp]}{payload[quota_left]} payload[sign] hmac.new( secret.encode(), raw.encode(), hashlib.sha256 ).hexdigest() # 重试 3 次间隔 5 秒、30 秒、300 秒不阻塞在线请求 for interval in (5, 30, 300): try: resp requests.post(merchant_callback_url, jsonpayload, timeout10) # 调用方回调接口要返回 HTTP 200 才算成功否则进入下一次重试 if resp.status_code 200: return True except requests.RequestException: pass time.sleep(interval) return False回调带了时间戳所以调用方收到回调后必须校验时间偏差超过五分钟直接丢弃。签名算法与网关签名保持一致调用方可以用同一套验签逻辑处理。回调场景里最容易踩的坑是回调接口地址本身配置错误二开版里这个地址通常在商户表里但有些源码包会把它写死在配置文件里导致所有商户共用一个回调地址测试时能收到、上线后全部推给了同一家排查时要先去确认回调地址是商户级别还是平台级别的配置。4. 网关鉴权与路由转发签名校验、限流和计费如何联动4.1 基于HMAC-SHA256的签名校验app_id公开secret只在服务端API管理系统的第一道门是鉴权。平台给每个调用方签发一个 app_id 加 secretapp_id 放在请求头里明文传输secret 不能出现在请求里否则抓包的人直接就能伪造调用。所以要做签名调用方把请求参数、时间戳、随机数一起算一个 HMAC-SHA256 签名服务端用同样的 secret 重算比对。# auth.py import hashlib import hmac import time from fastapi import Header, HTTPException def verify_sign( app_id: str Header(..., aliasX-App-Id), timestamp: str Header(..., aliasX-Timestamp), nonce: str Header(..., aliasX-Nonce), sign: str Header(..., aliasX-Sign), secret: str , ): # 时间窗 300 秒超过就拒绝。防止旧的合法请求被无限重放 if abs(int(time.time()) - int(timestamp)) 300: raise HTTPException(status_code401, detailtimestamp expired) # 拼接顺序必须固定两端用的规则要一模一样 raw f{app_id}{timestamp}{nonce} expect hmac.new(secret.encode(), raw.encode(), hashlib.sha256).hexdigest() # compare_digest 防时序攻击不能用 直接比 if not hmac.compare_digest(expect, sign): raise HTTPException(status_code401, detailsign mismatch) # 生产环境里nonce 要放进 Redis 做去重同一个 nonce 只能用一次 # 这里省略 Redis 部分避免与计费中间件抢连接影响性能 return app_id签名校验里最容易漏的是 nonce 去重。时间戳窗口 300 秒意味着一个合法签名在 5 分钟内都有效如果不去重攻击者抓包后能在这 5 分钟内无限重放。常见的做法是把 nonce 写进 Redis 的 set 结构窗口时间设置为与签名时间窗一致。secret 不落请求不代表可以明文存数据库二开版里 app_key 表的 secret 字段至少要做哈希处理我见过不少源码把 secret 以明文存一旦数据库被拖走全部调用方密钥直接暴露。4.2 限流不只看QPS把套餐剩余次数与滑动窗口绑在一起网关的第二道关卡是限流。传统限流只关心每秒请求数但在API管理系统里限流必须和套餐联动同样的 120 次每分钟按次计费和包周期套餐的触发条件不一样。二开版常见做法是把套餐表里的 rate_limit 和 window_sec 读出来作为限流阈值传给网关。# rate_limit.py import time import redis r redis.Redis(host127.0.0.1, port6379, db1, decode_responsesTrue) def sliding_window_check(app_id: str, limit: int 120, window_sec: int 60): # 用有序集合存每次调用的时间戳天然实现滑动窗口 key frate:{app_id} now time.time() # 先清掉窗口外的老记录避免 zset 无限膨胀 r.zremrangebyscore(key, 0, now - window_sec) count r.zcard(key) if count limit: raise PermissionError(rate limited) # 当前请求的时间戳入集合zadd 的 score 和 member 都用时间戳 r.zadd(key, {str(now): now}) # 窗口结束 60 秒后整把 key 过期防止死数据占内存 r.expire(key, window_sec 60) return count 1这里不是用 Redis 的 INCR 加过期时间做固定窗口而是用 zset 做滑动窗口差别在于固定窗口在窗口切换瞬间会出现 2 倍突发流量。zset 方案每次请求都要执行一次 zremrangebyscore窗口越大成本越高所以 window_sec 设定为 60 秒、limit 在 1000 以下时性能没有问题。如果套餐有更高限额我一般会把限流窗口改成 1 秒粒度加 60 秒聚合计数两种方案配合而不是只用滑动窗口。限流与计费的联动体现在顺序上先做限流检查再做计费扣减。限流失败不扣费计费失败不转发。两个中间件用同一个 app_id 做 key但 key 前缀不同一个是rate:一个是plan:避免互相覆盖。4.3 转发时替换上游密钥对接大模型API服务的统一出口网关的最后一步是转发到上游真实API服务。这一步是二开版系统和普通反向代理的关键区别普通代理只改 URLAPI计费系统还要在转发时替换认证信息。下游调用方用的是平台签发的 app_id而上游服务只认平台自己采购的 key两个 key 不能混。# forwarder.py import httpx async def forward_to_upstream(api_config: dict, headers: dict, body: bytes): # api_config 里是上游地址、请求方法、超时、以及平台集中管理的上游key # 常见做法是在管理后台维护一张上游API配置表对接DeepSeek、智谱这类大模型API服务 async with httpx.AsyncClient(timeoutapi_config[timeout]) as client: resp await client.request( methodapi_config[method], urlapi_config[url], headers{**headers, Authorization: fBearer {api_config[upstream_key]}}, contentbody, ) # 网关原样返回上游响应不解析body这样对下游调用方来说是透明的 return resp.status_code, resp.content转发层的超时参数必须比网关本身超时短不然网关等上游、下游等网关会出现两层超时叠加导致的 504。比如网关整体超时设 15 秒上游转发超时就要设 13 秒留 2 秒给签名校验和计费扣减。二开版里上游 key 应该集中放在配置表里由平台统一采购和续费不能把上游 key 下发给调用方否则调用方直接绕过平台调上游这个系统就变成纯转发工具而不是计费平台了。5. 二开版源码避坑部署、鉴权、计费最容易翻车的五个地方5.1 鉴权与路由方向上常踩的坑上游key配错、超时重试、时间窗过期现象一下游调用方明明拿了合法的 app_id转发时报llm-deepseek: no api key for provider route deepseek-official。原因二开版把上游 key 放在配置表后运维把平台自己的上游 key 误配到了下游应用的 app_key 上网关转发时没有做替换上游收到的是不存在的 key。解决转发前确认 api_config 表里 upstream_key 字段有值并且生产环境不要用 init 脚本里的测试 key。我这个坑实际踩过两次第一次是以为上游服务的问题排查了半小时才发现是 key 配错了桶。现象二一次请求因为网络抖动超时调用方重试后账单里出现两条扣费记录。原因计费中间件在转发前先扣费超时后客户端重试网关又走了一遍扣费逻辑同一个业务请求被扣了两次。这是“先扣后记”方案最典型的副作用。解决扣费之前先查 billing_log 的 request_id如果客户端重试时带上了原始 request_id直接返回第一次的响应不算钱。所以网关的响应头一定要返回 request_id并且引导调用方在重试时带上这个ID。现象三上游返回 400报this models maximum context length is 1048576 tokens这类参数超长错误。原因签名校验成功后请求转发给上游大模型API但请求体太大或参数超限上游拒绝了。这个不算计费系统自身缺陷但会让调用方误以为网关有问题。解决网关只转发不解析但可以在管理后台配置“上游错误码白名单”把这类 400 错误透传给调用方的同时标记为“不扣费”。是否对 4xx 扣费要提前定好规矩我一般遵循 5xx 不扣、4xx 扣一半、2xx必扣具体在二开版里用 status 字段区分。5.2 计费与对账方向上常踩的坑Redis丢了额度、流水丢尾部、统计口径打架现象四调用量远没到套餐上限却突然全部返回 402 余额不足。原因Redis 里的 plan: 前缀 key 因为内存淘汰策略被回收了计费中间件读到的 quota 是 0直接拒绝服务。二开版源码很多默认使用 Redisallkeys-lru淘汰策略计费 key 和普通缓存一起竞争内存一旦内存压力上来最先被淘汰的就是这些长尾 key。解决给计费 key 单独提一个 Redis 实例或者至少把淘汰策略改成volatile-lru并且写一个启动预热脚本把所有 enabled 状态套餐的 quota 从 MySQL 刷进 Redis。现象五计费 Worker 重启后Redis 流水队列里最后十秒的数据丢了账单少记了调用量。原因Worker 用 lpop 从billing:queue:{app_id}取数据进程在 lpop 之后、写 MySQL 之前崩溃记录就丢了。解决把 Redis list 换成 Redis Streams用 XREADGROUP XACK 做消费确认Worker 重启后先 XAUTOCLAIM 捞回未确认消息再继续处理。这个改造大概要花半天但能避免月底对账时账单少几十万的尴尬。计费统计口径还有一个长期存在的坑限流按滑动窗口算账单按自然日聚合两边数字永远对不上。原因60 秒滑动窗口在 23:59:30 到 00:00:30 这个区间内跨了两个自然日限流统计被切成了两段而账单按自然日切分后就少计或重计了一部分。解决在报表里同时标注“按自然日”和“按调用窗口”两种口径别强行让它们相等如果客户要求严格对账把计费周期改成从每日零点开始的固定 60 秒窗口牺牲一点滑动窗口的平滑性换取对账一致性。这个选择没有绝对正确关键是要在文档里写清楚。6. 上线前验证用并发脚本核对“扣费次数成功调用次数”再谈卖API6.1 200并发压测脚本计费上限要压出来上线前一天至少要做一次并发压测不是压 QPS而是压计费一致性。我常用的办法是起 200 个线程同时调同一个接口然后核对三个数字成功响应数、Redis 里扣减的额度、MySQL 流水条数。# perf_check.py import threading import requests URL http://127.0.0.1:9000/api/v1/demo HEADERS {X-App-Id: app_test_001, X-Sign: test_sign} ok fail 0 lock threading.Lock() def call(): global ok, fail try: r requests.get(URL, headersHEADERS, timeout5) with lock: if r.status_code 200: ok 1 else: fail 1 except Exception: with lock: fail 1 threads [threading.Thread(targetcall) for _ in range(200)] for t in threads: t.start() for t in threads: t.join() print(fok{ok} fail{fail})跑完这个脚本如果 ok fail 200 但 MySQL 流水条数小于 ok 数说明计费 Worker 消费存在延迟等 30 秒再查大概率能追上如果 Redis 扣减数与 ok 数对不上说明计费中间件有请求没走到扣费逻辑这时候要去查是不是限流中间件提前拦截了。6.2 一键核对脚本对Redis余额和MySQL流水压测通过后再做一次静态核对把 Redis 里的剩余额度、MySQL 流水总量、初始配额三者放在同一个脚本里对比对不上就说明“先扣后记”链路里有缺口。# reconcile.py import redis import pymysql r redis.Redis(host127.0.0.1, port6379, db1, decode_responsesTrue) conn pymysql.connect(host127.0.0.1, userroot, password***, databaseapi_platform) init_quota 10000 quota_left int(r.hget(plan:app_test_001, quota)) with conn.cursor() as cur: cur.execute( SELECT SUM(cost), COUNT(*) FROM billing_log WHERE app_id%s AND status1, (app_test_001,), ) row cur.fetchone() consumed row[0] or 0 print(f初始配额: {init_quota}) print(fRedis剩余: {quota_left}) print(f流水扣费: {consumed})这可能是整个系统上线前最有价值的一段脚本它验证的恰恰是“计费”和“调用量”这两件事是否真的等同。如果 Redis 剩余额度与 init_quota - consumed 不相等说明存在漏记或多扣的流水绝不带着这个差异上线。我自己的习惯是核对脚本一直保留每周跑一次计费系统最怕的不是没有日志而是日志和余额各说各话——黑匣子状态比明确报错更让人紧张。这套系统上线后我也多次调整过量价策略后来意识到二开版源码和维护者自己的工程习惯同样重要真正的成本不在部署那两天而在后续每次新套餐、新客户、上游接口变更时计费链路能不能跟着改对。希望这篇能帮你在动手前就把账算明白少踩几个我已经替你踩过的坑。本文还有配套的精品资源点击获取