游戏缓存与异步持久化的完美邂逅

1、问题提出

游戏服务器，需要频繁的读取玩家数据，同时也需求频发修改玩家数据，并持久化到数据库。为了提高游戏服务器的性能，我们应该怎么处理呢？

2、应用程序缓存

缓存，是指应用程序从数据库读取完数据之后，就将数据缓存在进程内存或第三方内存（例如redis）。游戏服务器对于玩家数据的读写是非常频繁的，为了减低数据库压力，通常会使用缓存。以下是一些使用缓存的好处：

1. 提高响应速度：缓存可以将之前计算或检索的结果存储在内存中，当下次相同的请求到达时，可以直接从缓存中获取结果，避免了重复的计算或数据库查询，从而大幅提高响应速度。

2. 减少对资源的访问压力：缓存可以减少对资源的频繁访问，比如数据库、网络等，从而减少对这些资源的压力。这可以提高应用程序的整体性能，并降低对资源的依赖。

3. 支持高并发：使用缓存可以缓解高并发环境下对数据库或其他资源的并发访问压力。通过将经常访问的数据存储在缓存中，可以提供更快的响应时间，并支持更高的并发请求。

3、spring使用缓存

3.1、SpringCache基本使用方法

在Spring中，可以通过Spring Cache来使用缓存。下面是使用Spring Cache的一般步骤：

1. 添加依赖：在项目的构建文件（如pom.xml）中添加Spring Cache的相关依赖。

2. 配置缓存管理器：在Spring的配置文件（如applicationContext.xml）中配置缓存管理器。可以选择使用Spring提供的缓存管理器实现，如ConcurrentMapCacheManager、EhCacheCacheManager等，也可以自定义缓存管理器。

3. 在需要缓存的方法上添加缓存注解：在需要进行缓存的方法上添加Spring Cache的缓存注解，如@Cacheable、@CachePut等。这些注解可以指定缓存的名称、缓存条目的键，以及在何时加入或刷新缓存条目。

4. 配置缓存注解的属性：根据需求，可以为缓存注解添加一些属性，如缓存的失效时间、编写缓存的键生成器等。

5. 启用缓存功能：在Spring的配置类上使用@EnableCaching注解，以启用Spring Cache的功能

SpringCache通过注解提供缓存服务，注解只是提供一个抽象的统一访问接口，而没有提供缓存的实现。对于每个版本的spring，其使用的缓存实现存在一定的差异性。例如springboot 2.X，提供以下的缓存实现。

public enum CacheType {
    GENERIC,
    JCACHE,
    EHCACHE,
    HAZELCAST,
    INFINISPAN,
    COUCHBASE,
    REDIS,
    CAFFEINE,
    SIMPLE,
    NONE;

    private CacheType() {
    }
}

3.2、SpringCache常用注解

SpringCache最重要有以下几个注解

1. @Cacheable：将方法的返回值缓存起来，并在下次调用时，直接从缓存中获取，而不执行方法体。

2. @CachePut：将方法的返回值缓存起来，与@Cacheable不同的是，@CachePut会每次都执行方法体，并将返回值放入缓存中。

3. @CacheEvict：从缓存中移除一个或多个条目。可以通过指定的key来删除具体的缓存条目，或者通过allEntries属性来删除所有的缓存条目。

3.3、使用进程缓存与进程外缓存的区别

SpringCache底层的缓存实现，即可以使用进程内缓存（例如EhCache），也可以使用进程外缓存（例如Redis）。得益于SpringCache的优秀API，在业务代码切换缓存实现，仅需修改配置文件，及针对不同缓存实现的个性化配置，使用缓存的业务代码几乎不用做任何修改。

使用进程内缓存特点：应用程序重启之后，缓存随即失效，需要重新加载。

使用进程外缓存特点：应用程序重启之后，只要redis不重启，缓存仍然生效（当然，Redis可以选择持久化，及时重启也能保存缓存数据）。对于进程外缓存，由于有对应的可视化工具，可以帮助用户加深对SpringCache的理解。

总结：对于活跃缓存数据比较多，推荐使用Redis等进程外缓存。而如果活跃缓存不是很多，直接使用进程内缓存即可。因为进程外缓存，虽然有诸多优点，但由于跨进程，甚至跨机器，需要额外使用网络io，程序与redis数据通信导致的序列化反序列化io，有大量io消耗。

3.4、缓存击穿与缓存雪崩

缓存击穿：指缓存中没有数据，所有的请求都落到了数据库上，可能导致数据库压力剧增。

解决方法：

1. 游戏服务器一般会缓存所有玩家基础信息，包括uid，等级，姓名等。在读取数据的时候进行预判，过滤无效的id

2. 缓存空对象：对查询结果为空的key，可以设置一个默认值或者空对象进行缓存，并设置一个较短的过期时间。

缓存雪崩：指缓存同一时间大量失效，导致所有请求都落到了数据库上，可能导致数据库压力剧增。

