From 2f8df5afa9f745ee8dd032b2d2cf6a97334ba0eb Mon Sep 17 00:00:00 2001
From: Vickey <891607480@qq.com>
Date: Thu, 30 Apr 2026 21:43:02 +0800
Subject: [PATCH] refactor(data): swap PointValueJob buffer outside the write
 lock
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The quartz-driven batch flush used to hold VALUE_LOCK.writeLock() across
the whole save — receiver threads calling addPointValues() stalled behind
DB I/O whenever a flush was in flight, throttling ingest under load.

The new flow snapshots POINT_VALUE_LIST into a private ArrayList under the
lock, clears the shared buffer, releases the lock, then runs save() on the
snapshot in a virtual thread. The lock window is now just a list copy plus
a clear.

Also: PointValueReceiver's enqueue path now wraps lock/unlock in try-
finally so an unexpected addPointValues() exception cannot leak the lock.
---
 .../pnoker/common/data/job/PointValueJob.java | 21 ++++++++++++-------
 .../data/rabbit/PointValueReceiver.java       |  7 +++++--
 2 files changed, 19 insertions(+), 9 deletions(-)

diff --git a/dc3-common/dc3-common-data/src/main/java/io/github/pnoker/common/data/job/PointValueJob.java b/dc3-common/dc3-common-data/src/main/java/io/github/pnoker/common/data/job/PointValueJob.java
index 18c55e75b..40830f55c 100644
--- a/dc3-common/dc3-common-data/src/main/java/io/github/pnoker/common/data/job/PointValueJob.java
+++ b/dc3-common/dc3-common-data/src/main/java/io/github/pnoker/common/data/job/PointValueJob.java
@@ -92,14 +92,21 @@ public class PointValueJob extends QuartzJobBean {
             log.debug("Point value receiver speed: {} /s, value size: {}, interval: {}", speed, getPointValuesSize(), interval);
         }
 
-        // Save point value array to Redis & PostgreSQL
-        virtualThreadExecutor.execute(() -> {
-            VALUE_LOCK.writeLock().lock();
-            if (!POINT_VALUE_LIST.isEmpty()) {
-                pointValueService.save(POINT_VALUE_LIST);
-                clearPointValues();
+        // Swap out the accumulated buffer under the lock; run the save on a private
+        // snapshot outside the lock so concurrent addPointValues callers are not blocked
+        // by DB I/O.
+        List<PointValueBO> snapshot;
+        VALUE_LOCK.writeLock().lock();
+        try {
+            if (POINT_VALUE_LIST.isEmpty()) {
+                return;
             }
+            snapshot = new ArrayList<>(POINT_VALUE_LIST);
+            POINT_VALUE_LIST.clear();
+        } finally {
             VALUE_LOCK.writeLock().unlock();
-        });
+        }
+
+        virtualThreadExecutor.execute(() -> pointValueService.save(snapshot));
     }
 }
diff --git a/dc3-common/dc3-common-data/src/main/java/io/github/pnoker/common/data/rabbit/PointValueReceiver.java b/dc3-common/dc3-common-data/src/main/java/io/github/pnoker/common/data/rabbit/PointValueReceiver.java
index 6bb67a784..4797afb5b 100644
--- a/dc3-common/dc3-common-data/src/main/java/io/github/pnoker/common/data/rabbit/PointValueReceiver.java
+++ b/dc3-common/dc3-common-data/src/main/java/io/github/pnoker/common/data/rabbit/PointValueReceiver.java
@@ -75,8 +75,11 @@ public class PointValueReceiver {
             } else {
                 // Save point value to schedule
                 PointValueJob.VALUE_LOCK.writeLock().lock();
-                PointValueJob.addPointValues(pointValueBO);
-                PointValueJob.VALUE_LOCK.writeLock().unlock();
+                try {
+                    PointValueJob.addPointValues(pointValueBO);
+                } finally {
+                    PointValueJob.VALUE_LOCK.writeLock().unlock();
+                }
             }
         } catch (Exception e) {
             log.error(e.getMessage(), e);