feat(face): 优化人脸库更新同步及识别引擎刷新机制

- 引入进程级互斥锁 FaceSyncLock,串行化 MQTT 实时更新与 HTTP 增量轮询人脸库写操作,避免重复入库
- 优化 MQTT 实时消息处理逻辑,批内去重并加入乱序守卫,确保数据正确性与一致性
- 将识别引擎内存刷新调度防抖至 scheduleFaceRefresh,合并短时间内多次变更减少资源消耗
- 在 BaseActivity 中实现防抖刷新机制,统一 MQTT、HTTP增量和定时轮询三路触发入口
- 移除应用启动时人脸 MQTT 订阅初始化,改为首次全量同步完成后启动,防止并发竞态
- LoginByFaceActivity 和 DeviceInitActivity 适配新增机制,优化人脸数据同步及识别引擎刷新流程
This commit is contained in:
mazengfei
2026-09-02 15:12:23 +08:00
parent 1bd57e0680
commit 994bcf5bc2
7 changed files with 94 additions and 51 deletions
@@ -3,7 +3,6 @@ package com.sw.platecabinet
import android.util.Log import android.util.Log
import com.sw.plate.App import com.sw.plate.App
import com.sw.plate.utils.AppUtil import com.sw.plate.utils.AppUtil
import com.sw.platecabinet.mqtt.FaceMqttSubscriber
import com.sw.platecabinet.utils.CrashHandler import com.sw.platecabinet.utils.CrashHandler
import com.sw.platecabinet.utils.SpTool import com.sw.platecabinet.utils.SpTool
import timber.log.Timber import timber.log.Timber
@@ -21,9 +20,6 @@ class MyApp : App() {
// 初始化崩溃处理器 // 初始化崩溃处理器
CrashHandler.init(this) CrashHandler.init(this)
// 启动人脸 MQTT 实时订阅(进程级单例,应用生命周期内保持连接)
FaceMqttSubscriber.start()
} }
/** /**
@@ -4,6 +4,8 @@ import android.app.Dialog
import android.content.Context import android.content.Context
import android.os.Build import android.os.Build
import android.os.Bundle import android.os.Bundle
import android.os.Handler
import android.os.Looper
import android.text.TextUtils import android.text.TextUtils
import android.view.KeyEvent import android.view.KeyEvent
import android.view.View import android.view.View
@@ -399,6 +401,19 @@ abstract class BaseActivity<VB : ViewBinding> : AppCompatActivity() {
val recognizeViewModel by viewModels<RecognizeViewModel>() val recognizeViewModel by viewModels<RecognizeViewModel>()
/**
* 引擎内存刷新防抖:合并短时间内连续的人脸变更(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() { fun startFaceTask() {
faceTaskJob = faceTaskJob =
intervalExecutor.startIntervalTaskWithInitialDelay(initialDelay, dealyMillis) { intervalExecutor.startIntervalTaskWithInitialDelay(initialDelay, dealyMillis) {
@@ -418,7 +433,7 @@ abstract class BaseActivity<VB : ViewBinding> : AppCompatActivity() {
pageNo = pageNo, pageNo = pageNo,
timestamp = timestamp timestamp = timestamp
) { ) {
recognizeViewModel.refreshFaceList() scheduleFaceRefresh()
} }
} }
@@ -5,6 +5,7 @@ import com.sw.plate.utils.ToastUtils
import com.sw.platecabinet.GlobalData import com.sw.platecabinet.GlobalData
import com.sw.platecabinet.member.databinding.ActivityDeviceInitBinding import com.sw.platecabinet.member.databinding.ActivityDeviceInitBinding
import com.sw.platecabinet.member.databinding.ItemTitleTimeBinding import com.sw.platecabinet.member.databinding.ItemTitleTimeBinding
import com.sw.platecabinet.mqtt.FaceMqttSubscriber
import com.sw.platecabinet.utils.SpTool import com.sw.platecabinet.utils.SpTool
import timber.log.Timber import timber.log.Timber
@@ -76,6 +77,9 @@ class DeviceInitActivity : BaseActivity<ActivityDeviceInitBinding>() {
} }
private fun goLoginActivity() { private fun goLoginActivity() {
// 首次全量同步(如需)已完成,此时启动人脸 MQTT 实时订阅,
// 避免与首次全量同步的 clearFaceData 产生并发写竞态
FaceMqttSubscriber.start()
val intent = Intent(this, LoginByFaceActivity::class.java) val intent = Intent(this, LoginByFaceActivity::class.java)
startActivity(intent) startActivity(intent)
finish() finish()
@@ -146,16 +146,13 @@ class LoginByFaceActivity : BaseActivity<ActivityLoginFaceBinding>(),
} }
/** /**
* MQTT 实时同步后的人脸库已落库,防抖重载识别引擎内存(合并短时间内连续变更) * MQTT 实时同步后的人脸库已落库,防抖重载识别引擎内存(合并短时间内连续变更)
* scheduleFaceRefresh 定义于 BaseActivity,统一 MQTT/HTTP 增量/定时轮询三路刷新入口。
*/ */
private val faceRefreshHandler = Handler(Looper.getMainLooper())
private val faceRefreshRunnable = Runnable { recognizeViewModel.refreshFaceList() }
@Subscribe(threadMode = ThreadMode.MAIN) @Subscribe(threadMode = ThreadMode.MAIN)
fun onFaceChanged(event: FaceChangedEvent) { fun onFaceChanged(event: FaceChangedEvent) {
Timber.d("onFaceChanged 收到人脸实时变更,2s 后刷新引擎内存") Timber.d("onFaceChanged 收到人脸实时变更,2s 后刷新引擎内存")
faceRefreshHandler.removeCallbacks(faceRefreshRunnable) scheduleFaceRefresh()
faceRefreshHandler.postDelayed(faceRefreshRunnable, 2000)
} }
/** /**
@@ -170,7 +167,7 @@ class LoginByFaceActivity : BaseActivity<ActivityLoginFaceBinding>(),
} }
Timber.d("onFaceSyncTrigger MQTT 已连接,执行 HTTP 增量补拉 timestamp=$timestamp") Timber.d("onFaceSyncTrigger MQTT 已连接,执行 HTTP 增量补拉 timestamp=$timestamp")
netViewModelV2.getFaceIncrementList(pageNo = 1, timestamp = timestamp) { netViewModelV2.getFaceIncrementList(pageNo = 1, timestamp = timestamp) {
recognizeViewModel.refreshFaceList() scheduleFaceRefresh()
} }
} }
@@ -1,6 +1,7 @@
package com.sw.platecabinet.mqtt package com.sw.platecabinet.mqtt
import com.google.gson.JsonParser import com.google.gson.JsonParser
import com.sw.plate.utils.Base64
import com.sw.plate.utils.arcface.FaceApi import com.sw.plate.utils.arcface.FaceApi
import com.sw.plate.utils.arcface.facedb.entity.FaceEntity import com.sw.plate.utils.arcface.facedb.entity.FaceEntity
import com.sw.platecabinet.GlobalData import com.sw.platecabinet.GlobalData
@@ -10,6 +11,7 @@ import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.withLock
import org.greenrobot.eventbus.EventBus import org.greenrobot.eventbus.EventBus
import timber.log.Timber import timber.log.Timber
@@ -150,7 +152,7 @@ object FaceMqttSubscriber {
} }
/** 解析广播 JSON 数组并落库(faceUpdateTimestamp 兼容 long/String 两种格式) */ /** 解析广播 JSON 数组并落库(faceUpdateTimestamp 兼容 long/String 两种格式) */
private fun handleFaceUpdate(payload: ByteArray) { private suspend fun handleFaceUpdate(payload: ByteArray) {
// 联调排查:记录原始报文预览(特征码很长,只打前 120 字符) // 联调排查:记录原始报文预览(特征码很长,只打前 120 字符)
val raw = String(payload, Charsets.UTF_8) val raw = String(payload, Charsets.UTF_8)
Timber.i("$TAG 收到人脸广播: size=${payload.size} 预览=${raw.take(120)}") Timber.i("$TAG 收到人脸广播: size=${payload.size} 预览=${raw.take(120)}")
@@ -206,13 +208,17 @@ object FaceMqttSubscriber {
* 按 userFaceId 查本地记录,消息时间戳不大于本地时间戳的条目直接丢弃, * 按 userFaceId 查本地记录,消息时间戳不大于本地时间戳的条目直接丢弃,
* 防止 QoS1 重复投递或乱序到达时旧数据覆盖新数据。 * 防止 QoS1 重复投递或乱序到达时旧数据覆盖新数据。
* *
* 与 HTTP 增量轮询通过 [FaceSyncLock] 串行化写库,避免同一 userFaceId 重复入库。
*
* 注意:本方法不推进增量水位。水位只由 HTTP 增量接口的响应推进, * 注意:本方法不推进增量水位。水位只由 HTTP 增量接口的响应推进,
* 后台推送失败只记日志不重发,若 MQ 消息把水位推到漏发变更之后, * 后台推送失败只记日志不重发,若 MQ 消息把水位推到漏发变更之后,
* 轮询/补拉将永远拉不到那条变更。 * 轮询/补拉将永远拉不到那条变更。
*/ */
private fun applyRealtimeUpdates(items: List<UserFaceModelV2>) { private suspend fun applyRealtimeUpdates(items: List<UserFaceModelV2>) {
if (items.isEmpty()) return if (items.isEmpty()) return
var changed = false
FaceSyncLock.mutex.withLock {
// 批内去重:同一 userFaceId 仅保留时间戳最大的一条(含其删除标志) // 批内去重:同一 userFaceId 仅保留时间戳最大的一条(含其删除标志)
val batch = items.groupBy { it.userFaceId } val batch = items.groupBy { it.userFaceId }
.flatMap { (key, list) -> .flatMap { (key, list) ->
@@ -236,9 +242,8 @@ object FaceMqttSubscriber {
} }
} }
} }
if (fresh.isEmpty()) return if (fresh.isEmpty()) return@withLock
var changed = false
for (msg in fresh) { for (msg in fresh) {
if (msg.faceDeleted == true) { if (msg.faceDeleted == true) {
val userFaceId = msg.userFaceId val userFaceId = msg.userFaceId
@@ -247,16 +252,25 @@ object FaceMqttSubscriber {
changed = true changed = true
} }
} else { } else {
// 优先按 userFaceId 精确判重,避免特征字段不一致导致重复入库
val userFaceId = msg.userFaceId
if (!userFaceId.isNullOrEmpty() && faceApi.queryByUserFaceId(userFaceId) != null) {
continue
}
val entity = buildEntity(msg) ?: continue val entity = buildEntity(msg) ?: continue
val userFaceId = msg.userFaceId
if (!userFaceId.isNullOrEmpty()) {
// 优先按 userFaceId 精确判重,避免特征字段不一致导致重复入库
if (faceApi.queryByUserFaceId(userFaceId) != null) continue
faceApi.insert(entity) 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 changed = true
} }
} }
}
if (changed) { if (changed) {
Timber.i("$TAG MQTT 实时更新完成 ${items.size} 条(实际变更 $changed 条),通知界面刷新引擎") Timber.i("$TAG MQTT 实时更新完成 ${items.size} 条(实际变更 $changed 条),通知界面刷新引擎")
@@ -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()
}
@@ -15,11 +15,13 @@ import com.sw.platecabinet.GlobalKey
import com.sw.platecabinet.model.DeviceConfigV2 import com.sw.platecabinet.model.DeviceConfigV2
import com.sw.platecabinet.model.response.UserFaceModel import com.sw.platecabinet.model.response.UserFaceModel
import com.sw.platecabinet.model.response.UserFaceModelV2 import com.sw.platecabinet.model.response.UserFaceModelV2
import com.sw.platecabinet.mqtt.FaceSyncLock
import com.sw.platecabinet.network.ApiClient import com.sw.platecabinet.network.ApiClient
import com.sw.platecabinet.repository.RemoteRepositoryV2 import com.sw.platecabinet.repository.RemoteRepositoryV2
import com.sw.platecabinet.utils.SpTool import com.sw.platecabinet.utils.SpTool
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.withContext import kotlinx.coroutines.withContext
import timber.log.Timber import timber.log.Timber
import java.io.File import java.io.File
@@ -190,7 +192,8 @@ class NetViewModelV2 : ViewModel() {
} }
withContext(Dispatchers.IO) { withContext(Dispatchers.IO) {
val list = response.data ?: emptyList() val list = response.data ?: emptyList()
updateFaceData(list) // 与 MQTT 实时更新串行化写库,避免同一 userFaceId 重复入库
FaceSyncLock.mutex.withLock { updateFaceData(list) }
// 取当前页最大时间戳,与已累积的比较取最大值 // 取当前页最大时间戳,与已累积的比较取最大值
val pageMaxTimestamp = maxFaceTimestamp(list) val pageMaxTimestamp = maxFaceTimestamp(list)
if (pageMaxTimestamp > faceTimestamp) { if (pageMaxTimestamp > faceTimestamp) {