From 994bcf5bc2a41908f2afe46395a0f5a0a16ce714 Mon Sep 17 00:00:00 2001 From: mazengfei <331023091@qq.com> Date: Wed, 2 Sep 2026 15:12:23 +0800 Subject: [PATCH] =?UTF-8?q?feat(face):=20=E4=BC=98=E5=8C=96=E4=BA=BA?= =?UTF-8?q?=E8=84=B8=E5=BA=93=E6=9B=B4=E6=96=B0=E5=90=8C=E6=AD=A5=E5=8F=8A?= =?UTF-8?q?=E8=AF=86=E5=88=AB=E5=BC=95=E6=93=8E=E5=88=B7=E6=96=B0=E6=9C=BA?= =?UTF-8?q?=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 引入进程级互斥锁 FaceSyncLock，串行化 MQTT 实时更新与 HTTP 增量轮询人脸库写操作，避免重复入库 - 优化 MQTT 实时消息处理逻辑，批内去重并加入乱序守卫，确保数据正确性与一致性 - 将识别引擎内存刷新调度防抖至 scheduleFaceRefresh，合并短时间内多次变更减少资源消耗 - 在 BaseActivity 中实现防抖刷新机制，统一 MQTT、HTTP增量和定时轮询三路触发入口 - 移除应用启动时人脸 MQTT 订阅初始化，改为首次全量同步完成后启动，防止并发竞态 - LoginByFaceActivity 和 DeviceInitActivity 适配新增机制，优化人脸数据同步及识别引擎刷新流程 --- .../main/java/com/sw/platecabinet/MyApp.kt | 4 - .../sw/platecabinet/activity/BaseActivity.kt | 17 +++- .../activity/DeviceInitActivity.kt | 4 + .../activity/LoginByFaceActivity.kt | 11 +-- .../platecabinet/mqtt/FaceMqttSubscriber.kt | 90 +++++++++++-------- .../com/sw/platecabinet/mqtt/FaceSyncLock.kt | 14 +++ .../platecabinet/viewmodel/NetViewModelV2.kt | 5 +- 7 files changed, 94 insertions(+), 51 deletions(-) create mode 100644 app/src/main/java/com/sw/platecabinet/mqtt/FaceSyncLock.kt diff --git a/app/src/main/java/com/sw/platecabinet/MyApp.kt b/app/src/main/java/com/sw/platecabinet/MyApp.kt index 321ebc5..3a6718a 100644 --- a/app/src/main/java/com/sw/platecabinet/MyApp.kt +++ b/app/src/main/java/com/sw/platecabinet/MyApp.kt @@ -3,7 +3,6 @@ package com.sw.platecabinet import android.util.Log import com.sw.plate.App import com.sw.plate.utils.AppUtil -import com.sw.platecabinet.mqtt.FaceMqttSubscriber import com.sw.platecabinet.utils.CrashHandler import com.sw.platecabinet.utils.SpTool import timber.log.Timber @@ -21,9 +20,6 @@ class MyApp : App() { // 初始化崩溃处理器 CrashHandler.init(this) - - // 启动人脸 MQTT 实时订阅（进程级单例，应用生命周期内保持连接） - FaceMqttSubscriber.start() } /** diff --git a/app/src/main/java/com/sw/platecabinet/activity/BaseActivity.kt b/app/src/main/java/com/sw/platecabinet/activity/BaseActivity.kt index 560e2b9..21e9576 100644 --- a/app/src/main/java/com/sw/platecabinet/activity/BaseActivity.kt +++ b/app/src/main/java/com/sw/platecabinet/activity/BaseActivity.kt @@ -4,6 +4,8 @@ import android.app.Dialog import android.content.Context import android.os.Build import android.os.Bundle +import android.os.Handler +import android.os.Looper import android.text.TextUtils import android.view.KeyEvent import android.view.View @@ -399,6 +401,19 @@ abstract class BaseActivity : AppCompatActivity() { val recognizeViewModel by viewModels() + /** + * 引擎内存刷新防抖：合并短时间内连续的人脸变更（MQTT 实时 + HTTP 增量补拉 + 定时轮询）， + * 避免多次全量重载 ArcSoft 引擎内存（removeFaceFeature(-1) + registerFaceFeature 非原子）。 + */ + private val faceRefreshHandler = Handler(Looper.getMainLooper()) + private val faceRefreshRunnable = Runnable { recognizeViewModel.refreshFaceList() } + + /** 防抖调度引擎内存刷新（2s 内连续变更合并为一次） */ + protected fun scheduleFaceRefresh() { + faceRefreshHandler.removeCallbacks(faceRefreshRunnable) + faceRefreshHandler.postDelayed(faceRefreshRunnable, 2000) + } + fun startFaceTask() { faceTaskJob = intervalExecutor.startIntervalTaskWithInitialDelay(initialDelay, dealyMillis) { @@ -418,7 +433,7 @@ abstract class BaseActivity : AppCompatActivity() { pageNo = pageNo, timestamp = timestamp ) { - recognizeViewModel.refreshFaceList() + scheduleFaceRefresh() } } diff --git a/app/src/main/java/com/sw/platecabinet/activity/DeviceInitActivity.kt b/app/src/main/java/com/sw/platecabinet/activity/DeviceInitActivity.kt index b56f864..48e11e8 100644 --- a/app/src/main/java/com/sw/platecabinet/activity/DeviceInitActivity.kt +++ b/app/src/main/java/com/sw/platecabinet/activity/DeviceInitActivity.kt @@ -5,6 +5,7 @@ import com.sw.plate.utils.ToastUtils import com.sw.platecabinet.GlobalData import com.sw.platecabinet.member.databinding.ActivityDeviceInitBinding import com.sw.platecabinet.member.databinding.ItemTitleTimeBinding +import com.sw.platecabinet.mqtt.FaceMqttSubscriber import com.sw.platecabinet.utils.SpTool import timber.log.Timber @@ -76,6 +77,9 @@ class DeviceInitActivity : BaseActivity() { } private fun goLoginActivity() { + // 首次全量同步（如需）已完成，此时启动人脸 MQTT 实时订阅， + // 避免与首次全量同步的 clearFaceData 产生并发写竞态 + FaceMqttSubscriber.start() val intent = Intent(this, LoginByFaceActivity::class.java) startActivity(intent) finish() diff --git a/app/src/main/java/com/sw/platecabinet/activity/LoginByFaceActivity.kt b/app/src/main/java/com/sw/platecabinet/activity/LoginByFaceActivity.kt index e93a2f7..8b7c5ed 100644 --- a/app/src/main/java/com/sw/platecabinet/activity/LoginByFaceActivity.kt +++ b/app/src/main/java/com/sw/platecabinet/activity/LoginByFaceActivity.kt @@ -146,16 +146,13 @@ class LoginByFaceActivity : BaseActivity(), } /** - * MQTT 实时同步后的人脸库已落库，防抖重载识别引擎内存（合并短时间内连续变更） + * MQTT 实时同步后的人脸库已落库，防抖重载识别引擎内存（合并短时间内连续变更）。 + * scheduleFaceRefresh 定义于 BaseActivity，统一 MQTT/HTTP 增量/定时轮询三路刷新入口。 */ - private val faceRefreshHandler = Handler(Looper.getMainLooper()) - private val faceRefreshRunnable = Runnable { recognizeViewModel.refreshFaceList() } - @Subscribe(threadMode = ThreadMode.MAIN) fun onFaceChanged(event: FaceChangedEvent) { Timber.d("onFaceChanged 收到人脸实时变更，2s 后刷新引擎内存") - faceRefreshHandler.removeCallbacks(faceRefreshRunnable) - faceRefreshHandler.postDelayed(faceRefreshRunnable, 2000) + scheduleFaceRefresh() } /** @@ -170,7 +167,7 @@ class LoginByFaceActivity : BaseActivity(), } Timber.d("onFaceSyncTrigger MQTT 已连接，执行 HTTP 增量补拉 timestamp=$timestamp") netViewModelV2.getFaceIncrementList(pageNo = 1, timestamp = timestamp) { - recognizeViewModel.refreshFaceList() + scheduleFaceRefresh() } } diff --git a/app/src/main/java/com/sw/platecabinet/mqtt/FaceMqttSubscriber.kt b/app/src/main/java/com/sw/platecabinet/mqtt/FaceMqttSubscriber.kt index c4fd673..088b562 100644 --- a/app/src/main/java/com/sw/platecabinet/mqtt/FaceMqttSubscriber.kt +++ b/app/src/main/java/com/sw/platecabinet/mqtt/FaceMqttSubscriber.kt @@ -1,6 +1,7 @@ package com.sw.platecabinet.mqtt import com.google.gson.JsonParser +import com.sw.plate.utils.Base64 import com.sw.plate.utils.arcface.FaceApi import com.sw.plate.utils.arcface.facedb.entity.FaceEntity import com.sw.platecabinet.GlobalData @@ -10,6 +11,7 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.withLock import org.greenrobot.eventbus.EventBus import timber.log.Timber @@ -150,7 +152,7 @@ object FaceMqttSubscriber { } /** 解析广播 JSON 数组并落库（faceUpdateTimestamp 兼容 long/String 两种格式） */ - private fun handleFaceUpdate(payload: ByteArray) { + private suspend fun handleFaceUpdate(payload: ByteArray) { // 联调排查：记录原始报文预览（特征码很长，只打前 120 字符） val raw = String(payload, Charsets.UTF_8) Timber.i("$TAG 收到人脸广播: size=${payload.size} 预览=${raw.take(120)}") @@ -206,55 +208,67 @@ object FaceMqttSubscriber { * 按 userFaceId 查本地记录，消息时间戳不大于本地时间戳的条目直接丢弃， * 防止 QoS1 重复投递或乱序到达时旧数据覆盖新数据。 * + * 与 HTTP 增量轮询通过 [FaceSyncLock] 串行化写库，避免同一 userFaceId 重复入库。 + * * 注意：本方法不推进增量水位。水位只由 HTTP 增量接口的响应推进， * 后台推送失败只记日志不重发，若 MQ 消息把水位推到漏发变更之后， * 轮询/补拉将永远拉不到那条变更。 */ - private fun applyRealtimeUpdates(items: List) { + private suspend fun applyRealtimeUpdates(items: List) { if (items.isEmpty()) return - // 批内去重：同一 userFaceId 仅保留时间戳最大的一条（含其删除标志） - val batch = items.groupBy { it.userFaceId } - .flatMap { (key, list) -> - if (key.isNullOrEmpty()) list - else listOfNotNull(list.maxByOrNull { it.faceUpdateTimestamp ?: 0L }) - } + var changed = false + FaceSyncLock.mutex.withLock { + // 批内去重：同一 userFaceId 仅保留时间戳最大的一条（含其删除标志） + val batch = items.groupBy { it.userFaceId } + .flatMap { (key, list) -> + if (key.isNullOrEmpty()) list + else listOfNotNull(list.maxByOrNull { it.faceUpdateTimestamp ?: 0L }) + } - // 乱序守卫：本地已有同 userFaceId 且时间戳不旧的记录则跳过该条 - val fresh = batch.filter { msg -> - val key = msg.userFaceId - if (key.isNullOrEmpty()) return@filter true - val local = faceApi.queryByUserFaceId(key) - val msgTs = msg.faceUpdateTimestamp ?: 0L - when { - local == null -> true - msgTs <= 0L -> true - msgTs > local.faceUpdateTimestamp -> true - else -> { - Timber.d("$TAG 实时消息乱序/重复，丢弃 userFaceId=$key ts=$msgTs") - false + // 乱序守卫：本地已有同 userFaceId 且时间戳不旧的记录则跳过该条 + val fresh = batch.filter { msg -> + val key = msg.userFaceId + if (key.isNullOrEmpty()) return@filter true + val local = faceApi.queryByUserFaceId(key) + val msgTs = msg.faceUpdateTimestamp ?: 0L + when { + local == null -> true + msgTs <= 0L -> true + msgTs > local.faceUpdateTimestamp -> true + else -> { + Timber.d("$TAG 实时消息乱序/重复，丢弃 userFaceId=$key ts=$msgTs") + false + } } } - } - if (fresh.isEmpty()) return + if (fresh.isEmpty()) return@withLock - var changed = false - for (msg in fresh) { - if (msg.faceDeleted == true) { - val userFaceId = msg.userFaceId - if (!userFaceId.isNullOrEmpty() && faceApi.queryByUserFaceId(userFaceId) != null) { - faceApi.deleteByUserFaceId(userFaceId) + for (msg in fresh) { + if (msg.faceDeleted == true) { + val userFaceId = msg.userFaceId + if (!userFaceId.isNullOrEmpty() && faceApi.queryByUserFaceId(userFaceId) != null) { + faceApi.deleteByUserFaceId(userFaceId) + changed = true + } + } else { + val entity = buildEntity(msg) ?: continue + val userFaceId = msg.userFaceId + if (!userFaceId.isNullOrEmpty()) { + // 优先按 userFaceId 精确判重，避免特征字段不一致导致重复入库 + if (faceApi.queryByUserFaceId(userFaceId) != null) continue + faceApi.insert(entity) + } else { + // userFaceId 为空时回退到特征判重（与 HTTP 增量逻辑对齐） + val featureStr = msg.resolveFeatureStr() + val existList = faceApi.queryAllByUserName(msg.userId) + val alreadyExists = existList.any { e -> + featureStr != null && Base64.encode(e.featureData) == featureStr + } + if (!alreadyExists) faceApi.insert(entity) + } changed = true } - } else { - // 优先按 userFaceId 精确判重，避免特征字段不一致导致重复入库 - val userFaceId = msg.userFaceId - if (!userFaceId.isNullOrEmpty() && faceApi.queryByUserFaceId(userFaceId) != null) { - continue - } - val entity = buildEntity(msg) ?: continue - faceApi.insert(entity) - changed = true } } diff --git a/app/src/main/java/com/sw/platecabinet/mqtt/FaceSyncLock.kt b/app/src/main/java/com/sw/platecabinet/mqtt/FaceSyncLock.kt new file mode 100644 index 0000000..90e491c --- /dev/null +++ b/app/src/main/java/com/sw/platecabinet/mqtt/FaceSyncLock.kt @@ -0,0 +1,14 @@ +package com.sw.platecabinet.mqtt + +import kotlinx.coroutines.sync.Mutex + +/** + * 本地人脸库写操作的进程级互斥锁。 + * + * MQTT 实时更新（[FaceMqttSubscriber]）与 HTTP 增量轮询（NetViewModelV2.getFaceIncrementList） + * 会并发写本地人脸库。二者对同一 userFaceId 的「判重 → 插入」不是原子操作， + * 并发时会产生重复记录；共享此锁将两路写入串行化。 + */ +object FaceSyncLock { + val mutex = Mutex() +} diff --git a/app/src/main/java/com/sw/platecabinet/viewmodel/NetViewModelV2.kt b/app/src/main/java/com/sw/platecabinet/viewmodel/NetViewModelV2.kt index 4e3fea4..4167735 100644 --- a/app/src/main/java/com/sw/platecabinet/viewmodel/NetViewModelV2.kt +++ b/app/src/main/java/com/sw/platecabinet/viewmodel/NetViewModelV2.kt @@ -15,11 +15,13 @@ import com.sw.platecabinet.GlobalKey import com.sw.platecabinet.model.DeviceConfigV2 import com.sw.platecabinet.model.response.UserFaceModel import com.sw.platecabinet.model.response.UserFaceModelV2 +import com.sw.platecabinet.mqtt.FaceSyncLock import com.sw.platecabinet.network.ApiClient import com.sw.platecabinet.repository.RemoteRepositoryV2 import com.sw.platecabinet.utils.SpTool import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withContext import timber.log.Timber import java.io.File @@ -190,7 +192,8 @@ class NetViewModelV2 : ViewModel() { } withContext(Dispatchers.IO) { val list = response.data ?: emptyList() - updateFaceData(list) + // 与 MQTT 实时更新串行化写库，避免同一 userFaceId 重复入库 + FaceSyncLock.mutex.withLock { updateFaceData(list) } // 取当前页最大时间戳，与已累积的比较取最大值 val pageMaxTimestamp = maxFaceTimestamp(list) if (pageMaxTimestamp > faceTimestamp) {