dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improvement to Device State Service

Browse Source
This commit is contained in:
Andrii Shvaika 2022-08-04 16:18:15 +03:00
parent 2c8b938540
commit ad590da509
2 changed files with 30 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
application/src/main/java/org/thingsboard/server/service/partition/AbstractPartitionBasedService.java
View File

@ -126,7 +126,7 @@ public abstract class AbstractPartitionBasedService<T extends EntityId> extends
} }
List<ListenableFuture<?>> fetchTasks = partitionedFetchTasks.remove(partition); List<ListenableFuture<?>> fetchTasks = partitionedFetchTasks.remove(partition);
if (fetchTasks != null) { if (fetchTasks != null) {
fetchTasks.forEach(f -> f.cancel(true)); fetchTasks.forEach(f -> f.cancel(false));
} }
partitionListChanged = true; partitionListChanged = true;
} }

43
application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java
View File

@ -307,8 +307,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
@Override @Override
public void onSuccess(@Nullable DeviceStateData state) { public void onSuccess(@Nullable DeviceStateData state) {
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, device.getId()); TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, device.getId());
if (partitionedEntities.containsKey(tpi)) { if (addDeviceUsingState(tpi, state)) {
addDeviceUsingState(tpi, state);
save(deviceId, ACTIVITY_STATE, false); save(deviceId, ACTIVITY_STATE, false);
callback.onSuccess(); callback.onSuccess();
} else { } else {
@ -361,25 +360,40 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
// hard-coded limit of 1000 is due to the Entity Data Query limitations and should not be changed. // hard-coded limit of 1000 is due to the Entity Data Query limitations and should not be changed.
for (List<DeviceIdInfo> partition : Lists.partition(entry.getValue(), 1000)) { for (List<DeviceIdInfo> partition : Lists.partition(entry.getValue(), 1000)) {
log.info("[{}] Submit task for device states: {}", entry.getKey(), partition.size()); log.info("[{}] Submit task for device states: {}", entry.getKey(), partition.size());
DevicePackFutureHolder devicePackFutureHolder = new DevicePackFutureHolder();
var devicePackFuture = deviceStateExecutor.submit(() -> { var devicePackFuture = deviceStateExecutor.submit(() -> {
List<DeviceStateData> states; try {
if (persistToTelemetry && !dbTypeInfoComponent.isLatestTsDaoStoredToSql()) { List<DeviceStateData> states;
states = fetchDeviceStateDataUsingSeparateRequests(partition); if (persistToTelemetry && !dbTypeInfoComponent.isLatestTsDaoStoredToSql()) {
} else { states = fetchDeviceStateDataUsingSeparateRequests(partition);
states = fetchDeviceStateDataUsingEntityDataQuery(partition); } else {
states = fetchDeviceStateDataUsingEntityDataQuery(partition);
}
if (devicePackFutureHolder.future != null && !devicePackFutureHolder.future.isCancelled()) {
for (var state : states) {
if (!addDeviceUsingState(entry.getKey(), state)) {
return;
}
checkAndUpdateState(state.getDeviceId(), state);
}
log.info("[{}] Initialized {} out of {} device states", entry.getKey().getPartition().orElse(0), counter.addAndGet(states.size()), entry.getValue().size());
}
} catch (Throwable t) {
log.error("Unexpected exception while device pack fetching", t);
throw t;
} }
for (var state : states) {
addDeviceUsingState(entry.getKey(), state);
checkAndUpdateState(state.getDeviceId(), state);
}
log.info("[{}] Initialized {} out of {} device states", entry.getKey().getPartition().orElse(0), counter.addAndGet(states.size()), entry.getValue().size());
}); });
devicePackFutureHolder.future = devicePackFuture;
result.computeIfAbsent(entry.getKey(), tmp -> new ArrayList<>()).add(devicePackFuture); result.computeIfAbsent(entry.getKey(), tmp -> new ArrayList<>()).add(devicePackFuture);
} }
} }
return result; return result;
} }
private static class DevicePackFutureHolder {
private volatile ListenableFuture<?> future;
}
void checkAndUpdateState(@Nonnull DeviceId deviceId, @Nonnull DeviceStateData state) { void checkAndUpdateState(@Nonnull DeviceId deviceId, @Nonnull DeviceStateData state) {
if (state.getState().isActive()) { if (state.getState().isActive()) {
updateInactivityStateIfExpired(System.currentTimeMillis(), deviceId, state); updateInactivityStateIfExpired(System.currentTimeMillis(), deviceId, state);
@ -391,14 +405,15 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
} }
} }
private void addDeviceUsingState(TopicPartitionInfo tpi, DeviceStateData state) { private boolean addDeviceUsingState(TopicPartitionInfo tpi, DeviceStateData state) {
Set<DeviceId> deviceIds = partitionedEntities.get(tpi); Set<DeviceId> deviceIds = partitionedEntities.get(tpi);
if (deviceIds != null) { if (deviceIds != null) {
deviceIds.add(state.getDeviceId()); deviceIds.add(state.getDeviceId());
deviceStates.putIfAbsent(state.getDeviceId(), state); deviceStates.putIfAbsent(state.getDeviceId(), state);
return true;
} else { } else {
log.debug("[{}] Device belongs to external partition {}", state.getDeviceId(), tpi.getFullTopicName()); log.debug("[{}] Device belongs to external partition {}", state.getDeviceId(), tpi.getFullTopicName());
throw new RuntimeException("Device belongs to external partition " + tpi.getFullTopicName() + "!"); return false;
} }
} }