添加 MySQL 8 到 MySQL 5.7 数据库迁移脚本

admin 页面 MQTT 服务状态增加去重队列长度显示
添加 MQTT 消息去重队列：Hook 层基于 payload+topic hash 去重，TTL 15 秒，定时清理
2026-06-30 12:36:43 +08:00 · 2026-06-30 12:18:40 +08:00 · 2026-06-30 12:10:29 +08:00
6 changed files with 369 additions and 3 deletions
@@ -0,0 +1,78 @@
+package mqttforward
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"sync"
+	"time"
+)
+
+const dedupTTL = 15 * time.Second
+
+type DedupQueue struct {
+	mu      sync.Mutex
+	entries map[string]time.Time
+	stopCh  chan struct{}
+}
+
+func NewDedupQueue() *DedupQueue {
+	return &DedupQueue{
+		entries: make(map[string]time.Time),
+		stopCh:  make(chan struct{}),
+	}
+}
+
+func (dq *DedupQueue) TryForward(topic string, payload []byte) bool {
+	hash := dedupHash(topic, payload)
+	now := time.Now()
+	dq.mu.Lock()
+	defer dq.mu.Unlock()
+	if expiry, ok := dq.entries[hash]; ok && now.Before(expiry) {
+		return false
+	}
+	dq.entries[hash] = now.Add(dedupTTL)
+	return true
+}
+
+func (dq *DedupQueue) Start() {
+	go func() {
+		ticker := time.NewTicker(dedupTTL)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				dq.cleanup()
+			case <-dq.stopCh:
+				return
+			}
+		}
+	}()
+}
+
+func (dq *DedupQueue) Stop() {
+	close(dq.stopCh)
+}
+
+func (dq *DedupQueue) Len() int {
+	dq.mu.Lock()
+	defer dq.mu.Unlock()
+	return len(dq.entries)
+}
+
+func (dq *DedupQueue) cleanup() {
+	now := time.Now()
+	dq.mu.Lock()
+	defer dq.mu.Unlock()
+	for hash, expiry := range dq.entries {
+		if now.After(expiry) {
+			delete(dq.entries, hash)
+		}
+	}
+}
+
+func dedupHash(topic string, payload []byte) string {
+	h := sha256.New()
+	h.Write([]byte(topic))
+	h.Write(payload)
+	return hex.EncodeToString(h.Sum(nil))
+}
@@ -30,6 +30,7 @@ type MQTTRuntimeStatus struct {
 	Stats       *mqttforwardpkg.Stats
 	ClientStats *mqttforwardpkg.ClientStats
 	DBQueue     *storepkg.WriteQueue
+	DedupQueue  *mqttforwardpkg.DedupQueue
 }

 // AdminMQTTStatus 是 admin 路由 GET /admin/mqtt-status 返回的 JSON 视图。
