refactor(scale): 修复线程安全、生命周期与连接管理问题
- ScaleWebSocketServer：三个状态 Map 改为 ConcurrentHashMap，connectionCount 改为 AtomicInteger - ScaleServiceManager：新增 serviceScope 统一管理协程生命周期，新增 @Volatile isStarted 防重复启动 - ScaleWebSocketClient：doConnect 写入新连接前先 close() 旧连接，防止 TCP 资源泄漏 - MdnsDiscoveryManager：移除 isResolving 上多余的 @Volatile 注解 - MasterScaleActivity：全限定类名改为 import + 短类名 - WeightUtil：删除 WeightListenerImpl 中注释掉的调试代码 - scale包架构分析.md：同步更新架构文档，已修复问题压缩为汇总表 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 4.6
@@ -28,7 +28,6 @@ class MdnsDiscoveryManager(private val context: Context) {
|
||||
|
||||
/** 待解析队列，串行处理避免并发解析失败 */
|
||||
private val resolveQueue = LinkedList<NsdServiceInfo>()
|
||||
@Volatile
|
||||
private var isResolving = false
|
||||
|
||||
/**
|
||||
|
||||
@@ -9,6 +9,8 @@ import com.shuwei.dish.match.utils.NetworkUtil
|
||||
import com.shuwei.dish.match.utils.WeightUtil
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.cancel
|
||||
import kotlinx.coroutines.flow.StateFlow
|
||||
import kotlinx.coroutines.launch
|
||||
|
||||
@@ -34,6 +36,9 @@ object ScaleServiceManager {
|
||||
|
||||
private const val TAG = "ScaleServiceManager"
|
||||
|
||||
@Volatile private var isStarted = false
|
||||
private var serviceScope: CoroutineScope? = null
|
||||
|
||||
private var mdnsRegister: MdnsRegisterManager? = null
|
||||
private var wsServer: ScaleWebSocketServer? = null
|
||||
private var udpSender: UdpBroadcastSender? = null
|
||||
@@ -135,6 +140,9 @@ object ScaleServiceManager {
|
||||
* 启动所有秤服务，在 Application.onCreate() 中调用
|
||||
*/
|
||||
fun start(context: Context) {
|
||||
if (isStarted) return
|
||||
isStarted = true
|
||||
serviceScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
|
||||
val deviceId = GlobalData.deviceId
|
||||
val role = GlobalData.deviceRole
|
||||
Log.d(TAG, "启动秤服务, deviceId=$deviceId, role=$role")
|
||||
@@ -180,14 +188,14 @@ object ScaleServiceManager {
|
||||
}
|
||||
it.onDeviceConnected = { remoteId ->
|
||||
// 子设备连接成功时，从 Room 读取全量配置并单独推送给该设备
|
||||
CoroutineScope(Dispatchers.IO).launch {
|
||||
val slots = BaseApp.instance!!.database.seasoningSlotDao()
|
||||
.queryAll()
|
||||
.map { slot ->
|
||||
serviceScope?.launch {
|
||||
val slots = BaseApp.instance?.database?.seasoningSlotDao()
|
||||
?.queryAll()
|
||||
?.map { slot ->
|
||||
ScaleEvent.SlotConfig(
|
||||
slot.deviceId, slot.address, slot.goodsId, slot.goodsName
|
||||
)
|
||||
}
|
||||
} ?: return@launch
|
||||
if (slots.isNotEmpty()) sendSeasoningConfigTo(remoteId, slots)
|
||||
}
|
||||
}
|
||||
@@ -246,6 +254,9 @@ object ScaleServiceManager {
|
||||
* 停止所有秤服务，在 Application.onTerminate() 或退出时调用
|
||||
*/
|
||||
fun stop() {
|
||||
serviceScope?.cancel()
|
||||
serviceScope = null
|
||||
isStarted = false
|
||||
mdnsDiscovery?.stopDiscovery()
|
||||
udpReceiver?.stop()
|
||||
udpSender?.stop()
|
||||
|
||||
@@ -138,7 +138,8 @@ class ScaleWebSocketClient {
|
||||
}
|
||||
})
|
||||
|
||||
connections[deviceId] = ws
|
||||
// 写入新连接前先关闭旧连接，避免 IP 变化时旧 ws 对象悬空造成 TCP 资源泄漏
|
||||
connections.put(deviceId, ws)?.close(1000, "新连接替换")
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -9,9 +9,11 @@ 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.ConcurrentHashMap
|
||||
import java.util.concurrent.Executors
|
||||
import java.util.concurrent.ScheduledFuture
|
||||
import java.util.concurrent.TimeUnit
|
||||
import java.util.concurrent.atomic.AtomicInteger
|
||||
|
||||
/**
|
||||
* 秤数据 WebSocket 服务端（主设备和子设备均运行）
|
||||
@@ -46,13 +48,13 @@ class ScaleWebSocketServer(private val deviceId: String, private val context: Co
|
||||
get() = server?.connections?.isNotEmpty() == true
|
||||
|
||||
/** 每个秤地址的最近一次推送时间，用于节流 */
|
||||
private val lastPushTime = mutableMapOf<Int, Long>()
|
||||
private val lastPushTime = ConcurrentHashMap<Int, Long>()
|
||||
|
||||
/** 缓存各秤最新数据，新客户端连接时立即同步当前状态 */
|
||||
private val latestData = mutableMapOf<Int, ScaleData>()
|
||||
private val latestData = ConcurrentHashMap<Int, ScaleData>()
|
||||
|
||||
/** 节流窗口内的延迟推送任务，key=address；保证窗口结束时最新数据被推送 */
|
||||
private val pendingPushTasks = mutableMapOf<Int, ScheduledFuture<*>>()
|
||||
private val pendingPushTasks = ConcurrentHashMap<Int, ScheduledFuture<*>>()
|
||||
|
||||
/**
|
||||
* 启动 WebSocket 服务器，并注册 WeightUtil 回调
|
||||
@@ -67,7 +69,7 @@ class ScaleWebSocketServer(private val deviceId: String, private val context: Co
|
||||
WeightUtil.addWeightListener(TAG) { address, state, weight ->
|
||||
Log.d(TAG, "WebSocket 设备${deviceId}推送数据, address=$address, state=$state, weight=$weight")
|
||||
val now = System.currentTimeMillis()
|
||||
// 方案三：数据未变化时直接跳过，避免无效推送
|
||||
// 数据未变化时直接跳过，避免无效推送
|
||||
val last = latestData[address]
|
||||
if (last?.weight == weight && last.state == state) return@addWeightListener
|
||||
|
||||
@@ -89,7 +91,7 @@ class ScaleWebSocketServer(private val deviceId: String, private val context: Co
|
||||
pendingPushTasks.remove(address)?.cancel(false)
|
||||
broadcast(gson.toJson(data))
|
||||
} else if (pendingPushTasks[address] == null) {
|
||||
// 方案一：窗口内首次触发时安排一个 trailing 推送，保证窗口末尾推最新值
|
||||
// 窗口内首次触发：安排 trailing 推送，保证窗口末尾推最新值
|
||||
val delay = THROTTLE_MS - (now - lastPush)
|
||||
if (!scheduler.isShutdown) {
|
||||
pendingPushTasks[address] = scheduler.schedule({
|
||||
@@ -156,10 +158,10 @@ class ScaleWebSocketServer(private val deviceId: String, private val context: Co
|
||||
WebSocketServer(InetSocketAddress(port)) {
|
||||
|
||||
/** 当前活跃连接计数，用于替代 connections.isNullOrEmpty() 避免竞态 */
|
||||
private var connectionCount = 0
|
||||
private val connectionCount = AtomicInteger(0)
|
||||
|
||||
override fun onOpen(conn: WebSocket, handshake: ClientHandshake) {
|
||||
connectionCount++
|
||||
connectionCount.incrementAndGet()
|
||||
Log.d(TAG, "主设备已连接: ${conn.remoteSocketAddress}")
|
||||
onConnectionChanged?.invoke(true)
|
||||
|
||||
@@ -194,8 +196,8 @@ class ScaleWebSocketServer(private val deviceId: String, private val context: Co
|
||||
override fun onClose(conn: WebSocket, code: Int, reason: String, remote: Boolean) {
|
||||
Log.d(TAG, "主设备已断开: ${conn.remoteSocketAddress}, reason=$reason")
|
||||
// 用计数器判断是否还有活跃连接，避免依赖 connections 集合的移除时机导致竞态
|
||||
if (--connectionCount <= 0) {
|
||||
connectionCount = 0
|
||||
if (connectionCount.decrementAndGet() <= 0) {
|
||||
connectionCount.set(0)
|
||||
onConnectionChanged?.invoke(false)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -19,6 +19,7 @@ import com.shuwei.dish.match.R
|
||||
import com.shuwei.dish.match.adapter.Seasoning18GridAdapter
|
||||
import com.shuwei.dish.match.adapter.Seasoning22GridAdapter
|
||||
import com.shuwei.dish.match.base.BaseActivity
|
||||
import com.shuwei.dish.match.base.GlobalData
|
||||
import com.shuwei.dish.match.dialog.CommonDialog
|
||||
import com.shuwei.dish.match.databinding.ActivityMasterScaleBinding
|
||||
import com.shuwei.dish.match.databinding.ListTypeScale1Binding
|
||||
@@ -99,7 +100,7 @@ class MasterScaleActivity : BaseActivity() {
|
||||
val flow = ScaleServiceManager.allScales ?: return
|
||||
lifecycleScope.launch {
|
||||
flow.collectLatest { scaleMap ->
|
||||
val localId = com.shuwei.dish.match.base.GlobalData.deviceId
|
||||
val localId = GlobalData.deviceId
|
||||
val localIp = NetworkUtil.getLocalIpAddress(this@MasterScaleActivity)
|
||||
val groups = scaleMap.values
|
||||
.groupBy { it.deviceId }
|
||||
|
||||
@@ -140,14 +140,6 @@ open class WeightListenerImpl : Weigher2.Listener {
|
||||
}
|
||||
|
||||
override fun onGetWeight(address: Int, state: Int, weight: Double) {
|
||||
// val stateStr = when (state) {
|
||||
// SensorScale.STATE_STABLE -> "稳定"
|
||||
// SensorScale.STATE_UNSTABLE -> "不稳定"
|
||||
// SensorScale.STATE_OVER_WEIGHT -> "量程溢出"
|
||||
// else -> ""
|
||||
// }
|
||||
// val result1 = String.format("$stateStr 重量：%s kg", weight / 1000f)
|
||||
// Log.d("WeightUtil", "ttlReturn, address:$address,$result1")
|
||||
}
|
||||
|
||||
override fun onReadIdentify(rate: Int) {
|
||||
|
||||
+44
-3
@@ -1,5 +1,7 @@
|
||||
# scale 包架构分析
|
||||
|
||||
> 最后更新：2026-05-25（同步本次优化修复内容）
|
||||
|
||||
## 概览
|
||||
|
||||
scale 包负责多设备秤数据的采集、发现、传输与聚合，采用 **mDNS + UDP 双路冗余发现 + WebSocket 长连接**的架构。
|
||||
@@ -115,9 +117,16 @@ scale 包负责多设备秤数据的采集、发现、传输与聚合，采用 *
|
||||
- 监听本机 `WeightUtil` 回调，将秤数据实时推送给所有已连接客户端（节流 100ms）
|
||||
- 新客户端连接时，立即推送所有秤的最新快照（`latestData` 缓存）；若 `latestData` 为空（刚重启尚无读数），延迟 3 秒后补推一次
|
||||
- 接收主设备下发的 `ScaleCommand`（清零）和 `ScaleEvent`（配置同步、清除数据）
|
||||
- 通过 `connectionCount` 计数器跟踪连接数，避免 `connections` 集合竞态问题
|
||||
- `latestData`/`lastPushTime`/`pendingPushTasks` 均使用 `ConcurrentHashMap`，保证 WeightUtil 回调线程与 java-websocket 服务端线程并发安全
|
||||
- 通过 `connectionCount`（`AtomicInteger`）原子跟踪连接数，避免 `connections` 集合竞态问题
|
||||
- 内置单线程 `scheduler`，用于延迟补推任务调度；`stop()` 时同步关闭
|
||||
|
||||
**节流策略（leading + trailing）：**
|
||||
1. 窗口已过（距上次推送 ≥ 100ms）：立即推送，取消已有的 trailing 任务
|
||||
2. 窗口内首次触发：安排一个 trailing 延迟任务，保证窗口末尾推最新值
|
||||
3. 窗口内再次触发：仅更新 `latestData`，等 trailing 任务触发时推送
|
||||
4. 数据未变化（weight 和 state 均相同）：直接跳过，不进入节流逻辑
|
||||
|
||||
### ScaleWebSocketClient（仅主设备运行）
|
||||
管理主设备与多台子设备的 WebSocket 长连接。
|
||||
|
||||
@@ -130,6 +139,7 @@ scale 包负责多设备秤数据的采集、发现、传输与聚合，采用 *
|
||||
**并发安全机制：**
|
||||
- **版本号防重复连接**：每次调用 `connect()` 时递增 `connectVersions[deviceId]`，`doConnect` 执行前校验版本号，版本不匹配（说明已有更新的连接请求）则直接放弃，避免 UDP 触发的新连接与指数退避重连任务并发建立两条连接
|
||||
- **原子移除防误删**：`onFailure`/`onClosed` 使用 `connections.remove(deviceId, webSocket)` 原子操作，只有移除的是自己的实例时才触发 `onDeviceDisconnected` 和重连，避免旧连接超时回调误删新连接引用
|
||||
- **旧连接显式关闭**：`doConnect` 末尾使用 `connections.put(deviceId, ws)?.close(1000, "新连接替换")`，IP 变化重连时先关闭旧连接，防止旧 ws 对象悬空造成 TCP 资源泄漏
|
||||
|
||||
---
|
||||
|
||||
@@ -153,8 +163,8 @@ scale 包负责多设备秤数据的采集、发现、传输与聚合，采用 *
|
||||
| 常量 | 说明 |
|
||||
|---|---|
|
||||
| `DEVICE_ID_2/22/18/1` | 各子设备固定 UUID |
|
||||
| `SCALE_ORDER_22` | 22个秤的物理位置排列顺序 |
|
||||
| `SCALE_ORDER_18` | 18个秤的物理位置排列顺序 |
|
||||
| `SCALE_ORDER_22` | 22个秤的物理位置排列顺序（地址范围 1-22） |
|
||||
| `SCALE_ORDER_18` | 18个秤的物理位置排列顺序（地址范围 1-18） |
|
||||
| `DEVICE_ORDER` | 设备在列表中的显示顺序 |
|
||||
|
||||
---
|
||||
@@ -164,6 +174,9 @@ scale 包负责多设备秤数据的采集、发现、传输与聚合，采用 *
|
||||
### ScaleServiceManager
|
||||
整个包的统一入口（单例），根据设备角色决定启动哪些服务，并将各组件串联起来。
|
||||
|
||||
- **防重复启动**：内置 `@Volatile isStarted` 标志，`start()` 入口加守卫，重复调用直接返回，防止端口冲突和资源泄漏
|
||||
- **协程生命周期管理**：维护 `serviceScope`（`Dispatchers.IO + SupervisorJob()`），所有内部异步操作均在此 Scope 内执行，`stop()` 时统一 `cancel()`，防止 NPE 和内存泄漏
|
||||
|
||||
**外部使用方式：**
|
||||
```kotlin
|
||||
// Application.onCreate()
|
||||
@@ -204,3 +217,31 @@ ScaleServiceManager.stop()
|
||||
```
|
||||
|
||||
外部代码只需与 `ScaleServiceManager` 交互，无需感知内部任何组件。
|
||||
|
||||
---
|
||||
|
||||
## 问题排查与优化建议
|
||||
|
||||
### ✅ 已修复问题（2026-05-25）
|
||||
|
||||
| 严重度 | 问题 | 位置 | 修复方式 |
|
||||
|---|---|---|---|
|
||||
| 🔴 严重 | 三个状态 Map 非线程安全 | `ScaleWebSocketServer.kt` | 改为 `ConcurrentHashMap` |
|
||||
| 🔴 严重 | `connectionCount` 非原子操作 | `ScaleWebSocketServer.kt` `InternalServer` | 改为 `AtomicInteger` |
|
||||
| 🟡 中等 | `CoroutineScope` 无生命周期绑定 | `ScaleServiceManager.kt` | 改为 `serviceScope`，`stop()` 时统一 cancel |
|
||||
| 🟡 中等 | `start()` 缺少防重复启动保护 | `ScaleServiceManager.kt` | 新增 `@Volatile isStarted` 守卫 |
|
||||
| 🟡 中等 | `doConnect` 旧连接未显式关闭 | `ScaleWebSocketClient.kt` | 改为 `connections.put()?.close()` |
|
||||
| 🟡 中等 | `@Volatile` 注解多余 | `MdnsDiscoveryManager.kt` | 移除 `@Volatile` |
|
||||
| 🟡 中等 | 全限定类名违反导入规范 | `MasterScaleActivity.kt` | 改为 import + 短类名 |
|
||||
| 🟢 轻微 | `onGetWeight` 残留注释掉的调试代码 | `WeightUtil.kt` | 删除注释代码块 |
|
||||
| 🟢 轻微 | "方案一"/"方案三"草稿注释 | `ScaleWebSocketServer.kt` | 改为描述行为意图的正式注释 |
|
||||
|
||||
---
|
||||
|
||||
### 🟢 轻微：ScaleDataAggregator 高频更新时 GC 压力（待观察）
|
||||
|
||||
**位置**：`ScaleDataAggregator.kt`，第 120 行
|
||||
|
||||
**问题**：每次秤数据更新都执行 `_allScales.value = HashMap(cache)`，在多台设备同时高频推送数据时（如 22 个秤 × 10Hz），每秒可能创建数百个临时 HashMap 对象，增加 GC 压力。
|
||||
|
||||
**建议**：这是 StateFlow 的惯用写法，短期内无需优化。若未来出现 GC 卡顿，可考虑引入防抖（debounce）合并多次更新后再发布快照。
|
||||
|
||||
Reference in New Issue
Block a user