fix(scale): 修复子设备重启后主设备看不到连接数据的问题，新增清除数据广播功能

- ScaleWebSocketClient：引入版本号机制防止 UDP 触发的新连接与指数退避重连任务并发建立两条连接；onFailure/onClosed 改用 remove(key, value) 原子操作，避免旧连接超时回调误删新连接引用 - ScaleWebSocketServer：新增 latestData 为空时延迟 3 秒补推机制，解决子设备刚重启时 WeightUtil 尚无读数导致主设备看不到数据的问题；新增 onClearData 回调处理清除数据事件；stop() 时同步关闭 scheduler - ScaleEvent：新增 TYPE_CLEAR_DATA 事件类型 - ScaleServiceManager：新增 onClearData 属性和 sendClearData() 方法 - SlaveActivity：监听 onClearData 事件，收到后清空本机数据并重置 SP 状态；启动时预填空占位格子 - SettingActivity/DbViewModel/DbRepository：clearAllTestData 重命名为 clearAllData，清除完成后通知子设备并跳转 HomeActivity - scale包架构分析.md：同步更新文档 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-08 15:48:59 +08:00
parent a0137d6979
commit 436de88d73
10 changed files with 147 additions and 31 deletions
@@ -227,7 +227,7 @@ class DbRepository {
     * 物理清空全部 4 张表的测试数据
     * 清除顺序：先删子表（食材），再删主表（菜品），最后删配置表（调料、槽位）
     */
-    suspend fun clearAllTestData() = withContext(Dispatchers.IO) {
+    suspend fun clearAllData() = withContext(Dispatchers.IO) {
        db.cookFoodGoodsDao().deleteAll()
        db.cookFoodDao().deleteAll()
        db.seasoningDao().deleteAll()
@@ -243,9 +243,9 @@ class DbViewModel : ViewModel() {
     * 物理清空全部 4 张表的测试数据，完成后通过回调通知 UI
     * @param onDone 清除完成后在主线程执行的回调
     */
-    fun clearAllTestData(onDone: () -> Unit) {
+    fun clearAllData(onDone: () -> Unit) {
        viewModelScope.launch {
-            rep.clearAllTestData()
+            rep.clearAllData()
            onDone()
        }
    }
@@ -34,5 +34,7 @@ data class ScaleEvent(
        const val TYPE_SEASONING_ADDED = "seasoning_added"
        /** 调料槽位配置同步事件：主设备配置变更后广播给子设备 */
        const val TYPE_SEASONING_CONFIG = "seasoning_config"
+        /** 清除数据事件：主设备通知所有子设备清除本机测试数据 */
+        const val TYPE_CLEAR_DATA = "clear_data"
    }
 }
@@ -71,6 +71,14 @@ object ScaleServiceManager {
        get() = wsServer?.onSeasoningConfig
        set(value) { wsServer?.onSeasoningConfig = value }

+    /**
+     * 收到主设备下发的清除数据事件时的回调，仅子设备有效
+     * 在子线程中调用，需自行切换到主线程更新 UI
+     */
+    var onClearData: ((ScaleEvent) -> Unit)?
+        get() = wsServer?.onClearData
+        set(value) { wsServer?.onClearData = value }
+
    /**
     * 向所有已连接子设备广播调料配置（主设备调用）
     * @param slots 全量槽位配置列表
@@ -79,6 +87,16 @@ object ScaleServiceManager {
        wsClient?.sendToAllSlaves(buildConfigEvent(slots))
    }

+    /**
+     * 通知所有已连接子设备清除本机数据（主设备调用）
+     * 子设备收到后执行与主设备相同的清除逻辑
+     */
+    fun sendClearData() {
+        wsClient?.sendToAllSlaves(
+            ScaleEvent(type = ScaleEvent.TYPE_CLEAR_DATA, deviceId = GlobalData.deviceId)
+        )
+    }
+
    /**
     * 向单台刚连接的子设备推送调料配置（连接时按需调用）
     * @param deviceId 目标子设备 ID
@@ -43,6 +43,11 @@ class ScaleWebSocketClient {
    private val reconnectDelays = ConcurrentHashMap<String, Long>()
    /** 已主动移除的设备集合，不对其触发重连 */
    private val removedDevices = ConcurrentHashMap.newKeySet<String>()
+    /**
+     * 各设备的连接版本号，每次调用 connect() 时递增
+     * doConnect 执行前校验版本，版本不匹配说明已有更新的连接请求，直接放弃本次连接
+     */
+    private val connectVersions = ConcurrentHashMap<String, Int>()

    private val scheduler: ScheduledExecutorService = Executors.newSingleThreadScheduledExecutor()

@@ -66,13 +71,21 @@ class ScaleWebSocketClient {
     */
    fun connect(deviceId: String, host: String, port: Int) {
        removedDevices.remove(deviceId)
-        // 取消旧 IP 的重连任务，防止 IP 变化时新旧任务并发竞争
+        // 取消旧重连任务，防止 IP 变化时新旧任务并发竞争
        reconnectTasks.remove(deviceId)?.cancel(false)
        reconnectDelays[deviceId] = RECONNECT_BASE_MS
-        doConnect(deviceId, host, port)
+        // 递增版本号：doConnect 执行前会校验版本，旧任务版本不匹配时直接放弃
+        val version = (connectVersions[deviceId] ?: 0) + 1
+        connectVersions[deviceId] = version
+        doConnect(deviceId, host, port, version)
    }

-    private fun doConnect(deviceId: String, host: String, port: Int) {
+    private fun doConnect(deviceId: String, host: String, port: Int, version: Int = connectVersions[deviceId] ?: 0) {
+        // 版本校验：若当前版本已被更新的 connect() 调用覆盖，放弃本次连接
+        if (connectVersions[deviceId] != version) {
+            Log.d(TAG, "放弃过期连接任务: $deviceId, version=$version")
+            return
+        }
        val url = "ws://$host:$port"
        val request = Request.Builder().url(url).build()

@@ -103,16 +116,23 @@ class ScaleWebSocketClient {

            override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
                Log.w(TAG, "连接子设备失败: $deviceId, ${t.message}")
-                connections.remove(deviceId)
-                onDeviceDisconnected?.invoke(deviceId)
-                scheduleReconnect(deviceId, host, port)
+                // 使用 remove(key, value) 确保只移除自己的实例
+                // 避免旧连接超时触发时误删新连接引用，导致 aggregator 错误清空新连接数据
+                val removed = connections.remove(deviceId, webSocket)
+                if (removed) {
+                    onDeviceDisconnected?.invoke(deviceId)
+                    scheduleReconnect(deviceId, host, port)
+                }
            }

            override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
                Log.d(TAG, "子设备连接已关闭: $deviceId, reason=$reason")
-                connections.remove(deviceId)
-                onDeviceDisconnected?.invoke(deviceId)
-                scheduleReconnect(deviceId, host, port)
+                // 同上：只有移除的是自己，才触发断线回调和重连
+                val removed = connections.remove(deviceId, webSocket)
+                if (removed) {
+                    onDeviceDisconnected?.invoke(deviceId)
+                    scheduleReconnect(deviceId, host, port)
+                }
            }
        })

@@ -129,10 +149,12 @@ class ScaleWebSocketClient {
        if (scheduler.isShutdown) return
        val delay = reconnectDelays[deviceId] ?: RECONNECT_BASE_MS
        Log.d(TAG, "将在 ${delay}ms 后重连子设备: $deviceId")
+        // 记录当前版本，延迟任务执行时再次校验，防止期间有新的 connect() 调用
+        val version = connectVersions[deviceId] ?: 0

        val future = scheduler.schedule({
            if (!removedDevices.contains(deviceId)) {
-                doConnect(deviceId, host, port)
+                doConnect(deviceId, host, port, version)
            }
        }, delay, TimeUnit.MILLISECONDS)

@@ -188,6 +210,7 @@ class ScaleWebSocketClient {
        reconnectTasks.remove(deviceId)?.cancel(false)
        connections.remove(deviceId)?.close(1000, "设备离线")
        reconnectDelays.remove(deviceId)
+        connectVersions.remove(deviceId)
        Log.d(TAG, "已断开子设备: $deviceId")
    }

@@ -9,6 +9,8 @@ import org.java_websocket.WebSocket
 import org.java_websocket.handshake.ClientHandshake
 import org.java_websocket.server.WebSocketServer
 import java.net.InetSocketAddress
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit

 /**
 * 秤数据 WebSocket 服务端（主设备和子设备均运行）
@@ -20,10 +22,14 @@ class ScaleWebSocketServer(private val deviceId: String, private val context: Co
        private const val TAG = "ScaleWebSocketServer"
        /** 节流间隔：两次推送之间的最小间隔（毫秒） */
        private const val THROTTLE_MS = 100L
+        /** latestData 为空时延迟补推的时间（毫秒），给 WeightUtil 时间产生第一次读数 */
+        private const val RETRY_PUSH_DELAY_MS = 3_000L
    }

    private val gson = Gson()
    private var server: InternalServer? = null
+    /** 延迟补推任务的调度器 */
+    private val scheduler = Executors.newSingleThreadScheduledExecutor()

    /** 主设备连接状态变化回调：true=已连接，false=已断开；在主线程外调用，需自行切换线程 */
    var onConnectionChanged: ((connected: Boolean) -> Unit)? = null
@@ -31,6 +37,9 @@ class ScaleWebSocketServer(private val deviceId: String, private val context: Co
    /** 收到主设备下发的调料配置同步事件时的回调，在子线程调用 */
    var onSeasoningConfig: ((event: ScaleEvent) -> Unit)? = null

+    /** 收到主设备下发的清除数据事件时的回调，在子线程调用 */
+    var onClearData: ((event: ScaleEvent) -> Unit)? = null
+
    /** 当前是否有主设备连接 */
    val isConnected: Boolean
        get() = server?.connections?.isNotEmpty() == true
@@ -77,6 +86,8 @@ class ScaleWebSocketServer(private val deviceId: String, private val context: Co
        WeightUtil.removeWeightListener(TAG)
        onConnectionChanged = null
        onSeasoningConfig = null
+        onClearData = null
+        scheduler.shutdownNow()
        try {
            server?.stop(1000)
        } catch (e: Exception) {
@@ -124,13 +135,32 @@ class ScaleWebSocketServer(private val deviceId: String, private val context: Co
            connectionCount++
            Log.d(TAG, "主设备已连接: ${conn.remoteSocketAddress}")
            onConnectionChanged?.invoke(true)
+
            // 新客户端连接后，立即推送所有秤的最新数据
-            latestData.values.forEach { data ->
-                try {
-                    conn.send(gson.toJson(data))
-                } catch (e: Exception) {
-                    Log.w(TAG, "初始推送失败: ${e.message}")
+            if (latestData.isNotEmpty()) {
+                latestData.values.forEach { data ->
+                    try {
+                        conn.send(gson.toJson(data))
+                    } catch (e: Exception) {
+                        Log.w(TAG, "初始推送失败: ${e.message}")
+                    }
                }
+            } else {
+                // latestData 为空说明刚重启，WeightUtil 还未产生读数
+                // 延迟 3 秒后补推一次，确保主设备能看到本机秤
+                Log.d(TAG, "latestData 为空，${RETRY_PUSH_DELAY_MS}ms 后补推")
+                scheduler.schedule({
+                    if (conn.isOpen && latestData.isNotEmpty()) {
+                        latestData.values.forEach { data ->
+                            try {
+                                conn.send(gson.toJson(data))
+                                Log.d(TAG, "补推数据: address=${data.address}, weight=${data.weight}")
+                            } catch (e: Exception) {
+                                Log.w(TAG, "补推失败: ${e.message}")
+                            }
+                        }
+                    }
+                }, RETRY_PUSH_DELAY_MS, TimeUnit.MILLISECONDS)
            }
        }

@@ -151,6 +181,7 @@ class ScaleWebSocketServer(private val deviceId: String, private val context: Co
                    val event = gson.fromJson(message, ScaleEvent::class.java)
                    when (event.type) {
                        ScaleEvent.TYPE_SEASONING_CONFIG -> onSeasoningConfig?.invoke(event)
+                        ScaleEvent.TYPE_CLEAR_DATA -> onClearData?.invoke(event)
                    }
                    return
                }
@@ -12,9 +12,9 @@ import androidx.core.content.ContextCompat
 import com.shuwei.dish.match.base.BaseActivity
 import com.shuwei.dish.match.databinding.ActivitySettingBinding
 import com.shuwei.dish.match.dialog.CommonDialog
+import com.shuwei.dish.match.scale.ScaleServiceManager
 import com.shuwei.dish.match.utils.SpTool
 import com.shuwei.dish.match.utils.ext.clickWithDebounce
-import com.shuwei.dish.match.utils.ext.gone
 import com.shuwei.dish.match.utils.ext.startActivity
 import com.shuwei.dish.match.utils.ext.toast
 import com.shuwei.dish.match.utils.ext.visible
@@ -49,7 +49,7 @@ class SettingActivity : BaseActivity() {
            it.clickWithDebounce {
                binding.llDbInspect.visible()
                binding.llScaleObserve.visible()
-                binding.llClearTestData.visible()
+                binding.llClearData.visible()
            }
        })

@@ -85,24 +85,28 @@ class SettingActivity : BaseActivity() {
        binding.llScaleObserve.clickWithDebounce {
            startActivity<MasterScaleActivity>()
        }
-        binding.llClearTestData.clickWithDebounce {
-            showClearTestDataDialog()
+        binding.llClearData.clickWithDebounce {
+            showClearDataDialog()
        }
    }

    /**
     * 弹出清除测试数据的二次确认弹窗
     */
-    private fun showClearTestDataDialog() {
+    private fun showClearDataDialog() {
        CommonDialog(this)
-            .setTitle("测试数据清除")
+            .setTitle("数据清除")
            .setContent("将物理删除全部 4 张表数据（菜品、食材、调料、槽位），此操作不可恢复，同时重置菜品模式和食堂id，确认继续？")
            .setNegativeButton("取消")
            .setPositiveButton("确认清除") {
-                appViewModel.clearAllTestData {
+                appViewModel.clearAllData {
                    SpTool.cookMode = -1
                    SpTool.canteenId = "0"
-                    toast("测试数据已清除")
+                    // 通知所有已连接子设备同步清除数据
+                    ScaleServiceManager.sendClearData()
+                    startActivity<HomeActivity> {
+                        flags = Intent.FLAG_ACTIVITY_NEW_TASK or Intent.FLAG_ACTIVITY_CLEAR_TASK
+                    }
                }
            }
            .show()
@@ -18,7 +18,6 @@ import com.shuwei.dish.match.adapter.Seasoning18GridAdapter
 import com.shuwei.dish.match.adapter.Seasoning22GridAdapter
 import com.shuwei.dish.match.adapter.ScaleRowAdapter
 import com.shuwei.dish.match.base.BaseActivity
-import com.shuwei.dish.match.base.BaseApp
 import com.shuwei.dish.match.base.GlobalData
 import com.shuwei.dish.match.databinding.ActivitySlaveBinding
 import com.shuwei.dish.match.dialog.CommonDialog
@@ -28,6 +27,7 @@ import com.shuwei.dish.match.scale.ScaleDeviceConfig
 import com.shuwei.dish.match.scale.ScaleServiceManager
 import com.shuwei.dish.match.utils.NetworkUtil
 import com.shuwei.dish.match.utils.SizeTool
+import com.shuwei.dish.match.utils.SpTool
 import com.shuwei.dish.match.utils.WeightUtil
 import com.shuwei.dish.match.utils.ext.dp
 import kotlinx.coroutines.launch
@@ -77,6 +77,7 @@ class SlaveActivity : BaseActivity() {
        listenLocalScales()
        listenMasterConnection()
        listenSeasoningConfig()
+        listenClearData()
        addBackKeyListener()
    }

@@ -109,6 +110,8 @@ class SlaveActivity : BaseActivity() {
                    it.onItemClick = { scale, _ -> showTareDialog(scale.address) }
                }
                binding.rvScaleList.adapter = scale22Adapter
+                // 启动时用空哨兵预填全部格子，不依赖 WeightUtil 回调才显示结构
+                scale22Adapter!!.updateByAddress(emptyList())
            }
            ScaleDeviceConfig.DEVICE_ID_18 -> {
                binding.rvScaleList.updateLayoutParams<LinearLayout.LayoutParams> { topMargin = 0 }
@@ -121,12 +124,18 @@ class SlaveActivity : BaseActivity() {
                    it.onItemClick = { scale, _ -> showTareDialog(scale.address) }
                }
                binding.rvScaleList.adapter = scale18Adapter
+                // 启动时用空哨兵预填全部格子，不依赖 WeightUtil 回调才显示结构
+                scale18Adapter!!.updateByAddress(emptyList())
            }
            else -> {
                binding.rvScaleList.updateLayoutParams<LinearLayout.LayoutParams> { topMargin = 20.dp }
                binding.rvScaleList.layoutManager = LinearLayoutManager(this)
                linearAdapter.onItemClick = { item -> showTareDialog(item.address) }
                binding.rvScaleList.adapter = linearAdapter
+                // 启动时预填空占位，不依赖 WeightUtil 回调才显示列表结构
+                ScaleDeviceConfig.SCALE_ORDER_1.forEach { addr ->
+                    linearAdapter.updateItem(ScaleRowAdapter.ScaleItem(addr, 0.0, WeightUtil.STATE_STABLE))
+                }
            }
        }
    }
@@ -177,6 +186,23 @@ class SlaveActivity : BaseActivity() {
        }
    }

+    /**
+     * 监听主设备下发的清除数据事件
+     * 收到后清空本机全部 4 张表数据并重置 SP 状态，与主设备保持一致
+     * 清除完成后刷新 UI，清空所有秤的调料名称显示
+     */
+    private fun listenClearData() {
+        ScaleServiceManager.onClearData = {
+            lifecycleScope.launch {
+                appViewModel.clearAllData {
+                    SpTool.cookMode = -1
+                    SpTool.canteenId = "0"
+                    runOnUiThread { applySlots(emptyList()) }
+                }
+            }
+        }
+    }
+
    /**
     * 将槽位配置应用到对应 adapter（按地址匹配格子位置）
     * 同时以传入列表为准全量刷新 slotNameMap：
@@ -174,7 +174,7 @@
    </LinearLayout>

    <LinearLayout
-        android:id="@+id/llClearTestData"
+        android:id="@+id/llClearData"
        android:layout_width="match_parent"
        android:layout_height="120dp"
        android:layout_marginHorizontal="32dp"
@@ -191,7 +191,7 @@
            android:layout_height="wrap_content"
            android:layout_marginStart="32dp"
            android:layout_weight="1"
-            android:text="测试数据清除"
+            android:text="数据清除"
            android:textColor="@color/black333"
            android:textSize="32sp"
            android:textStyle="bold" />
@@ -63,6 +63,7 @@ scale 包负责多设备秤数据的采集、发现、传输与聚合，采用 *
 |---|---|---|
 | `seasoning_added` | 子设备 → 主设备 | 某秤检测到调料添加，携带 `delta`（重量变化量） |
 | `seasoning_config` | 主设备 → 子设备 | 调料槽位配置同步，携带 `List<SlotConfig>` |
+| `clear_data` | 主设备 → 子设备 | 通知所有子设备清除本机全部测试数据并重置 SP 状态 |

 ### ScaleCommand
 主设备向子设备发送的**控制指令**，子设备收到后校验 `deviceId` 是否匹配自身再执行。
@@ -112,9 +113,10 @@ scale 包负责多设备秤数据的采集、发现、传输与聚合，采用 *

 **职责：**
 - 监听本机 `WeightUtil` 回调，将秤数据实时推送给所有已连接客户端（节流 100ms）
- 新客户端连接时，立即推送所有秤的最新快照（`latestData` 缓存）
- 接收主设备下发的 `ScaleCommand`（清零）和 `ScaleEvent`（配置同步）
+- 新客户端连接时，立即推送所有秤的最新快照（`latestData` 缓存）；若 `latestData` 为空（刚重启尚无读数），延迟 3 秒后补推一次
+- 接收主设备下发的 `ScaleCommand`（清零）和 `ScaleEvent`（配置同步、清除数据）
 - 通过 `connectionCount` 计数器跟踪连接数，避免 `connections` 集合竞态问题
+- 内置单线程 `scheduler`，用于延迟补推任务调度；`stop()` 时同步关闭

 ### ScaleWebSocketClient（仅主设备运行）
 管理主设备与多台子设备的 WebSocket 长连接。
@@ -125,6 +127,10 @@ scale 包负责多设备秤数据的采集、发现、传输与聚合，采用 *
 - 向指定设备或全部设备发送指令/事件
 - 子设备首次连接成功时触发 `onDeviceConnected`，供主设备推送全量配置

+**并发安全机制：**
+- **版本号防重复连接**：每次调用 `connect()` 时递增 `connectVersions[deviceId]`，`doConnect` 执行前校验版本号，版本不匹配（说明已有更新的连接请求）则直接放弃，避免 UDP 触发的新连接与指数退避重连任务并发建立两条连接
+- **原子移除防误删**：`onFailure`/`onClosed` 使用 `connections.remove(deviceId, webSocket)` 原子操作，只有移除的是自己的实例时才触发 `onDeviceDisconnected` 和重连，避免旧连接超时回调误删新连接引用
+
 ---

 ## 数据聚合（1个）
@@ -172,12 +178,18 @@ ScaleServiceManager.sendTare(deviceId, address)
 // 主设备广播调料配置
 ScaleServiceManager.sendSeasoningConfig(slots)

+// 主设备通知所有子设备清除数据
+ScaleServiceManager.sendClearData()
+
 // 主设备监听子设备秤事件（如调料添加）
 ScaleServiceManager.onScaleEvent = { event -> ... }

 // 子设备监听主设备下发的调料配置同步
 ScaleServiceManager.onSeasoningConfig = { event -> ... }

+// 子设备监听主设备下发的清除数据指令
+ScaleServiceManager.onClearData = { ... }
+
 // 子设备向主设备广播秤事件
 ScaleServiceManager.broadcastEvent(event)