专栏：FRP 内网穿透从入门到实战
前置阅读：已完成正篇部署与 webServer 仪表盘配置（第二篇、第四篇）。
本篇目标：解决边缘网关临时远程运维中的两个工程难题——动态端口获取与空闲自动断开。

前言

在视频物联网平台中，大量设备通过边缘网关以长连接方式注册上线。运维人员远程调试网关 Web 管理页时，平台会向边缘网关下发指令，由网关侧 frpc 临时开启 TCP 穿透，把内网 80 端口暴露到公网 frps。

这和正篇里「一台设备固定映射一个 remotePort」的场景不同，有两个现实约束：

1. 端口多且临时：每台网关、每次远程会话都可能新建代理，不能给每个 client 预分配固定端口；
2. 用完即关：远程运维结束或操作者离开，穿透隧道不能一直挂着，否则公网端口长期暴露。

本篇方案源自畅联云平台边缘网关远程运维的生产实践，用 Python 示例说明如何通过 frps webServer API 解决上述两个问题。

一、业务架构：平台下发，网关执行

整体链路如下：

云平台 ──长连接──▶ 边缘网关（frpc）
  │                        │
  │  ① 下发 openRemoteConfig │  ② remotePort=0 动态建代理
  │                        │
  └──── ③ webServer API ───┼──▶ frps 查询实际端口
                           │
  ④ 定时任务轮询流量 ──────┘  ⑤ 空闲 15 分钟 → 下发 closeRemoteConfig

关键约定：

项目	说明
代理命名	{网关序列号}-80，如 GW20240001-80
动态端口	下发 remotePort = 0，由 frps 从端口池自动分配
端口查询	GET /api/proxies/{proxyName}，读取 conf.remotePort
活跃检测	每 2 分钟拉取 GET /api/proxy/tcp，对比 todayTrafficIn
空闲阈值	入站流量连续 15 分钟无变化，判定为空闲并关闭远程配置

二、frps 侧前置配置

动态端口分配需要在 frps 开启 webServer，并预留端口范围：

# frps.toml
bindPort = 7000
auth.method = "token"
auth.token = "YourStrongToken"

# 仪表盘 / API（平台内网调用，勿对公网裸奔）
webServer.addr = "0.0.0.0"
webServer.port = 7500
webServer.user = "admin"
webServer.password = "YourDashboardPass"

# 动态端口池（remotePort=0 时从此范围分配）
allowPorts = [
  { start = 20000, end = 30000 }
]

log.to = "/var/log/frps.log"
log.level = "info"

边缘网关侧 frpc 通过平台指令运行时创建代理（无需写死在 toml 里），核心参数示意：

{
  "clientID": "GW20240001",
  "serverAddr": "frps公网IP",
  "serverPort": 7000,
  "authToken": "YourStrongToken",
  "proxyName": "GW20240001-80",
  "remotePort": 0,
  "localIP": "127.0.0.1",
  "localPort": 80
}

remotePort = 0 是本篇第一个问题的关键：不指定固定端口，交给 frps 动态分配。

三、问题一：获取动态分配的 remotePort

3.1 为什么需要反查端口

代理创建是异步的：平台下发指令后，frpc 连接 frps 并完成注册需要 1～3 秒。此时平台还不知道最终映射到哪个公网端口，运维人员也就无法打开远程 Web 页。

推荐做法：下发开启指令后，轮询 frps webServer API，直到代理状态为 online 且拿到 conf.remotePort。

3.2 API 说明

接口	方法	用途
/api/proxies/{name}	GET	查询单个代理详情（含动态端口）
/api/proxy/tcp	GET	列出所有 TCP 代理（含流量统计）

认证方式：HTTP Basic Auth，用户名密码对应 webServer.user / webServer.password。

单个代理响应示例：

{
  "name": "GW20240001-80",
  "clientID": "GW20240001",
  "status": "online",
  "todayTrafficIn": 1024,
  "todayTrafficOut": 512,
  "lastStartTime": "08-13 14:30:00",
  "conf": {
    "remotePort": 23841
  }
}

3.3 Python 实现：等待动态端口就绪

import base64
import time
from dataclasses import dataclass
from typing import Optional

import requests


@dataclass
class FrpServerConfig:
    inner_ip: str          # frps 内网 IP，平台与 frps 同 VPC 时使用
    web_port: int          # webServer 端口，默认 7500
    web_user: str
    web_password: str
    out_ip: str            # frps 公网 IP，拼接远程访问 URL 用


class FrpWebApiClient:
    """frps webServer API 封装"""

    def __init__(self, config: FrpServerConfig):
        self.config = config
        token = base64.b64encode(
            f"{config.web_user}:{config.web_password}".encode()
        ).decode()
        self.headers = {"Authorization": f"Basic {token}"}

    @property
    def base_url(self) -> str:
        return f"http://{self.config.inner_ip}:{self.config.web_port}"

    def get_proxy(self, proxy_name: str) -> Optional[dict]:
        url = f"{self.base_url}/api/proxies/{proxy_name}"
        resp = requests.get(url, headers=self.headers, timeout=5)
        resp.raise_for_status()
        return resp.json()

    def list_tcp_proxies(self) -> list[dict]:
        url = f"{self.base_url}/api/proxy/tcp"
        resp = requests.get(url, headers=self.headers, timeout=5)
        resp.raise_for_status()
        data = resp.json()
        return data.get("proxies") or []


