Оптимизация сценария высокочастотной записи 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, использующий кэш, с его прослушиванием событий (см. класс com.google.common.cache.RemovalCause), redis инициирует операцию записи, addListSync синхронизируется с использованием метода блокировки, высокопараллельные сценарии 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 CacheConfig 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, он анализирует модель значений в кеше и сохраняет ее. После выполнения вышеуказанных шагов просто объявите на бизнес-уровне

@Autowired
private BufferCache buffCache;

Просто вызовите его метод addListSync.

Суммировать

Общая идея состоит в том, чтобы использовать локальный кеш, чтобы разделить давление высокочастотной записи.На самом деле, этот метод подходит не только для повторной записи, но и для других сценариев.Конкретный метод использования может быть расширен в соответствии с бизнес-сценарием. .