自定义缓存规则节点

正文与代码来自原稿。对齐 TB 自带缓存体系，增加 ruleChainCache，规则节点内读写，避免反复查库。

1、Caffeine / Redis 缓存实现

RuleChainCaffeineCache

package org.thingsboard.server.cache;

import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.cache.CacheManager;
import org.springframework.stereotype.Service;
import org.thingsboard.server.common.data.CacheConstants;

@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "caffeine", matchIfMissing = true)
@Service("RuleChainCache")
public class RuleChainCaffeineCache extends CaffeineTbTransactionalCache<String, String> {

    public RuleChainCaffeineCache(CacheManager cacheManager) {
        super(cacheManager, CacheConstants.RULE_CHAIN_CACHE);
    }
}

RuleChainRedisCache

package org.thingsboard.server.cache;

import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.stereotype.Service;
import org.thingsboard.server.common.data.CacheConstants;

@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "redis")
@Service("RuleChainCache")
public class RuleChainRedisCache extends RedisTbTransactionalCache<String, String> {

    public RuleChainRedisCache(TBRedisCacheConfiguration configuration,
                               CacheSpecsMap cacheSpecsMap,
                               RedisConnectionFactory connectionFactory) {
        super(CacheConstants.RULE_CHAIN_CACHE, cacheSpecsMap, connectionFactory,
                configuration, new TbFSTRedisSerializer<>());
    }
}

2、CacheConstants

public static final String RULE_CHAIN_CACHE = "ruleChainCache";

3、thingsboard.yml

cache:
  type: "${CACHE_TYPE:caffeine}"
  maximumPoolSize: "${CACHE_MAXIMUM_POOL_SIZE:16}"
  attributes:
    enabled: "${CACHE_ATTRIBUTES_ENABLED:true}"
  specs:
    ruleChainCache:
      timeToLiveInMinutes: "${CACHE_SPECS_RULE_CHAIN_CACHE_TTL:1440}"
      maxSize: "${CACHE_SPECS_RULE_CHAIN_CACHE_MAX_SIZE:10000}"

使用 Redis 时注意 maxmemory-policy（如 allkeys-lru）等官方要求。

4、自定义节点 TbCacheNode

通过 SpringContextUtil 取对应 Bean，对消息或元数据做 SET / GET。

package org.thingsboard.rule.engine.custom.cache;

import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.ObjectUtils;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.rule.engine.api.*;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.server.cache.RuleChainCaffeineCache;
import org.thingsboard.server.cache.RuleChainRedisCache;
import org.thingsboard.server.cache.TbTransactionalCache;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.util.SpringContextUtil;

import java.util.Map;
import java.util.concurrent.ExecutionException;

import static org.thingsboard.common.util.DonAsynchron.withCallback;

@RuleNode(
        type = ComponentType.ACTION,
        name = "use cache",
        configClazz = TbCacheNodeConfiguration.class,
        nodeDescription = "Use cache",
        nodeDetails = "Use cache",
        uiResources = {"static/rulenode/rulenode-core-config.js"},
        icon = "add_circle")
@Slf4j
public class TbCacheNode implements TbNode {

    private TbCacheNodeConfiguration config;
    private TbTransactionalCache<String, String> tbTransactionalCache;

    @Override
    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
        this.config = TbNodeUtils.convert(configuration, TbCacheNodeConfiguration.class);
        if ("caffeine".equals(config.getCacheType())) {
            this.tbTransactionalCache = SpringContextUtil.getBean(RuleChainCaffeineCache.class);
        } else if ("redis".equals(config.getCacheType())) {
            this.tbTransactionalCache = SpringContextUtil.getBean(RuleChainRedisCache.class);
        } else {
            throw new RuntimeException("Cache type error");
        }
    }

    @Override
    public void onMsg(TbContext ctx, TbMsg msg)
            throws ExecutionException, InterruptedException, TbNodeException {
        try {
            withCallback(operateCache(msg),
                    res -> ctx.tellSuccess(msg),
                    t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
        } catch (Exception e) {
            e.printStackTrace();
            ctx.tellFailure(msg, e);
        }
    }

    private ListenableFuture<? extends Object> operateCache(TbMsg msg) {
        if ("SET".equals(config.getMethod())) {
            if (ObjectUtils.isEmpty(config.getCacheKey())
                    || ObjectUtils.isEmpty(config.getCacheValueKey())) {
                throw new RuntimeException("parameter error");
            }
            if ("msg".equals(config.getMsgMethod())) {
                JsonNode jsonNode = JacksonUtil.toJsonNode(msg.getData());
                String key = jsonNode.get(config.getCacheKey()).asText();
                String value = jsonNode.get(config.getCacheValueKey()).asText();
                tbTransactionalCache.put(key, value);
            } else if ("metadata".equals(config.getMsgMethod())) {
                Map<String, String> data = msg.getMetaData().getData();
                String key = data.get(config.getCacheKey());
                String value = data.get(config.getCacheValueKey());
                tbTransactionalCache.put(key, value);
            } else {
                throw new RuntimeException("method only msg or metadata");
            }
            return Futures.immediateFuture(true);
        } else if ("GET".equals(config.getMethod())) {
            if (ObjectUtils.isEmpty(config.getCacheKey())
                    || ObjectUtils.isEmpty(config.getCacheValueKey())) {
                throw new RuntimeException("parameter error");
            }
            if ("msg".equals(config.getMsgMethod())) {
                JsonNode jsonNode = JacksonUtil.toJsonNode(msg.getData());
                String key = jsonNode.get(config.getCacheKey()).asText();
                Object value = tbTransactionalCache.get(key).get();
                msg.getMetaData().putValue(config.getCacheValueKey(), (String) value);
                return Futures.immediateFuture(true);
            } else if ("metadata".equals(config.getMsgMethod())) {
                Map<String, String> data = msg.getMetaData().getData();
                String key = data.get(config.getCacheKey());
                Object value = tbTransactionalCache.get(key).get();
                msg.getMetaData().putValue(config.getCacheValueKey(), (String) value);
                return Futures.immediateFuture(true);
            } else {
                throw new RuntimeException("method only msg or metadata");
            }
        } else {
            throw new RuntimeException("method only SET or GET");
        }
    }
}

TbCacheNodeConfiguration

package org.thingsboard.rule.engine.custom.cache;

import lombok.Data;
import org.thingsboard.rule.engine.api.NodeConfiguration;

@Data
public class TbCacheNodeConfiguration implements NodeConfiguration<TbCacheNodeConfiguration> {

    private String cacheType;      // caffeine、redis
    private String msgMethod;      // msg 或 metadata
    private String method;         // SET、GET
    private String cacheKey;       // 用作缓存 key 的字段名
    private String cacheValueKey;  // SET 时取 value 的字段；GET 时写入 metadata 的键

    @Override
    public TbCacheNodeConfiguration defaultConfiguration() {
        TbCacheNodeConfiguration configuration = new TbCacheNodeConfiguration();
        configuration.setCacheType("caffeine");
        configuration.setMsgMethod("metadata");
        configuration.setMethod("SET");
        configuration.setCacheKey("");
        configuration.setCacheValueKey("");
        return configuration;
    }
}

辅

SpringContextUtil 需自行提供或从 ApplicationContext 注入；GET 后建议 TbMsg.transformMsg 再向下传递，避免原地改元数据在部分版本不生效。

5、小结

先注册缓存 specs，再在规则节点里 SET/GET，适合跨消息共享中间状态（计数、会话标记等）。