def get_remote_config_url(
    frp_client: FrpWebApiClient,
    proxy_name: str,
    retry: int = 3,
    interval: float = 3.0,
) -> Optional[str]:
    """
    等待动态代理上线，返回远程 Web 访问 URL。
    逻辑：下发 openRemoteConfig 后，轮询直到 status=online 且 remotePort 可用。
    """
    for attempt in range(retry + 1):
        proxy = frp_client.get_proxy(proxy_name)
        if proxy and proxy.get("status") == "online":
            remote_port = (proxy.get("conf") or {}).get("remotePort")
            if remote_port:
                return f"http://{frp_client.config.out_ip}:{remote_port}"
        if attempt < retry:
            time.sleep(interval)
    return None


# ---- 使用示例 ----
if __name__ == "__main__":
    frp = FrpWebApiClient(FrpServerConfig(
        inner_ip="10.0.0.5",
        web_port=7500,
        web_user="admin",
        web_password="YourDashboardPass",
        out_ip="123.45.67.89",
    ))

    gateway_serial = "GW20240001"
    proxy_name = f"{gateway_serial}-80"

    # ① 此处由平台向边缘网关下发 openRemoteConfig（remotePort=0）
    # send_open_remote_config(gateway_serial, proxy_name)

    # ② 轮询 API 获取动态端口
    url = get_remote_config_url(frp, proxy_name)
    if url:
        print(f"远程配置地址：{url}")
    else:
        print("代理未能在预期时间内上线")

生产要点：

• 平台调用 API 应走 frps 内网 IP（inner_ip），避免 webServer 暴露公网；
• 重试 3 次、间隔 3 秒，是生产环境验证过的参数；
• 返回 URL 直接给前端，运维人员点击即可访问网关 Web 页。

四、问题二：空闲 15 分钟自动断开

4.1 为什么用流量而不是连接数

TCP 连接在「页面打开但无操作」时可能仍然 ESTABLISHED，单纯看连接数无法判断是否真的在用。更可靠的做法是：对比 todayTrafficIn 是否在持续变化——有交互就有流量，无交互则流量停滞。

4.2 检测策略（与生产 Job 一致）

规则	说明
执行频率	每 2 分钟一次
活跃判定	todayTrafficIn 与上次不同 → 该代理活跃
空闲判定	流量连续 15 分钟无变化 → 空闲候选
启动保护	代理 lastStartTime 在 5 分钟内 → 跳过（避免刚创建就被误关）
多代理合并	同一 clientID 下多个 proxy，任一活跃则整台网关视为活跃
关闭动作	向边缘网关下发 closeRemoteConfig，frpc 删除临时代理

4.3 Python 实现：空闲检测定时任务

import logging
from datetime import datetime, timedelta
from typing import Dict, Set, Tuple

logger = logging.getLogger(__name__)

# 流量缓存：key = "{frp_server_id}_{proxy_name}" → (last_traffic_in, last_change_time)
_traffic_cache: Dict[str, Tuple[int, datetime]] = {}

INACTIVE_THRESHOLD_MINUTES = 15
STARTUP_GRACE_MINUTES = 5
CACHE_CLEANUP_HOURS = 24


def _parse_start_time(last_start_time: str) -> datetime | None:
    """解析 frps 返回的启动时间，格式如 '08-13 14:30:00'"""
    if not last_start_time:
        return None
    try:
        year = datetime.now().year
        return datetime.strptime(f"{year}-{last_start_time}", "%Y-%m-%d %H:%M:%S")
    except ValueError:
        return None


def _is_started_within_grace_period(last_start_time: str) -> bool:
    start = _parse_start_time(last_start_time)
    if not start:
        return False
    return abs((datetime.now() - start).total_seconds()) <= STARTUP_GRACE_MINUTES * 60


def check_proxy_activity(
    frp_server_id: str,
    proxy: dict,
    active_client_ids: Set[str],
    inactive_candidates: Set[str],
) -> None:
    """检测单个代理的活跃状态，更新缓存与候选集合"""
    name = proxy.get("name")
    client_id = proxy.get("clientID")
    if not name or not client_id:
        return

    cache_key = f"{frp_server_id}_{name}"
    current_traffic = proxy.get("todayTrafficIn") or 0
    now = datetime.now()

    cached = _traffic_cache.get(cache_key)
    if cached is None:
        _traffic_cache[cache_key] = (current_traffic, now)
        return

    last_traffic, last_change_time = cached
    if current_traffic != last_traffic:
        _traffic_cache[cache_key] = (current_traffic, now)
        active_client_ids.add(client_id)
    else:
        idle_minutes = (now - last_change_time).total_seconds() / 60
        if idle_minutes >= INACTIVE_THRESHOLD_MINUTES:
            inactive_candidates.add(client_id)


