flowable源码解读之LRU缓存设计

流程数据的定义在flowable中是比较复杂的, 涉及到多张数据库表关联关系，这些在一个流程引擎中也是最为核心的数据，并且需要进行频繁的读取，所以为了实现读取的高效性，flowable将流程定义的数据放在内存中，需要时直接从内存中获取。本文就简单看一看flowable是怎么实现缓存的。

1. 用户定义的入口

先回顾一下，如果我们要在Spring环境中使用flowable必须采用类似方式定义一个bean

<bean id="processEngineConfiguration"
      class="org.flowable.spring.SpringProcessEngineConfiguration">

SpringProcessEngineConfiguration继承于ProcessEngineConfigurationImpl，在ProcessEngineConfigurationImpl源码中可以看到:

    protected int processDefinitionCacheLimit = -1; // By default, no limit
    protected DeploymentCache<ProcessDefinitionCacheEntry> processDefinitionCache;

    protected int processDefinitionInfoCacheLimit = -1; // By default, no limit
    protected ProcessDefinitionInfoCache processDefinitionInfoCache;

    protected int knowledgeBaseCacheLimit = -1;
    protected DeploymentCache<Object> knowledgeBaseCache;

    protected int appResourceCacheLimit = -1;
    protected DeploymentCache<Object> appResourceCache;

因此我们可以在bean定义中可以设置processDefinitionCacheLimit等属性的值即可控制缓存的容量，如果没有进行设置将不会对其限制，为以防止OOM异常，建议可以设置一个，当超出容量时flowable引擎将会通过LRU算法进行移除。

2. 缓存的初始化

在flowable流程引擎在Spring环境中的启动源码分析这篇文章我们已经知道了flowble流程的初始化过程，那么对缓存的初始化肯定也能在相关内里找到，在ProcessEngineConfigurationImpl的init方法中有如下几行代码：

        initProcessDefinitionCache();
        initProcessDefinitionInfoCache();
        initAppResourceCache();
        initKnowledgeBaseCache();

以initProcessDefinitionCache为例看一下方法实现，不用多说:

    public void initProcessDefinitionCache() {
        if (processDefinitionCache == null) {
            if (processDefinitionCacheLimit <= 0) {
                processDefinitionCache = new DefaultDeploymentCache<ProcessDefinitionCacheEntry>();
            } else {
                processDefinitionCache = new DefaultDeploymentCache<ProcessDefinitionCacheEntry>(processDefinitionCacheLimit);
            }
        }
    }

进入DefaultDeploymentCache的构造方法，当没有设置缓存大小时通过无参构造方法创建的是一个同步的Map, 重点可以看一下下面的有参构造函数实现:

/** Cache with no limit */
public DefaultDeploymentCache() {
    this.cache = Collections.synchronizedMap(new HashMap<String, T>());
}

/**
 * Cache which has a hard limit: no more elements will be cached than the limit.
 */
public DefaultDeploymentCache(final int limit) {
    this.cache = Collections.synchronizedMap(new LinkedHashMap<String, T>(limit + 1, 0.75f, true) { // +1 is needed, because the entry is inserted first, before it is removed
        // 0.75 is the default (see javadocs)
        // true will keep the 'access-order', which is needed to have a real LRU cache
        private static final long serialVersionUID = 1L;

        protected boolean removeEldestEntry(Map.Entry<String, T> eldest) {
            boolean removeEldest = size() > limit;
            if (removeEldest && logger.isTraceEnabled()) {
                logger.trace("Cache limit is reached, {} will be evicted", eldest.getKey());
            }
            return removeEldest;
        }

    });
}

有参构造方法核心是基于LinkedHashMap并且重写了removeEldestEntry方法，当超出容量时会返回true, 查看LinkedHashMap可以知道当调用put或putAll返回前会根据该方法返回的值决定是否移除最老的一个元素，从而实现了LRU缓存算法。

3. 缓存的写入与更新

在流程部署时肯定会涉及到相关数据的更新，通过RepositoryServiceImpl.deploy->DeployCmd.executeDeploy查看有如下代码:

        // Actually deploy
        commandContext.getProcessEngineConfiguration().getDeploymentManager().deploy(deployment, deploymentSettings);

查看DeploymentManager.deploy->BpmnDeployer.deploy代码:

        cachingAndArtifactsManager.updateCachingAndArtifacts(parsedDeployment);

