高频写入redis场景优化

工作中经常遇到要对 redis 进行高频写入，但是对于读取时数据的实时性要求又不高的场景。为了优化性能，决定采用本地缓存一部分数据整合后写入。

依赖

<dependency>
	<groupId>com.google.guava</groupId>
	<artifactId>guava</artifactId>
	<version>19.0-rc2</version>
</dependency>
复制代码

基础类

public class BufferCache implements Closeable {
    // CacheBuilder的构造函数是私有的，只能通过其静态方法newBuilder()来获得CacheBuilder的实例
    private Cache localCacheData;
    private static int maxItemSize = 1000;
    private static String key = "defaultKey";
    private static final Object lock = new Object();

    public BufferCache(String key, int currencyLevel, int writeExpireTime,
                       int accessExpireTime, int initialCapacity, int maximumSize,
                       int maxItemSize, RemovalListener removalListener) {
        currencyLevel = currencyLevel < 1 ? 1 : currencyLevel;
        initialCapacity = initialCapacity < 100 ? 100 : initialCapacity;
        if (key!=null&&key.isEmpty()) {
            BufferCache.key = key;
        }

        BufferCache.maxItemSize = maxItemSize;

        localCacheData = CacheBuilder.newBuilder()
                // 设置并发级别为8，并发级别是指可以同时写缓存的线程数
                .concurrencyLevel(currencyLevel)
                // 设置写缓存后expireTime秒钟过期
                .expireAfterWrite(writeExpireTime, TimeUnit.SECONDS)
                // 设置请求后expireTime秒钟过期
                .expireAfterAccess(accessExpireTime, TimeUnit.SECONDS)
                // 设置缓存容器的初始容量为10
                .initialCapacity(initialCapacity)
                // 设置缓存最大容量为Integer.MAX_VALUE，超过Integer.MAX_VALUE之后就会按照LRU最近虽少使用算法来移除缓存项
                .maximumSize(maximumSize)
                // 设置要统计缓存的命中率
                .recordStats()
                // 设置缓存的移除通知
                .removalListener(removalListener)
                // build方法中可以指定CacheLoader，在缓存不存在时通过CacheLoader的实现自动加载缓存
                .build();

        Runtime.getRuntime().addShutdownHook(
                new Thread(() -> localCacheData.invalidate(key)));
    }

    public void addListSync(String key, Object value) {
        synchronized (lock) {
            List<Object> gs = (List<Object>) localCacheData.getIfPresent(key);
            if (gs == null) {
                gs = new ArrayList<>();
            }
            gs.add(value);
            localCacheData.put(key, gs);

            // 如果队列长度超过设定最大长度则清除key
            if (gs.size() > maxItemSize) {
                localCacheData.invalidate(key);
            }
        }
    }

    public void addListSync(Object value) {
        addListSync(BufferCache.key, value);
    }

    @Override
    public void close() {
        localCacheData.invalidate(key);
    }
}
复制代码

采用 google 的 cache，利用其监听事件(详见 com.google.common.cache.RemovalCause 类)触发写入redis操作，addListSync方法中使用 synchronized 进行加锁，防止高并发场景下List数据错误。

新建配置文件

cache.key=name
cache.currencyLevel=1
cache.writeExpireTime=900
cache.accessExpireTime=600
cache.initialCapacity=1
cache.maximumSize=1000
cache.maxItemSize=1000
复制代码

针对不同业务场景可以自定义不同的配置参数

业务实现

@Configuration
@ConditionalOnResource(resources = "bufferCache.properties")
@PropertySource(value = "bufferCache.properties", ignoreResourceNotFound = true)
public class GuildCacheConfig implements ApplicationContextAware {
	private ApplicationContext ctx;

	@Bean("buffCache")
	@ConditionalOnProperty(prefix = "cache", value = "currencyLevel")
	public BufferCache guildBuffCache(@Value("${cache.key}") String key,
			@Value("${cache.currencyLevel}") int currencyLevel,
			@Value("${cache.writeExpireTime}") int writeExpireTime,
			@Value("${cache.accessExpireTime}") int accessExpireTime,
			@Value("${cache.initialCapacity}") int initialCapacity,
			@Value("${cache.maximumSize}") int maximumSize,
			@Value("${cache.maxItemSize}") int maxItemSize) {

		// 异步监听
		RemovalListener<String, List<GuildActiveEventEntity>> async = RemovalListeners
				.asynchronous(new MyRemovalListener(),
						ExecutorServiceUtil.getExecutorServiceByType(
								ExecutorServiceUtil.ExecutorServiceType.BACKGROUND));
		return new BufferCache(key, currencyLevel, writeExpireTime,
				accessExpireTime, initialCapacity, maximumSize, maxItemSize,
				async);
	}

	@Override
	public void setApplicationContext(ApplicationContext applicationContext)
			throws BeansException {
		ctx = applicationContext;
	}

	// 创建一个监听器
	private class MyRemovalListener
			implements RemovalListener<String, List<GuildActiveEventEntity>> {
		@Override
		public void onRemoval(
				RemovalNotification<String, List<GuildActiveEventEntity>> notification) {
			RemovalCause cause = notification.getCause();

			// 当超出缓存队列限制大小时或者key过期或者主动清除key时更新数据
			if (cause.equals(RemovalCause.SIZE)
					|| cause.equals(RemovalCause.EXPIRED)
					|| cause.equals(RemovalCause.EXPLICIT)) {
				//根据不同业务场景调用不同业务方法进行写入操作
			}

		}
	}
}

复制代码

此类实现 ApplicationContextAware 为了获取指定业务方法 Bean ，进行解析缓存中value模型后进行存储。 在以上几个步骤都完成后，只需在业务层声名

@Autowired
private BufferCache buffCache;
复制代码

调用其addListSync方法即可。

总结

总体思路是使用本地缓存去分担高频写的压力，此方法其实不仅仅适用与redis的写入，还可用于其他场景，具体使用方法可以按照业务场景自己扩展。