分享一个Webhook分发/转发平台

基于 Hyperf 3.x + Swow 协程引擎构建的 Webhook 消息分发与请求转发平台。提供可视化管理后台，支持主题（Topic）、订阅者（Subscriber）管理、Webhook 回调接收、常驻消息分发器，以及独立的请求转发（Endpoint Forwarding）功能。

功能总览

1. 管理后台（Web UI）

管理台 http://<host>:9501/ — 主题、订阅者、消息三大板块的可视化管理，深色模式、响应式布局，无外部 CDN 依赖（完全离线可用）。
请求转发页 http://<host>:9501/forward — 展示对外域名、固定主题 endpoint、订阅者列表、转发示例与最近转发记录。
内置离线演示模式，无后端时也可预览界面效果。

2. 订阅主题（Topics）

创建 / 编辑 / 删除主题，字段：key（唯一）、title（标题）、response_data（回调响应数据）。
主题 key 创建后不可修改（外部系统的回调地址依赖它），只允许修改标题与响应数据。
key 校验规则：仅允许字母、数字、下划线、中划线、点，长度 ≤ 64。
response_data 将作为 /webhook 回调入口的响应体原样返回给外部系统；若以 { 或 [ 开头会校验 JSON 格式合法性。

3. 订阅者（Subscribers）

创建 / 编辑 / 删除订阅者，字段：name（名称）、url（完整回调地址，如 http://192.168.2.12/webhook/payful/notify）、status。
多对多订阅：订阅者可订阅多个主题，只有订阅了某主题的订阅者才会收到该主题的消息。
分发器投递成功 / 失败会自动回写订阅者健康状态（1-正常 / 2-无响应）与最后成功时间。

4. Webhook 回调接收入口

GET|POST|PUT /webhook/{topicKey}/{topicId}

外部系统只需配置这一个地址，消息落库后由常驻分发器异步投递给所有订阅该主题的订阅者。
响应体为主题配置的 response_data（外部契约由主题自定义），分发摘要通过响应头返回：
- X-Webhook-Topic：命中的主题 key
- X-Webhook-Key-Matched：路径中的 key 与 topicId 对应主题是否一致
- X-Webhook-Subscriber-Count / X-Webhook-Message-Count / X-Webhook-Message-Ids
请求体处理：JSON 原样保存；表单 / 查询参数归一化为 JSON；纯文本、XML 等原样保存。

5. 常驻消息分发器（Dispatcher）

基于 Hyperf 自定义进程的常驻后台任务，随服务启动自动运行。
messages 表即投递队列，处理逻辑：
1. 原子领取一批可投递消息（待发送 / 崩溃遗留的发送中 / 到期重试的失败消息），条件更新天然支持多实例部署不重复投递；
2. 协程并发 POST 到订阅者回调地址（并发上限 10，单批 50 条，总超时 10s）；
3. 回写消息状态、HTTP 响应状态码、响应内容、尝试次数与订阅者健康状态。
失败自动重试：最多 5 次，线性退避（间隔 = 30s × 已尝试次数），超过后保持「发送失败」终态。
例外：请求转发（/endpoint）消息的重试不由分发器直接发 HTTP，而是交回 ForwardService 处理（见第 6 节），因为它要发往「订阅者地址 + 去掉 /endpoint 的路径」，与普通分发目标不同。

消息状态：0-待发送、1-发送中、2-发送成功、3-发送失败。

6. 请求转发（独立功能，固定主题 `endpoint`）

GET|POST|PUT|DELETE|PATCH /endpoint/{path...}

独立于原有 webhook 分发链路：只有订阅了固定主题 key endpoint 的订阅者会被转发。
转发规则：/endpoint 之后的部分去掉前缀后，原样拼接到订阅者域名。
- 例：POST https://your-domain/endpoint/webhook/payful/notify → 转发给所有订阅 endpoint 主题的订阅者：POST {订阅者url}/webhook/payful/notify
订阅者只需设置名称、域名/地址并订阅 endpoint 主题即可。
每次转发都会在消息记录中保存：
- request_method：原始请求方式（POST / GET / PUT …）
- request_path：完整原始地址（对外域名 + 路径，如 https://your-domain/endpoint/webhook/payful/notify）
- target_url：完整转发目标地址
- response_status / response_content：订阅者返回的状态码与响应体
转发结果同时通过 X-Forward-* 响应头暴露（主题、方法、原路径、目标路径、成功数等）。
转发失败自动重试：失败的消息以「发送失败」落库后，由常驻分发器按退避策略重试（间隔 30s × 次数，最多 5 次）。重试仍由 ForwardService 执行：按记录中的原始请求方式（request_method）重新请求转发地址（target_url），而不是退化成直接请求订阅者域名 + 空路径，因此 GET / PUT / DELETE / PATCH 的重试方法与首次转发完全一致。
对外域名由环境变量 WEBHOOK_DOMAIN 配置；未配置时自动退回当前请求的 scheme + host。

已知限制：转发只拼接路径部分，查询串（?a=1）既不转发也不记录。

7. 消息记录（只读）

支持按主题、订阅者、状态、关键词（覆盖消息内容、响应内容、原路径、转发目标）筛选。
每条消息可查看：请求方式、原路径（含域名）→ 转发路径、订阅者、状态、HTTP 响应状态码、响应内容、尝试次数等。

8. 消息记录自动清理（过期数据）

独立的常驻进程 MessageCleanupProcess 负责，和分发器互不影响，不占用请求处理资源：
- 服务启动时先清理一次（延迟 5 秒，等数据库就绪）；
- 之后每天定点执行一次（默认 03:00，可配置）。
清理规则：删除 created_at 早于「当前时间 − 保留天数」的消息；保留天数默认 30 天，由 .env 的 MESSAGE_RETENTION_DAYS 配置。
默认保留未完成消息（0-待发送 / 1-发送中），只删「发送成功 / 发送失败」的终态历史记录，避免误删尚未投递出去的回调。
删除按批进行（每批 1000 条、批间让出协程，单轮上限 20 万条），不会长时间锁表或拖慢服务。
也可以手动执行或交给系统定时任务：php bin/hyperf.php message:cleanup（见「使用指南 · 场景 C」）。

系统要求

组件	要求
PHP	>= 8.1，且必须安装 Swow 扩展（本项目实测使用 PHP 8.2 NTS + Swow；PHP 8.4 目前无可用 Swow 扩展，请勿使用）
Composer	>= 2.x
MySQL	>= 5.7（utf8mb4）
Redis	可选（框架组件引用，核心功能未强依赖）
操作系统	Linux / Windows / macOS 均可

PHP 必需扩展：pdo、pdo_mysql、json、openssl（HTTPS）、Swow。

快速开始（部署）

1. 安装依赖

composer install

2. 配置环境

复制并编辑 .env（若不存在可从 .env.example 复制）：

APP_NAME=Webhook分发平台

# MySQL
DB_DRIVER=mysql
DB_HOST=127.0.0.1
DB_PORT=3306
DB_DATABASE=webhook
DB_USERNAME=root
DB_PASSWORD=root
DB_CHARSET=utf8mb4

# Redis（可选）
REDIS_HOST=127.0.0.1
REDIS_AUTH=(null)
REDIS_PORT=6379
REDIS_DB=0

# Webhook 对外接入域名（请求转发功能用于记录完整原路径）
WEBHOOK_DOMAIN=https://your-domain.com

# 消息记录自动清理（过期消息在服务启动时 + 每天定点删除）
# 是否启用自动清理（false 则不启动清理进程）
MESSAGE_CLEANUP_ENABLE=true
# 保留天数：删除 created_at 早于「now - 该天数」的消息；0 或负数表示不清理
MESSAGE_RETENTION_DAYS=30
# 每天执行时刻（HH:MM）
MESSAGE_CLEANUP_AT=03:00
# 是否保留未完成（待发送 / 发送中）的过期消息
MESSAGE_CLEANUP_KEEP_UNFINISHED=true

WEBHOOK_DOMAIN 建议配置为外部系统实际访问本服务的域名（如内网穿透 / 公网域名）。配置后，转发消息记录中的「原路径」会以该域名开头，完整可追溯。

消息清理的 4 个参数都有默认值，不配置时按「保留 30 天、每天 03:00、保留未完成消息」运行。

3. 初始化数据库

创建 webhook 数据库后导入表结构：

mysql -u root -p -e "CREATE DATABASE IF NOT EXISTS webhook DEFAULT CHARSET utf8mb4"
mysql -u root -p webhook < sql/init.sql

包含 4 张表：

表	说明
`topics`	订阅主题（key 唯一）
`subcribers`	订阅者
`subscriptions`	主题-订阅者多对多关系
`messages`	消息记录 / 投递队列

4. 启动服务

php bin/hyperf.php start

HTTP 服务监听 0.0.0.0:9501（可在 config/autoload/server.php 修改）。
两个常驻自定义进程随服务一起启动，无需额外操作：
- 消息分发器 app/Process/WebhookDispatcherProcess.php；
- 过期消息清理 app/Process/MessageCleanupProcess.php（启动后先清一次，之后每天定点再清；设 MESSAGE_CLEANUP_ENABLE=false 则不启动该进程）。

生产环境建议使用 Supervisor 守护：

[program:webhook]
command=php /path/to/bin/hyperf.php start
autostart=true
autorestart=true
user=www
stdout_logfile=/var/log/webhook.log

使用指南

场景 A：Webhook 消息分发

在管理台「订阅主题」创建主题，例如 key = payful，并配置回调响应数据（外部系统收到的响应体）。
在「订阅者」创建订阅者，填写名称与完整回调地址（如 http://192.168.2.12/webhook/payful/notify）。
为订阅者勾选订阅主题 payful。
外部系统向 POST http://<host>:9501/webhook/payful/{topicId} 发送回调。
分发器自动将消息 POST 给所有订阅者，消息列表可查看投递状态、响应状态码与响应内容。

场景 B：请求转发

在 .env 配置 WEBHOOK_DOMAIN（对外可访问的域名）。
创建订阅者，填写名称与目标域名/地址（如 http://192.168.2.12）。
让该订阅者订阅固定主题 endpoint（主题不存在时先创建 key 为 endpoint 的主题）。
外部请求 https://your-domain/endpoint/<任意路径>，会被转发为 {订阅者地址}/<任意路径>，请求方式与请求体原样保留。
在 /forward 页面或消息列表查看每次转发的原路径（含域名）、转发路径、响应状态与响应内容。

场景 C：清理过期消息记录

方式一：随服务自动执行（默认开启）

在 .env 中配置保留天数后重启服务即可，无需其他操作：

MESSAGE_CLEANUP_ENABLE=true
# 保留 30 天
MESSAGE_RETENTION_DAYS=30
# 每天 03:00 执行
MESSAGE_CLEANUP_AT=03:00
MESSAGE_CLEANUP_KEEP_UNFINISHED=true

服务日志中能看到执行结果：

[cleanup] 消息清理进程已启动：保留 30 天，每天 03:00 执行，启动时先清理一次（待发送 / 发送中的过期消息不删）
[cleanup] （启动）清理完成：删除 12 条 30 天前的历史消息（早于 2026-08-24 03:00:00），保留未完成 0 条，分 1 批，耗时 0.12s

方式二：手动执行 / 交给系统定时任务

# 先演习：只统计会被删除的条数，不实际删除
php bin/hyperf.php message:cleanup --dry-run

# 按 .env 的保留天数执行
php bin/hyperf.php message:cleanup

# 临时覆盖保留天数（例如只清 7 天前的记录）
php bin/hyperf.php message:cleanup --days=7

若改用系统定时任务执行，可把 MESSAGE_CLEANUP_ENABLE 设为 false（避免与进程重复），示例：

# Linux crontab：每天 03:00 执行
0 3 * * * cd /path/to/project && php bin/hyperf.php message:cleanup >> /var/log/webhook-cleanup.log 2>&1

管理 API 一览

统一 JSON 信封响应：{ code, message, data }。

方法	路径	说明
GET	`/api/overview`	概览统计
GET	`/api/options`	下拉选项（主题精简列表等）
GET / POST	`/api/topics`	主题列表（支持 `keyword`）/ 新增
GET / PUT / DELETE	`/api/topics/{id}`	主题详情 / 更新 / 删除（`force=1` 连同消息删除）
GET / POST	`/api/subscribers`	订阅者列表（`keyword`、`status`）/ 新增
GET / PUT / DELETE	`/api/subscribers/{id}`	订阅者详情 / 更新 / 删除（`force=1` 连同消息删除）
GET	`/api/messages`	消息列表（`topic_key`、`subcriber_id`、`status`、`keyword`）
GET	`/api/messages/{id}`	消息详情

快速验证

# 创建主题
curl -X POST http://127.0.0.1:9501/api/topics \
  -H "Content-Type: application/json" \
  -d '{"key":"payful","title":"支付回调","response_data":"{\"code\":0,\"msg\":\"success\"}"}'

# 模拟外部回调
curl -X POST http://127.0.0.1:9501/webhook/payful/{topicId} \
  -H "Content-Type: application/json" \
  -d '{"order_no":"20260920001","amount":100}'

# 验证请求转发
curl -X POST https://your-domain/endpoint/webhook/payful/notify \
  -H "Content-Type: application/json" \
  -d '{"event":"paid"}'

项目结构（核心）

app/
├── Controller/
│   ├── WebhookController.php     # /webhook 回调接收入口
│   ├── EndpointController.php    # /endpoint 请求转发入口
│   ├── TopicController.php       # 主题管理 API
│   ├── SubscriberController.php  # 订阅者管理 API
│   ├── MessageController.php     # 消息查询 API（只读）
│   ├── MetaController.php        # 概览与元数据
│   └── AdminController.php       # 管理后台页面
├── Service/
│   ├── WebhookService.php        # 回调落库与消息生成
│   ├── DispatcherService.php     # 常驻分发器（领取/并发投递/重试）
│   ├── ForwardService.php        # 请求转发（固定主题 endpoint）
│   ├── TopicService.php          # 主题业务
│   ├── SubscriberService.php     # 订阅者业务
│   ├── MessageService.php        # 消息查询/格式化/响应清洗
│   └── MessageCleanupService.php # 过期消息清理（保留天数 / 分批删除）
├── Process/
│   ├── WebhookDispatcherProcess.php  # 自定义进程（分发器宿主）
│   └── MessageCleanupProcess.php     # 自定义进程（启动时 + 每天定点清理过期消息）
├── Command/
│   └── MessageCleanupCommand.php     # message:cleanup 手动清理命令
└── Support/
    └── SocketHttpClient.php      # 基于 Swow Socket 的 HTTP 客户端
storage/view/
├── admin.html                    # 管理台 SPA
└── forward.html                  # 请求转发页
sql/init.sql                      # 数据库初始化脚本

开发与质量检查

# 静态分析（level 0）
composer analyse

# 代码风格修复
composer cs-fix

# 单元测试
composer test

注意事项

必须使用带 Swow 扩展的 PHP（推荐 PHP 8.1 ~ 8.2），否则服务无法启动。
主题 key 与订阅 endpoint 固定主题的转发规则是外部契约的一部分，创建后请勿随意变更。
删除主题/订阅者时若存在关联消息，默认拒绝，需显式传 force=1 才会一并删除。
分发器失败重试上限 5 次（线性退避 30s × 次数），达到上限后消息保持「发送失败」，可在消息列表排查响应内容。
转发消息的重试由 ForwardService 按 target_url + 原始请求方式发出；升级前产生、没有 target_url 的历史记录，会用「订阅者地址 + 原路径去掉 /endpoint」还原转发地址。
查询串不参与转发与记录；如需透传查询参数，请将其放入请求体或路径中。
messages.created_at 建有普通索引 idx_created_at，供过期清理按时间筛选。新装库由 sql/init.sql 自带；从旧版本升级请手动执行： ALTER TABLE messages ADD KEY idx_created_at (created_at);
过期清理默认只删「发送成功 / 发送失败」的终态记录，待发送 / 发送中 的过期消息会被保留；如确需一并删除，设 MESSAGE_CLEANUP_KEEP_UNFINISHED=false。
清理进程在服务启动后延迟 5 秒执行第一次（等数据库就绪），之后每天定点执行；MESSAGE_RETENTION_DAYS 设为 0 或负数即完全关闭清理（只打印一行「未执行清理」日志）。

用户中心

近期文章

近期评论

归档

分类

通过电子邮件订阅博客

PHP · 2026年9月28日 0

分享一个Webhook分发/转发平台

功能总览

1. 管理后台（Web UI）

2. 订阅主题（Topics）

3. 订阅者（Subscribers）

4. Webhook 回调接收入口

5. 常驻消息分发器（Dispatcher）

6. 请求转发（独立功能，固定主题 `endpoint`）

7. 消息记录（只读）

8. 消息记录自动清理（过期数据）

系统要求

快速开始（部署）

1. 安装依赖

2. 配置环境

3. 初始化数据库

4. 启动服务

使用指南

场景 A：Webhook 消息分发

场景 B：请求转发

场景 C：清理过期消息记录

管理 API 一览

快速验证

项目结构（核心）

开发与质量检查

注意事项

相关

您可能还喜欢...

发表回复取消回复

PHP · 2026年9月28日 0

功能总览

1. 管理后台（Web UI）

2. 订阅主题（Topics）

3. 订阅者（Subscribers）

4. Webhook 回调接收入口

5. 常驻消息分发器（Dispatcher）

6. 请求转发（独立功能，固定主题 endpoint）

7. 消息记录（只读）

8. 消息记录自动清理（过期数据）

系统要求

快速开始（部署）

1. 安装依赖

2. 配置环境

3. 初始化数据库

4. 启动服务

使用指南

场景 A：Webhook 消息分发

场景 B：请求转发

场景 C：清理过期消息记录

管理 API 一览

快速验证

项目结构（核心）

开发与质量检查

注意事项

相关

您可能还喜欢...

Nginx如何实现负载均衡？且保证会话不丢失？

一文看懂 | 今年央视315都曝光了啥？啄木鸟、借贷宝、诺诺网等被点名

PHP魔术方法：__invoke()

发表回复 取消回复

6. 请求转发（独立功能，固定主题 `endpoint`）

发表回复取消回复