class AutoUpdateThingsDataManager extends LocalFileThingsDataManager {

    private final Map<ThingId, Updater> updaters = Caffeine
        .newBuilder()
        //10分钟没有任何读取则dispose取消订阅
        .expireAfterAccess(Duration.ofMinutes(10))
        .<ThingId, Updater>removalListener((key, value, cause) -> {
            if (cause == RemovalCause.EXPIRED) {
                if (value != null) {
                    value.dispose();
                }
            }
        })
        .build()
        .asMap();

    private final EventBus eventBus;

    public AutoUpdateThingsDataManager(String fileName, EventBus eventBus) {
        super(fileName);
        this.eventBus = eventBus;
    }

    @Override
    public Mono<List<ThingProperty>> getProperties(String thingType, String thingId, String property, long from, long to) {
        Updater updater = getUpdater(thingType, thingId);
        updater.tryLoad(property);
        Mono<Void> loader = updater.loader;
        if (updater.loading && loader != null) {
            return loader
                .then(Mono.defer(() -> super.getProperties(thingType, thingId, property, from, to)));
        }
        return super.getProperties(thingType, thingId, property, from, to);
    }

    @Override
    public Mono<ThingProperty> getLastProperty(String thingType,
                                               String thingId,
                                               String property,
                                               long baseTime) {<FILL_FUNCTION_BODY>}

    @Override
    public Mono<ThingProperty> getFirstProperty(String thingType, String thingId, String property) {
        Updater updater = getUpdater(thingType, thingId);
        updater.tryLoad(property);

        Mono<Void> loader = updater.loader;

        if (updater.loading && loader != null) {
            return loader
                .then(Mono.defer(() -> super.getFirstProperty(thingType, thingId, property)));
        }
        return super.getFirstProperty(thingType, thingId, property);
    }

    private Updater getUpdater(String thingType, String thingId) {
        ThingId key = ThingId.of(thingType, thingId);
        return updaters
            .computeIfAbsent(key, this::createUpdater);
    }

    private Mono<Void> loadData(String thingType, String thingId, String property) {
        //fixme 不加载不存在的数据,如果查库可能会导致瞬间压力过高
        return Mono.empty();
    }

    protected Updater createUpdater(ThingId id) {
        return new Updater(id.getType(), id.getId());
    }

    @SneakyThrows
    private ByteBuf encodeHistory(PropertyHistory history) {
        ByteBuf buf = Unpooled.buffer();
        try (ObjectOutput out = createOutput(buf)) {
            history.writeExternal(out);
        }
        return buf;
    }

    @SneakyThrows
    private PropertyHistory decodeHistory(ByteBuf buf) {
        PropertyHistory history = new PropertyHistory();
        try (ObjectInput input = createInput(buf)) {
            history.readExternal(input);
        }
        return history;
    }

    @Override
    public void shutdown() {
        super.shutdown();
        updaters.values().forEach(Disposable::dispose);
    }

    @Getter
    @Setter
    @AllArgsConstructor(staticName = "of")
    @NoArgsConstructor
    public static class ThingHistoryRequest {
        private String thingType;
        private String thingId;
        private String property;
    }

    protected class Updater implements Disposable {
        private final String thingType;
        private final String thingId;
        private final Disposable disposable;

        private final Set<String> include = ConcurrentHashMap.newKeySet();

        private boolean loading;
        private Mono<Void> loader;

        public Updater(String thingType, String thingId) {
            this.thingType = thingType;
            this.thingId = thingId;
            //订阅整个集群的消息来更新本地的缓存数据
            disposable = eventBus
                .subscribe(
                    Subscription.builder()
                                .subscriberId("thing-data-property-updater")
                                .topics(ThingConstants.Topics.properties(ThingType.of(thingType), thingId))
                                .local()
                                .broker()
                                .priority(Integer.MIN_VALUE)
                                .build(),
                    ThingMessage.class
                )
                .doOnNext(this::doUpdate)
                .subscribe();
        }

