flowable源码解读——内存缓存设计
最近检查flowable缓存问题,顺便看了下源码,flowable默认是将缓存存储在内存中,以减少对数据库的压力,以下是flowable缓存类的设计。
DeploymentCache接口设计:
//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//
package org.flowable.common.engine.impl.persistence.deploy;
public interface DeploymentCache<T> {
T get(String var1);
boolean contains(String var1);
void add(String var1, T var2);
void remove(String var1);
void clear();
}
DefaultDeploymentCache<T>类设计:
/* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.flowable.common.engine.impl.persistence.deploy;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Default cache: keep everything in memory, unless a limit is set.
*
* @author Joram Barrez
*/
public class DefaultDeploymentCache<T> implements DeploymentCache<T> {
private static final Logger LOGGER = LoggerFactory.getLogger(DefaultDeploymentCache.class);
protected Map<String, T> cache;
/** Cache with no limit */
public DefaultDeploymentCache() {
this.cache = Collections.synchronizedMap(new HashMap<>());
}
/**
* 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;
@Override
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;
}
});
}
@Override
public T get(String id) {
return cache.get(id);
}
@Override
public void add(String id, T obj) {
cache.put(id, obj);
}
@Override
public void remove(String id) {
cache.remove(id);
}
@Override
public boolean contains(String id) {
return cache.containsKey(id);
}
@Override
public void clear() {
cache.clear();
}
// For testing purposes only
public Collection<T> getAll() {
return cache.values();
}
// For testing purposes only
public int size() {
return cache.size();
}
}
通过源码可以看到 flowable 是用 Collections.synchronizedMap 的方式保证缓存写入的线程安全问题。创建线程安全的Map主要有以下三种方式:
// 无非就是以下三种方式
Map<String, Object> map2 = new Hashtable<String, Object>();Map<String, Object> map3 = new ConcurrentHashMap<String, Object>();
Map<String, Object> map4 = Collections.synchronizedMap(new HashMap<String, Object>());
Collections.synchronizedMap 是使用一个包装器类
来实现的。ConcurrentHashMap 是使用数组+链表
来实现的。
1・Collections.synchronizedMap() : 这是一个静态工厂方法,它接收一个现有的 Map 实例作为参数,并返回一个经过包装的线程安全的 Map。使用一个全局的锁来确保线程安全,无论是读取还是写入操作都会获取这把锁,这意味着所有的操作都是互斥的,即在同一时刻只能有一个线程进行操作,因此不能说它是原子级别的操作,除非操作本身就是原子性的(例如,简单的 get 或 put 操作针对的是单个对象引用)。在高并发环境下,当多个线程同时访问不同的键值对时,可能会导致性能瓶颈。
2・ConcurrentHashMap :采用了一种粒度更细的锁机制(通常称为分段锁或者分区锁),将内部的数据结构划分为多个部分或桶(bucket),每个桶都可以独立地加锁和解锁,从而实现了更高的并发性能。在大多数情况下,多个线程可以同时对不同键进行读写操作,而不会相互阻塞,除非它们碰巧要修改同一个桶内的数据。
通过以上对比可以看到 Collections.synchronizedMap()的方式并不是原子级别的锁定,在高并发下有性能瓶颈,不太理解为什么flowable不用ConcurrentHashMap 这种方式做内存缓存,也许是考虑对流程定义的读写不会太频繁才这么设计。
如果对并发很高的场景 可以用ConcurrentHashMap 这种方式实现,以下是实现代码:
package org.flowable.common.engine.impl.persistence.deploy;
import java.util.Collection;
import java.util.concurrent.ConcurrentHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class DefaultDeploymentCacheConcurrent<T> implements DeploymentCache<T> {
private static final Logger LOGGER = LoggerFactory.getLogger(DefaultDeploymentCacheConcurrent.class);
protected ConcurrentHashMap<String, T> cache;
public DefaultDeploymentCacheConcurrent() {
this.cache = new ConcurrentHashMap<>();
}
public DefaultDeploymentCacheConcurrent(final int limit) {
this.cache = new ConcurrentHashMap<>(limit + 1, 0.75f, 1); // concurrencyLevel set to 1 for simplicity
}
public T get(String id) {
return this.cache.get(id);
}
public void add(String id, T obj) {
this.cache.put(id, obj);
}
public void remove(String id) {
this.cache.remove(id);
}
public boolean contains(String id) {
return this.cache.containsKey(id);
}
public void clear() {
this.cache.clear();
}
public Collection<T> getAll() {
return this.cache.values();
}
public int size() {
return this.cache.size();
}
}
以上是基于内存的缓存设计,这边现在有项目是部署在负载环境下,内存缓存已经不能满足需求,后续会将内存存储到redis库。