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 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);