解决方法：

1. 设置XX时间没进行读写才移除缓存，例如使用Caffeine缓存，可以设置expireAfterWrite等参数，只要数据属于热门数据（近期有访问），则不会从缓存移除。

2. 设置永不过期，如果使用redis缓存，由于redis只能设置全局ttl过期时间，无法刷新访问时间，只能选择永不过期。

3.5、SpringCache使用Redis进程外缓存

1.引入Redis相关依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
 

2.在application.properties或application.yml文件中配置Redis连接信息

##使用redis缓存
spirng.cache.type=redis

##redis相关配置
spring.redis.host=127.0.0.1
spring.redis.port=6379
spring.redis.password=
spring.redis.database=0

3.修改redis使用json数据格式

springcache redis默认的序列化方式基于jdk自带的序列化方式。（这里需要吐槽一下，jdk自带的序列化非常垃圾，根本没什么人使用。实体需要实现Serializable接口不说，性能又差，无法跨语言，全身上下尽是缺点）

@Configuration
public class RedisConfig  extends CachingConfigurerSupport {


    @Bean
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(factory);
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        MyObjectMapper objectMapper = new MyObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

    @Bean
    public CacheManager cacheManager(RedisConnectionFactory factory) {
        RedisSerializer<String> redisSerializer = new StringRedisSerializer();
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        MyObjectMapper objectMapper = new MyObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
        // 配置序列化
        RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig();
        RedisCacheConfiguration redisCacheConfiguration = config.serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(redisSerializer)).serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer));
        return RedisCacheManager.builder(factory).cacheDefaults(redisCacheConfiguration).build();

    }


    private class MyObjectMapper extends ObjectMapper {

        private static final long serialVersionUID = 1L;

        public MyObjectMapper() {
            super();
            this.configure(MapperFeature.USE_ANNOTATIONS, false);
            // 只针对非空的值进行序列化
            this.setSerializationInclusion(JsonInclude.Include.NON_NULL);
            this.enableDefaultTyping(DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY);
            this.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
            this.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
        }
    }

}

4、封装SpringCache操作缓存

4.1、SpringCache操作缓存的坑

SpringCache默认使用代理机制来实现，如果在同一个类内部，调用有缓存注解的方法，是不会触发缓存的，例如下面的代码。

@Service
public class MyServiceImpl implements MyService {

    @Cacheable("myCache")
    @Override
    public String getValue() {
        System.out.println("Getting value from getValue() method");
        return "myValue";
    }

    @Override
    public void callGetValue() {
        // 以下代码不会触发缓存
        String value =  this.getValue();
        System.out.println("Value: " + value);
    }
}

有多种方法解决以上的问题，比如：

• 使用AspectJ实现AOP（编译阶段织入），不采用默认的Proxy实现AOP。
• 分离缓存实现与业务调用代码，数据缓存单独放在一个类，跟其他调用缓存的业务代码分离开。

本文选择第二种方式进行演示。

4.2、对缓存业务代码加一层封装

对于缓存服务，我们只关心对缓存数据进行查询，更新，删除等基本操作，不提供其他与缓存无关的业务代码，如下

public interface EntityCacheService<E extends BaseEntity<PK>, PK extends Serializable & Comparable<PK>> {


    /**
     * 根据id获取实体
     * @param id
     * @return
     */
    E getEntity(PK id);

    /**
     * 更新/插入实体
     * 若实体已存在于数据库，则执行更新操作；否则，执行插入操作
     * @param entity
     * @return
     */
    BaseEntity<PK> putEntity(E entity);

    /**
     * 移除实体
     * @param id
     * @return
     */
    default BaseEntity<PK> removeEntity(PK id) {
        throw new UnsupportedOperationException();
    }
}

其中，BaseEntity是实体记录，主要有以下方法

import lombok.Getter;
import lombok.Setter;
import org.springframework.data.annotation.Transient;

public interface BaseEntity<PK extends Serializable & Comparable<PK>> {

    /**
    * 实体唯一主键
    */
	<PK> PK getId();

    /**
    * 查询/设置删除状态
    */
	@Getter
    @Setter
    @Transient
    private boolean delete;

}

这两个方法非常重要，与异步持久化强相关，后文详述。

针对具体数据表的缓存操作，示例代码如下

@Service
public class PlayerCacheService implements EntityCacheService<Player> {

    @Autowired
    PlayerRepository playerRepository;

    @Cacheable(value = "player")
    @Override
    public Code get(String id) {
        Optional<Player> optional = playerRepository.findById(id);
        System.out.println("load from db");
        return optional.orElse(null);
    }

    @CachePut(value = "code", key="#entity.id")
    @Override
    public BaseEntity put(Player entity) {
        SpringContext.getDbService().saveToDb(entity);
        return entity;
    }

    @CacheEvict(value = "player")
    @Override
    public void remove(String id) {
        Optional<Player> optional = playerRepository.findById(id);
        optional.ifPresent(SpringContext.getDbService()::deleteFromDb);
    }

}