        private void doUpdate(ThingMessage thingMessage) {
            if (!(thingMessage instanceof PropertyMessage)) {
                return;
            }
            PropertyMessage message = (PropertyMessage) thingMessage;
            try {
                Map<String, Object> properties = message.getProperties();
                if (properties == null) {
                    return;
                }
                for (Map.Entry<String, Object> entry : properties.entrySet()) {
                    String property = entry.getKey();

                    if (include.contains(property)) {
                        updateProperty0(thingType,
                                        thingId,
                                        property,
                                        message
                                            .getPropertySourceTime(property)
                                            .orElse(message.getTimestamp()),
                                        entry.getValue(),
                                        message.getPropertyState(property).orElse(null)
                        );
                    }
                }
            } catch (Throwable ignore) {
                //ignore
            }
        }

        private void tryLoad(String property) {
            if (include.add(property)) {
                this.loading = true;
                this.loader = loadData(thingType, thingId, property)
                    .doAfterTerminate(() -> {
                        loading = false;
                        loader = null;
                    })
                    .cache();
            }
        }

        @Override
        public void dispose() {
            disposable.dispose();
            include.clear();
        }

        @Override
        public boolean isDisposed() {
            return disposable.isDisposed();
        }
    }

}

        Updater updater = getUpdater(thingType, thingId);
        updater.tryLoad(property);
        Mono<Void> loader = updater.loader;

        if (updater.loading && loader != null) {
            return loader
                .then(Mono.defer(() -> super.getLastProperty(thingType, thingId, property, baseTime)));
        }
        return super.getLastProperty(thingType, thingId, property, baseTime);
    
<methods>public void <init>(java.lang.String) ,public void <init>(MVStore) ,public Mono<org.jetlinks.core.things.ThingProperty> getFirstProperty(java.lang.String, java.lang.String, java.lang.String) ,public Mono<java.lang.Long> getFirstPropertyTime(java.lang.String, java.lang.String) ,public Mono<org.jetlinks.community.things.data.ThingEvent> getLastEvent(java.lang.String, java.lang.String, java.lang.String, long) ,public Mono<org.jetlinks.core.things.ThingProperty> getLastProperty(java.lang.String, java.lang.String, java.lang.String, long) ,public Mono<java.lang.Long> getLastPropertyTime(java.lang.String, java.lang.String, long) ,public Mono<List<org.jetlinks.core.things.ThingProperty>> getProperties(java.lang.String, java.lang.String, java.lang.String, long) ,public Mono<List<org.jetlinks.core.things.ThingProperty>> getProperties(java.lang.String, java.lang.String, java.lang.String, long, long) ,public Mono<java.lang.Void> removeEvent(java.lang.String, java.lang.String, java.lang.String) ,public Mono<java.lang.Void> removeProperties(java.lang.String, java.lang.String) ,public Mono<java.lang.Void> removeProperty(java.lang.String, java.lang.String, java.lang.String) ,public void shutdown() ,public Mono<java.lang.Void> updateEvent(java.lang.String, java.lang.String, java.lang.String, long, java.lang.Object) ,public final Mono<java.lang.Void> updateProperty(java.lang.String, java.lang.String, org.jetlinks.core.things.ThingProperty) ,public Mono<java.lang.Void> updateProperty(java.lang.String, java.lang.String, java.lang.String, long, java.lang.Object, java.lang.String) <variables>static int DEFAULT_MAX_STORE_SIZE_EACH_KEY,private static final AtomicIntegerFieldUpdater<org.jetlinks.community.things.data.LocalFileThingsDataManager> TAG_INC,private final Map<org.jetlinks.community.things.data.LocalFileThingsDataManager.StoreKey,org.jetlinks.community.things.data.LocalFileThingsDataManager.PropertyHistory> historyCache,private final non-sealed MVMap<java.lang.Long,org.jetlinks.community.things.data.LocalFileThingsDataManager.PropertyHistory> historyStore,protected final non-sealed MVStore mvStore,private final Map<java.lang.String,java.lang.Integer> tagCache,private volatile int tagInc,private final non-sealed MVMap<java.lang.String,java.lang.Integer> tagStore