什么是边缘计算服务器

# 在边缘服务器上安装 K3s（轻量 Kubernetes，单二进制，约 100MB）
curl -sfL https://get.k3s.io | sh -s - server \
  --disable traefik \
  --disable servicelb \
  --node-label topology.kubernetes.io/zone=edge-shanghai-01 \
  --node-taint node-role.kubernetes.io/edge=true:NoExecute \
  --data-dir /data/k3s

# 查看节点状态与资源
sudo kubectl get nodes -o wide
sudo kubectl describe node | grep -A5 "Taints"

# 查看节点资源分配情况
sudo kubectl top node

apiVersion: apps/v1
kind: Deployment
metadata:
  name: edge-inference
  namespace: edge
spec:
  replicas: 1
  selector:
    matchLabels:
      app: edge-inference
  template:
    metadata:
      labels:
        app: edge-inference
    spec:
      nodeSelector:
        topology.kubernetes.io/zone: edge-shanghai-01
      tolerations:
        - key: node-role.kubernetes.io/edge
          operator: Equal
          value: "true"
          effect: NoExecute
      containers:
        - name: infer
          image: registry.example.com/edge/yolo-infer:1.4.2
          resources:
            requests:
              cpu: "2"
              memory: 4Gi
            limits:
              cpu: "4"
              memory: 8Gi
          volumeMounts:
            - name: local-cache
              mountPath: /var/cache/infer
      volumes:
        - name: local-cache
          hostPath:
            path: /data/cache
            type: DirectoryOrCreate

#!/usr/bin/env python3
# edge_aggregator.py —— 边缘节点本地聚合，仅回传摘要，断网时落盘续传
import json, time, os, sqlite3
from datetime import datetime

LOCAL_DB = "/data/cache/edge_buffer.db"
UPLOAD_THRESHOLD = 500   # 累积 500 条才回传一次中心云


def init_db():
    os.makedirs(os.path.dirname(LOCAL_DB), exist_ok=True)
    conn = sqlite3.connect(LOCAL_DB)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS events (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            ts TEXT NOT NULL,
            device_id TEXT NOT NULL,
            payload TEXT NOT NULL,
            uploaded INTEGER DEFAULT 0
        )
    """)
    conn.execute("CREATE INDEX IF NOT EXISTS idx_uploaded ON events(uploaded)")
    conn.commit()
    return conn


def write_event(conn, device_id, payload):
    conn.execute(
        "INSERT INTO events(ts, device_id, payload, uploaded) VALUES (?,?,?,0)",
        (datetime.utcnow().isoformat(), device_id, json.dumps(payload, ensure_ascii=False))
    )
    conn.commit()


def pending_count(conn):
    return conn.execute("SELECT COUNT(*) FROM events WHERE uploaded=0").fetchone()[0]


def flush_to_cloud(conn):
    """回传中心云：成功才标记 uploaded=1，失败保留本地下轮重试"""
    rows = conn.execute(
        "SELECT id, ts, device_id, payload FROM events WHERE uploaded=0 ORDER BY id LIMIT ?",
        (UPLOAD_THRESHOLD,)
    ).fetchall()
    if not rows:
        return 0
    try:
        # TODO: 替换为真实的中心云上报接口，超时必须设置，避免边缘线程被拖死
        # requests.post("https://cloud.example.com/api/ingest", json=rows, timeout=5)
        ids = [r[0] for r in rows]
        conn.executemany("UPDATE events SET uploaded=1 WHERE id=?", [(i,) for i in ids])
        conn.commit()
        return len(ids)
    except Exception as exc:
        print(f"[WARN] 回传失败，数据保留本地: {exc}")
        return 0


if __name__ == "__main__":
    conn = init_db()
    while True:
        # 示例：模拟从设备读取数据，实际应替换为 MQTT / Modbus / SDK 订阅
        write_event(conn, "cam-001", {"type": "intrusion", "score": 0.93})
        if pending_count(conn) >= UPLOAD_THRESHOLD:
            sent = flush_to_cloud(conn)
            print(f"[INFO] 已回传 {sent} 条，本地待传 {pending_count(conn)} 条")
        time.sleep(1)

# 到中心云的往返延迟（回源路径）
ping -c 20 -i 0.2 cloud.example.com

# 到边缘节点的往返延迟（本地路径）
ping -c 20 -i 0.2 edge-gw.local

# 逐跳定位延迟发生在哪一段
mtr -n -c 50 --report cloud.example.com

# TCP 层延迟与建连耗时
curl -w "connect:%{time_connect}s ttfb:%{time_starttransfer}s total:%{time_total}s\n" \
     -o /dev/null -s https://edge-gw.local/health

# 实测边缘节点到中心云的可用带宽
iperf3 -c cloud.example.com -t 30 -P 4