orm方案采用springdata jpa接口

@Repository
public interface PlayerRepository extends MongoRepository<Player, String> {


}

5、异步持久化

5.1、异步持久化机制

游戏里玩家数据的变动是非常频繁的，例如连续开100个道具，在战场杀怪刷经验等，如果玩家的每一个操作都持久化到数据库，无疑对数据库的压力非常大。因此，游戏服务器采用的是异步持久化。具体来说，异步持久化有以下三种策略。

• 基于队列：将所有需要持久化的实体进行排队，需要对重复插入的数据进行去重。
• 定时入库：以一定的频率周期性批量插入
• 延迟入库：对每一个实体，单独延迟XX时间后再入库

读者可根据需要，综合使用上面几种策略。

如果只使用异步持久化，或者只使用缓存，无法解决下面的问题。

• 游戏数据不仅仅频发读取，同时修改频率也非常高。如果只使用缓存，那么每次修改数据都要实时写入数据库，会导致数据库出现性能瓶颈。
• 如果只使用异步持久化，那么一旦重新从数据库读取数据，会造成“脏读”。即，异步持久化的数据还没真正保存，新的读取操作已经开始了，这时，读取的数据是过时数据。

只有缓存与异步持久化同时使用，才能碰撞出完美火花。对于玩家数据，一旦从数据库读取之后，便保存起来，下次读取不再操作数据库。（当然，对于沉默数据，设置失效时间，避免内存爆炸）。玩家的每次操作，只修改内存，再异步持久化到数据库。

5.2、异步持久化API

本文异步策略采用定时策略作为演示。持久化线程每隔XX毫秒持久化一波数据。

基本策略如下：

• 充分利用多核处理器的优势，使用线程组进行持久化。每个持久化容器保存一个更新队列。
• 持久化线程的run()方法是一个死循环，周期性取出数据，并进行持久化。
• 对于在同一个周期重复加入的实体数据进行去重，由于持久化容器统一处理不同的数据表，要求所有的实体记录id全局唯一（BaseEntity的方法getId()方法发挥作用）。最简单的，可以在每个实体的id前面该实体对应的表名。
• 充分利用orm工具的updateOrInsert机制，统一处理实体的插入/更新操作，而对于删除操作，增加一个标记字段。（BaseEntity的delete属性发挥作用）

异步持久化工具代码如下：

@Service
public class DbService {

    @Autowired
    private MongoTemplate mongoTemplate;

    private final AtomicBoolean run = new AtomicBoolean(true);

    private final int WORKER_CAPACITY = Math.max(4, Runtime.getRuntime().availableProcessors()) / 2;
    private Worker[] workers;

    @PostConstruct
    private void init() {
        workers = new Worker[WORKER_CAPACITY];
        NamedThreadFactory namedThreadFactory = new NamedThreadFactory("web-db-service");
        for (int i = 0; i < WORKER_CAPACITY; i++) {
            Worker worker = new Worker();
            workers[i] = worker;
            namedThreadFactory.newThread(worker).start();
        }
    }

    public void saveToDb(BaseEntity entity) {
        int index = Math.abs(entity.getId().hashCode()) % WORKER_CAPACITY;
        workers[index].addToQueue(entity);
    }

    public void deleteFromDb(BaseEntity entity) {
        entity.setDelete(true);
        saveToDb(entity);
    }

    public void shutDownGracefully() {
        for (int i = 0; i < workers.length; i++) {
            Worker worker = workers[i];
            worker.shutDown();
        }
    }

    @Override
    public String toString() {
        Map<Integer, Integer> data = new HashMap<>();
        for (int i = 0; i < WORKER_CAPACITY; i++) {
            Worker w = workers[i];
            data.put(i, w.queueSize());
        }
        return JsonUtil.object2String(data);
    }

    private class Worker implements Runnable {

        private Map<String, BaseEntity> data = new ConcurrentHashMap<>();

        void addToQueue(BaseEntity ent) {
            this.data.put(ent.getId(), ent);
        }

        @Override
        public void run() {
            while (run.get()) {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException ignore) {

                }
                if (data.isEmpty()) {
                    continue;
                }
                // 引用替换，转移数据
                Map<String, BaseEntity> image = data;
                this.data = new ConcurrentHashMap<>();
                image.forEach((key, value) -> {
                    try {
                        // 优先执行删除操作
                        if (value.isDelete()) {
                            mongoTemplate.remove(value);
                        } else {
                            mongoTemplate.save(value);
                        }
                    } catch (Exception exception) {
                        LoggerUtil.error("", exception);
                    }
                });
            }
        }

        void shutDown() {
            data.forEach((key, value) -> {
                try {
                    saveToDb(value);
                } catch (Exception exception) {
                    LoggerUtil.error("", exception);
                }
            });
        }

        public int queueSize() {
            return data.size();
        }

    }
}