@@ -50,6 +51,7 @@ type AdminMQTTStatus struct {
 	MessagesSent        int64             `json:"messages_sent"`
 	MessagesDropped     int64             `json:"messages_dropped"`
 	DBWriteQueueLength  int               `json:"db_write_queue_length"`
+	DedupQueueLength    int               `json:"dedup_queue_len"`
 	Retained            int64             `json:"retained"`
 	Inflight            int64             `json:"inflight"`
 	InflightDropped     int64             `json:"inflight_dropped"`
@@ -71,7 +73,7 @@ type AdminMQTTClient struct {
 // Status 实现 MQTTStatusProvider。
 func (m MQTTRuntimeStatus) Status() AdminMQTTStatus {
 	if m.Server == nil || m.Server.Info == nil {
-		return AdminMQTTStatus{Running: false, Address: m.Address, TLS: m.TLS, DBWriteQueueLength: m.DBQueue.Len()}
+		return AdminMQTTStatus{Running: false, Address: m.Address, TLS: m.TLS, DBWriteQueueLength: m.DBQueue.Len(), DedupQueueLength: m.dedupQueueLen()}
 	}
 	info := m.Server.Info.Clone()
 	status := AdminMQTTStatus{
@@ -91,6 +93,7 @@ func (m MQTTRuntimeStatus) Status() AdminMQTTStatus {
 		MessagesSent:        m.Stats.Forwarded(),
 		MessagesDropped:     m.Stats.Dropped(),
 		DBWriteQueueLength:  m.DBQueue.Len(),
+		DedupQueueLength:    m.dedupQueueLen(),
 		Retained:            info.Retained,
 		Inflight:            info.Inflight,
 		InflightDropped:     info.InflightDropped,
@@ -136,6 +139,13 @@ func mqttClientInfo(c *mqtt.Client) mqttClientInfoView {
 	}
 }

+func (m MQTTRuntimeStatus) dedupQueueLen() int {
+	if m.DedupQueue == nil {
+		return 0
+	}
+	return m.DedupQueue.Len()
+}
+
 // DisconnectClient 实现 MQTTStatusProvider：发送 Disconnect 报文并关闭连接。
 // 使用 ErrAdministrativeAction 作为断开理由，便于日志区分。
 func (m MQTTRuntimeStatus) DisconnectClient(clientID string) bool {
@@ -57,8 +57,9 @@ type meshtasticFilterHook struct {
 	settings        *rspkg.Cache
 	pkiResolver     func(toNodeNum, fromNodeNum uint32) ([]byte, []byte, bool)
 	autoAcker       func(record map[string]any)
-	consoleLog      bool // 控制台是否打印 MQTT 连接/订阅事件
+	consoleLog       bool // 控制台是否打印 MQTT 连接/订阅事件
 	packetConsoleLog bool // 控制台是否打印 Meshtastic 数据包
+	dedupQueue       *mqttforwardpkg.DedupQueue
 }

 // ID 返回用于识别 Meshtastic payload 过滤器的 hook 名称。
@@ -225,6 +226,10 @@ func (h *meshtasticFilterHook) OnPublish(cl *mqtt.Client, pk packets.Packet) (pa
 		h.rejectPublish(cl, pk, record)
 		return pk, packets.ErrRejectPacket
 	}
+	if h.dedupQueue != nil && !h.dedupQueue.TryForward(pk.TopicName, pk.Payload) {
+		h.stats.IncDropped()
+		return pk, packets.ErrRejectPacket
+	}
 	h.stats.IncForwarded()

 	h.dbQueue.EnqueueRecord(record, mqttClientInfoFromClient(cl))
@@ -469,7 +474,7 @@ func run(cfg *configpkg.Config) error {
 		if err != nil {
 			return err
 		}
-		mqttStatus := webpkg.MQTTRuntimeStatus{Server: server, Address: mqttAddr, TLS: cfg.MQTT.TLS.Enabled, Stats: messageStats, ClientStats: clientStats, DBQueue: dbQueue}
+		mqttStatus := webpkg.MQTTRuntimeStatus{Server: server, Address: mqttAddr, TLS: cfg.MQTT.TLS.Enabled, Stats: messageStats, ClientStats: clientStats, DBQueue: dbQueue, DedupQueue: mqttHook.dedupQueue}
 		handler := webpkg.NewRouter(cfg.Web, cfg.ConsoleLog.Web, store, sessions, mqttStatus, blocking, forwardManager, settings, botSender, aiService)
 		webAddresses := []string{}
 		if cfg.Web.PortEnabled {
@@ -521,6 +526,9 @@ func run(cfg *configpkg.Config) error {
 	if err := server.Close(); err != nil && runErr == nil {
 		runErr = err
 	}
+	if mqttHook.dedupQueue != nil {
+		mqttHook.dedupQueue.Stop()
+	}
 	return runErr
 }

@@ -529,6 +537,8 @@ func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *stor
 	if err := server.AddHook(new(mqttauth.AllowHook), nil); err != nil {
 		return nil, nil, "", err
 	}
+	dedupQueue := mqttforwardpkg.NewDedupQueue()
+	dedupQueue.Start()
 	hook := &meshtasticFilterHook{
 		server:           server,
 		key:              cfg.Key,
@@ -540,6 +550,7 @@ func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *stor
 		pkiResolver:      botpkg.NewPKIKeyResolver(store),
 		consoleLog:       cfg.ConsoleLog.MQTT,
 		packetConsoleLog: cfg.ConsoleLog.Meshtastic,
+		dedupQueue:       dedupQueue,
 	}
 	if err := server.AddHook(hook, nil); err != nil {
 		return nil, nil, "", err
@@ -200,6 +200,7 @@ onBeforeUnmount(() => {
        <div><span>订阅数</span><strong>{{ status.subscriptions }}</strong></div>
        <div><span>转发消息</span><strong>{{ status.messages_sent }}</strong></div>
        <div><span>数据库队列</span><strong>{{ status.db_write_queue_length }}</strong></div>
+        <div><span>去重队列</span><strong>{{ status.dedup_queue_len }}</strong></div>
        <a class="status-card-link" href="/admin/discard_details"><span>丢弃消息</span><strong>{{ status.messages_dropped }}</strong></a>
        <div><span>收到包</span><strong>{{ status.packets_received }}</strong></div>
        <div><span>发送包</span><strong>{{ status.packets_sent }}</strong></div>
@@ -364,6 +364,7 @@ export interface AdminMqttStatus {
  messages_sent: number
  messages_dropped: number
  db_write_queue_length: number
+  dedup_queue_len: number
  retained: number
  inflight: number
  inflight_dropped: number
@@ -0,0 +1,265 @@
+#!/usr/bin/env python3
+"""
+MySQL 8 -> MySQL 5.7 数据库迁移脚本
+
+依赖: pip install pymysql
+
+使用方法:
+    python py/migrate_mysql8_to_mysql57.py
+
+功能:
+    1. 从源 MySQL 8 读取表结构，修正 collation 后在目标 MySQL 5.7 建表
+    2. 逐表批量拷贝数据
+    3. 修正 auto-increment 值
+
+注意:
+    - 目标库应为空库（无同名表）
+    - 运行前请确保源库和目标库都可连接
+"""
+
+from __future__ import annotations
+
+import re
+import sys
+from typing import Any
+
+import pymysql
+import pymysql.cursors
+
+# ============================================================
+# 硬编码数据库配置
+# ============================================================
+
+SOURCE_CONFIG = {
+    "host": "127.0.0.1",
+    "port": 3306,
+    "user": "root",
+    "password": "PLEASE_CHANGE_ME",
+    "database": "meshtastic",
+    "charset": "utf8mb4",
+}
+
+TARGET_CONFIG = {
+    "host": "127.0.0.1",
+    "port": 3307,
+    "user": "root",
+    "password": "PLEASE_CHANGE_ME",
+    "database": "meshtastic",
+    "charset": "utf8mb4",
+}
+
+BATCH_SIZE = 5000
+
+# ============================================================
+# 工具函数
+# ============================================================
+
+
+def _conn(config: dict[str, Any]) -> pymysql.Connection:
+    """创建数据库连接，使用 DictCursor 方便按列名访问"""
+    return pymysql.connect(
+        host=config["host"],
+        port=config["port"],
+        user=config["user"],
+        password=config["password"],
+        database=config["database"],
+        charset=config["charset"],
+        cursorclass=pymysql.cursors.DictCursor,
+    )
+
+
+def fix_collation(ddl: str) -> str:
+    """将 MySQL 8 默认 collation 替换为 MySQL 5.7 兼容版本"""
+    ddl = ddl.replace("utf8mb4_0900_ai_ci", "utf8mb4_general_ci")
+    ddl = ddl.replace("utf8mb4_0900_as_ci", "utf8mb4_general_ci")
+    return ddl
+
+
+def get_all_tables(conn: pymysql.Connection) -> list[str]:
+    with conn.cursor() as cur:
+        cur.execute("SHOW TABLES")
+        key = list(cur.description[0])[0]
+        return [row[key] for row in cur.fetchall()]
+
+
+def get_table_columns(conn: pymysql.Connection, table: str) -> list[str]:
+    """返回表的所有列名（按顺序）"""
+    with conn.cursor() as cur:
+        cur.execute("SHOW COLUMNS FROM `%s`" % table)
+        return [row["Field"] for row in cur.fetchall()]
+
+
+def get_auto_increment(
+    conn: pymysql.Connection, table: str
+) -> int | None:
+    """获取某张表当前的 AUTO_INCREMENT 值"""
+    with conn.cursor() as cur:
+        cur.execute(
+            "SELECT AUTO_INCREMENT FROM information_schema.TABLES "
+            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s",
+            (table,),
+        )
+        row = cur.fetchone()
+        if row and row["AUTO_INCREMENT"]:
+            return int(row["AUTO_INCREMENT"])
+    return None
+
+
+def row_count(conn: pymysql.Connection, table: str) -> int:
+    with conn.cursor() as cur:
+        cur.execute("SELECT COUNT(*) AS cnt FROM `%s`" % table)
+        return cur.fetchone()["cnt"]
+
+
+# ============================================================
+# 主流程
+# ============================================================
+
+
+def migrate() -> int:
+    src = _conn(SOURCE_CONFIG)
+    tgt = _conn(TARGET_CONFIG)
+    print(f"[连接] 源 MySQL 8 @ {SOURCE_CONFIG['host']}:{SOURCE_CONFIG['port']}")
+    print(f"[连接] 目标 MySQL 5.7 @ {TARGET_CONFIG['host']}:{TARGET_CONFIG['port']}")
+    print()
+
+    # 1. 在目标库建库（如还不存在）
+    try:
+        with tgt.cursor() as cur:
+            cur.execute(
+                "CREATE DATABASE IF NOT EXISTS `%s` "
+                "CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci"
+                % TARGET_CONFIG["database"]
+            )
+            cur.execute("USE `%s`" % TARGET_CONFIG["database"])
+        tgt.select_db(TARGET_CONFIG["database"])
+    except Exception as e:
+        print(f"[错误] 创建目标数据库失败: {e}")
+        return 1
+
+    tables = get_all_tables(src)
+    print(f"[发现] 源库共 {len(tables)} 张表: {', '.join(tables)}")
+    print()
+
+    # 2. 逐表建表（修正 collation）
+    print("=" * 60)
+    print("阶段 1: 在目标库创建表结构")
+    print("=" * 60)
+    for idx, table in enumerate(tables, 1):
+        with src.cursor() as cur:
+            cur.execute("SHOW CREATE TABLE `%s`" % table)
+            row = cur.fetchone()
+            ddl = row["Create Table"]
+        ddl = fix_collation(ddl)
+
+        try:
+            with tgt.cursor() as cur:
+                cur.execute(ddl)
+            tgt.commit()
+            print(f"  [{idx:2d}/{len(tables)}] OK `{table}`")
+        except Exception as e:
+            print(f"  [{idx:2d}/{len(tables)}] 错误 `{table}`: {e}")
+            tgt.rollback()
+            return 1
+
+    print()
+
+    # 3. 逐表拷贝数据
+    print("=" * 60)
+    print("阶段 2: 拷贝表数据")
+    print("=" * 60)
+
+    total_rows_copied = 0
+    auto_increments: dict[str, int | None] = {}
+
+    for idx, table in enumerate(tables, 1):
+        columns = get_table_columns(src, table)
+        if not columns:
+            auto_increments[table] = None
+            print(f"  [{idx:2d}/{len(tables)}] SKIP `{table}` (0 列)")
+            continue
+
+        col_quoted = ", ".join("`%s`" % c for c in columns)
+        placeholders = ", ".join(["%s"] * len(columns))
+        insert_sql = "INSERT INTO `%s` (%s) VALUES (%s)" % (
+            table,
+            col_quoted,
+            placeholders,
+        )
+
+        table_count = 0
+        with src.cursor() as read_cur:
+            read_cur.execute("SELECT * FROM `%s`" % table)
+            batch = read_cur.fetchmany(BATCH_SIZE)
+            while batch:
+                rows_values = [
+                    [row.get(c) for c in columns] for row in batch
+                ]
+                try:
+                    with tgt.cursor() as write_cur:
+                        write_cur.executemany(insert_sql, rows_values)
+                    tgt.commit()
+                    table_count += len(rows_values)
+                    print(
+                        f"\r  [{idx:2d}/{len(tables)}] `{table}` -> {table_count} 行",
+                        end="",
+                        flush=True,
+                    )
+                except Exception as e:
+                    print()
+                    print(f"  [{idx:2d}/{len(tables)}] 错误 `{table}`: {e}")
+                    tgt.rollback()
+                    return 1
+                batch = read_cur.fetchmany(BATCH_SIZE)
+
+        # 修正 auto_increment
+        ai = get_auto_increment(src, table)
+        auto_increments[table] = ai
+        if ai is not None:
+            with tgt.cursor() as cur:
+                cur.execute(
+                    "ALTER TABLE `%s` AUTO_INCREMENT = %s" % (table, ai)
+                )
+            tgt.commit()
+
+        total_rows_copied += table_count
+        print(
+            f"\r  [{idx:2d}/{len(tables)}] `{table}` -> {table_count} 行  [OK]"
+        )
+
+    # 4. 验证
+    print()
+    print("=" * 60)
+    print("阶段 3: 验证行数")
+    print("=" * 60)
+    all_match = True
+    src_total = 0
+    tgt_total = 0
+    for table in tables:
+        s = row_count(src, table)
+        t = row_count(tgt, table)
+        src_total += s
+        tgt_total += t
+        status = "OK" if s == t else "不匹配!"
+        if s != t:
+            all_match = False
+        print(f"  `{table}`: 源={s}  目标={t}  [{status}]")
+
+    print()
+    print(f"  总计: 源={src_total}  目标={tgt_total}")
+
+    src.close()
+    tgt.close()
+
+    if all_match:
+        print()
+        print("迁移完成，所有表行数一致。")
+        return 0
+    else:
+        print()
+        print("警告: 部分表行数不一致，请检查。")
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(migrate())
Author	SHA1	Message	Date
kevin	a75ab812d2	添加 MySQL 8 到 MySQL 5.7 数据库迁移脚本	2026-06-30 12:36:43 +08:00
kevin	01c7275763	admin 页面 MQTT 服务状态增加去重队列长度显示	2026-06-30 12:18:40 +08:00
kevin	4782e84c15	添加 MQTT 消息去重队列：Hook 层基于 payload+topic hash 去重，TTL 15 秒，定时清理	2026-06-30 12:10:29 +08:00