refactor(data): swap PointValueJob buffer outside the write lock

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.
This commit is contained in:
Vickey
2026-04-30 21:43:02 +08:00
parent e6641048f9
commit 2f8df5afa9
2 changed files with 19 additions and 9 deletions
@@ -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));
}
}
@@ -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);