def run_activity_check(
    frp_client: FrpWebApiClient,
    frp_server_id: str,
    close_remote_config_fn,
) -> str:
    """
    主检测流程，等价于生产环境 FrpProxyActivityJob。
    close_remote_config_fn: callable(client_id) → 向边缘网关下发关闭指令
    """
    active_client_ids: Set[str] = set()
    inactive_candidates: Set[str] = set()
    current_keys: Set[str] = set()

    proxies = frp_client.list_tcp_proxies()
    for proxy in proxies:
        if proxy.get("status") == "offline":
            continue
        if _is_started_within_grace_period(proxy.get("lastStartTime", "")):
            continue

        cache_key = f"{frp_server_id}_{proxy['name']}"
        current_keys.add(cache_key)
        check_proxy_activity(
            frp_server_id, proxy, active_client_ids, inactive_candidates
        )

    # 多 proxy 合并：任一活跃 → 整台网关活跃
    final_inactive = inactive_candidates - active_client_ids

    # 清理 24 小时未出现的缓存
    now = datetime.now()
    expired = [
        k for k, (_, t) in _traffic_cache.items()
        if k not in current_keys
        and (now - t).total_seconds() >= CACHE_CLEANUP_HOURS * 3600
    ]
    for k in expired:
        del _traffic_cache[k]

    closed = 0
    for client_id in final_inactive:
        try:
            close_remote_config_fn(client_id)
            closed += 1
            logger.info("网关 %s 空闲超时，已关闭远程配置", client_id)
        except Exception:
            logger.exception("关闭网关 %s 远程配置失败", client_id)

    return (
        f"检测 {len(proxies)} 个代理，"
        f"发现 {len(final_inactive)} 个不活跃网关，"
        f"已关闭 {closed} 个"
    )

4.4 接入定时调度

import schedule
import time

frp = FrpWebApiClient(FrpServerConfig(
    inner_ip="10.0.0.5",
    web_port=7500,
    web_user="admin",
    web_password="YourDashboardPass",
    out_ip="123.45.67.89",
))

def close_remote_config(client_id: str) -> None:
    """向边缘网关下发 closeRemoteConfig（平台长连接通道实现）"""
    # platform.send_command(client_id, "closeRemoteConfig", {})
    pass

def job():
    result = run_activity_check(frp, frp_server_id="1", close_remote_config_fn=close_remote_config)
    print(f"[{datetime.now():%H:%M:%S}] {result}")

schedule.every(2).minutes.do(job)

if __name__ == "__main__":
    while True:
        schedule.run_pending()
        time.sleep(1)

五、两个问题的协作时序

运维人员点击「远程配置」
        │
        ▼
平台 ──openRemoteConfig(remotePort=0)──▶ 边缘网关 frpc 建代理
        │
        ▼
平台轮询 GET /api/proxies/{serial}-80（最多 3 次）
        │
        ▼
返回 http://公网IP:{动态端口} 给前端
        │
        ▼
运维人员操作网关 Web 页（产生 todayTrafficIn 变化）
        │
        ▼
定时任务每 2 分钟检测流量
        │
        ├── 有变化 → 重置计时，保持穿透
        │
        └── 15 分钟无变化 → closeRemoteConfig → 代理删除，端口释放

六、常见问题

Q1：remotePort=0 后 API 一直查不到端口？

• 检查 frps 是否配置了 allowPorts 端口池；
• 确认边缘网关 frpc 已成功连接 frps（仪表盘可见 client）；
• 适当加大轮询次数或间隔。

Q2：刚打开远程页就被自动关闭？

• 确认定时任务对 lastStartTime 5 分钟内的新代理做了启动保护；
• 首次检测只写缓存、不做空闲判定。

Q3：页面开着但流量不变，会被误关吗？

• 会。纯静态页面无请求时，流量确实可能不变。生产环境可结合前端心跳，或适当延长阈值（如 30 分钟）。

Q4：webServer API 要不要暴露公网？

• 不要。平台与 frps 同 VPC，走内网 IP 调用即可；仪表盘端口安全组仅放行运维 IP。

本篇小结

1. 边缘网关临时远程运维场景下，remotePort=0 + webServer API 反查，是动态端口的标准解法；

2. GET /api/proxies/{name} 获取实际端口，GET /api/proxy/tcp 批量拉取流量统计；

3. 以 todayTrafficIn 变化判断活跃，15 分钟无变化自动关闭，比看连接数更可靠；

4. 多代理按 clientID 合并判定，新代理 5 分钟保护期避免误关；

5. 方案可直接移植到同类视频物联网平台的边缘网关远程运维场景。

系列导航

• 正篇回顾：建议从第一篇至第七篇顺序阅读
• 本篇为番外实战，聚焦边缘网关 + FRP 工程化，与正篇互补

💡 提示：若你的业务涉及批量边缘网关与视频设备接入，可了解畅联云平台——面向视频物联网的底层设备接入平台。若只需个人级 FRP 穿透，正篇教程已足够。