然后想看CachingAndArtifactsManager.updateCachingAndArtifacts方法源码即具体更新缓存的实现:

    /**
     * Ensures that the process definition is cached in the appropriate places, including the deployment's collection of deployed artifacts and the deployment manager's cache, as well as caching any
     * ProcessDefinitionInfos.
     */
    public void updateCachingAndArtifacts(ParsedDeployment parsedDeployment) {
        CommandContext commandContext = Context.getCommandContext();
        final ProcessEngineConfigurationImpl processEngineConfiguration = Context.getProcessEngineConfiguration();
        DeploymentCache<ProcessDefinitionCacheEntry> processDefinitionCache = processEngineConfiguration.getDeploymentManager().getProcessDefinitionCache();
        DeploymentEntity deployment = parsedDeployment.getDeployment();

        for (ProcessDefinitionEntity processDefinition : parsedDeployment.getAllProcessDefinitions()) {
            BpmnModel bpmnModel = parsedDeployment.getBpmnModelForProcessDefinition(processDefinition);
            Process process = parsedDeployment.getProcessModelForProcessDefinition(processDefinition);
            ProcessDefinitionCacheEntry cacheEntry = new ProcessDefinitionCacheEntry(processDefinition, bpmnModel, process);
            processDefinitionCache.add(processDefinition.getId(), cacheEntry);
            addDefinitionInfoToCache(processDefinition, processEngineConfiguration, commandContext);

            // Add to deployment for further usage
            deployment.addDeployedArtifact(processDefinition);
        }
    }

4. 缓存的读取

一个典型的读取场景就是在启动流程的时候，所以查看RuntimeServiceImpl.startProcessInstanceByKey->StartProcessInstanceCmd.execute方法源码:

// Find the process definition
ProcessDefinition processDefinition = null;
if (processDefinitionId != null) {

    processDefinition = deploymentCache.findDeployedProcessDefinitionById(processDefinitionId);
    if (processDefinition == null) {
        throw new FlowableObjectNotFoundException("No process definition found for id = '" + processDefinitionId + "'", ProcessDefinition.class);
    }

}

接下来进入DeploymentManager.findDeployedProcessDefinitionById可以看到, 首先会从缓存中查找，如果没有则从数据库中加载:

public ProcessDefinition findDeployedProcessDefinitionById(String processDefinitionId) {
    if (processDefinitionId == null) {
        throw new FlowableIllegalArgumentException("Invalid process definition id : null");
    }

    // first try the cache
    ProcessDefinitionCacheEntry cacheEntry = processDefinitionCache.get(processDefinitionId);
    ProcessDefinition processDefinition = cacheEntry != null ? cacheEntry.getProcessDefinition() : null;

    if (processDefinition == null) {
        processDefinition = processDefinitionEntityManager.findById(processDefinitionId);
        if (processDefinition == null) {
            throw new FlowableObjectNotFoundException("no deployed process definition found with id '" + processDefinitionId + "'", ProcessDefinition.class);
        }
        processDefinition = resolveProcessDefinition(processDefinition).getProcessDefinition();
    }
    return processDefinition;
}

然后看一下resolveProcessDefinition方法, 当缓存中没有数据时会调用deploy方法来重新加载缓存。

/**
 * Resolving the process definition will fetch the BPMN 2.0, parse it and store the {@link BpmnModel} in memory.
 */
public ProcessDefinitionCacheEntry resolveProcessDefinition(ProcessDefinition processDefinition) {
    String processDefinitionId = processDefinition.getId();
    String deploymentId = processDefinition.getDeploymentId();

    ProcessDefinitionCacheEntry cachedProcessDefinition = processDefinitionCache.get(processDefinitionId);

    if (cachedProcessDefinition == null) {
        if (Flowable5Util.isFlowable5ProcessDefinition(processDefinition, processEngineConfiguration)) {
            return Flowable5Util.getFlowable5CompatibilityHandler().resolveProcessDefinition(processDefinition);
        }

        DeploymentEntity deployment = deploymentEntityManager.findById(deploymentId);
        deployment.setNew(false);
        deploy(deployment, null);
        cachedProcessDefinition = processDefinitionCache.get(processDefinitionId);

        if (cachedProcessDefinition == null) {
            throw new FlowableException("deployment '" + deploymentId + "' didn't put process definition '" + processDefinitionId + "' in the cache");
        }
    }
    return cachedProcessDefinition;
}

总结一下：flowable缓存的实现核心即基于LinkedHashMap并通过重写其removeEldestEntry方法实现LRU缓存移除算法。以流程定义缓存为例可以知道，每次部署时会将流程定义的数据加入缓存，每次流程启动时都会尝试去缓存中获取数据，如果缓存中有就直接返回，如果没有就从数据库中加载并放入缓存以供下次使用。