异步持久化策略对比

1.背景

游戏服务器其中一项重点工作，就是对游戏玩家的数据进行持久化，保证下次登录可以再续前缘。如果游戏服务器架构里没有缓存，每次操作都需要读写数据库，无疑对数据库带来非常大的压力。一旦使用缓存，就伴随异常持久化的需求。也就是说，玩家数据先写入缓存，然后服务器在稍后的某个时候，再把数据回写到数据库。

这里的“某个时候”，就有几种策略。

排队策略：采用“先进先出”的策略，排在前面的数据先入库，多次操作去重。

延迟策略：先拿个号，XX时间后，再入库，期间的多次操作直接进行数据合并。

定时策略：基于cron表达，例如每隔五分钟，将排队的所有数据批量更新。

2.代码实现

2.1.实体接口

说实在，加这个接口多少带点无奈，因为加了就强迫用户的领域类实现该接口，对用户的现有代码造成了入侵。但是不加这个，又无法解决一个难题。当用户引入了Redis等进程外缓，每次从Redis读取的数据都是重新序列化的。持久化容器没办法判断前后两个对象是不是同一个对象。

所以这个接口的唯一目的，就是提供一个getId()接口，以便容器能够去重。

import java.io.Serializable;public interface Entity<ID extends Serializable & Comparable<ID>> {ID getId();default String getKey() {return getClass().getSimpleName() + "@" + getId().toString();}}

2.2.持久化容器接口

容器接收每一个需要持久化的对象，然后按自己的策略进行入库，需要一个接收的接口。每个项目使用不同的orm工具库，不同的数据库，如何保存对象这个功能也要作为接口定义。最后，为了支持完美停机，保证服务关闭之前，将所有等待数据进行批量操作，需要另外一个接口。总体，设计如下:

public interface PersistContainer {/*** 接收实体* @param entity*/void receive(Entity<?> entity);/*** 优雅退出，会保证所有等待队列的数据都会跑完*/void shutdownGraceful();SavingStrategy getSavingStrategy();
}

/*** 持久化策略*/
public interface SavingStrategy {/*** 真正执行入库工作，* @param entity*/void doSave(Entity<?> entity);
}

2.3.排队容器

该策略是一种典型的生产者消费者模型，所有的持久化元素会不断的排队，同时持久化线程会不断的消费。如果数据更新非常频繁，那么线程会一直工作，直到消费队伍为空。先来看一段有问题的代码

/*** 以队列的形式持久化*/
public class QueueContainer implements PersistContainer {private final Logger logger = LoggerFactory.getLogger("DbContainer");private final AtomicBoolean run = new AtomicBoolean(true);private final String name;private final BlockingQueue<Entity<?>> queue = new LinkedBlockingDeque<>();private final Set<String> savingQueue = new ConcurrentHashSet<>();private static final NamedThreadFactory namedThreadFactory = new NamedThreadFactory("jforgame-persist-queue-thread");private SavingStrategy savingStrategy;public QueueContainer(String name, SavingStrategy savingStrategy) {// 启动线程跑namedThreadFactory.newThread(this::run).start();this.name = name;this.savingStrategy = savingStrategy;this.run();}@Overridepublic void receive(Entity<?> entity) {String key = entity.getKey();// t2时刻线程a执行if (savingQueue.contains(key)) {return;}this.queue.add(entity);}private void run() {while (run.get()) {Entity<?> entity = null;try {entity = queue.take();savingStrategy.doSave(entity);  // t1时刻线程b执行savingQueue.remove(entity.getKey()); // t3时刻线程b执行} catch (ConcurrentModificationException e1) {// 有可能是并发抛错，重新放入队列receive(entity);} catch (Exception e) {// 其他异常，重复放入队列，还是无法解决问题，有问题的终究有问题// 这里需要一个强通知操作，例如通知开发人员，起来修bug了logger.error("", e);}}}@Overridepublic void shutdownGraceful() {run.compareAndSet(true, false);for (; ; ) {if (!queue.isEmpty()) {saveAllBeforeShutDown();} else {break;}}logger.info("db container [{}] close ok", name);}@Overridepublic SavingStrategy getSavingStrategy() {return savingStrategy;}private void saveAllBeforeShutDown() {while (!queue.isEmpty()) {Iterator<Entity<?>> it = queue.iterator();while (it.hasNext()) {Entity<?> ent = it.next();it.remove();savingStrategy.doSave(ent);}}}
}

这里有个线程并发问题，当持久化线程执行完  savingStrategy.doSave(entity); 这句代码的时候，还没来得及执行savingQueue.remove(entity.getKey())的时候，业务执行执行了receive()方法，就会导致此次保存的动作消失了。

解决的方法也有点意思，只需要消费的时候，代码的顺序调整下即可

savingQueue.remove(entity.getKey()); 
savingStrategy.doSave(entity);  

入库的时候，只要获取到对象，立马释放savingQueue，保证业务线程可以重新加入队列。

2.4.延迟容器 

该策略当一个记录想要入库的时候，先让它等一等，想一想，还有没有其他数据需要合并，不管有没有，隔XX时间一定执行保存动作。

/*** 以延迟执行的形式持久化*/
public class DelayContainer implements PersistContainer {private static final Logger logger = LoggerFactory.getLogger(DelayContainer.class);private static final ScheduledExecutorService service = Executors.newScheduledThreadPool(1, new NamedThreadFactory("common-scheduler"));private final Set<String> savingQueue = new ConcurrentHashSet<>();private ConcurrentMap<String, Node> queue = new ConcurrentHashMap<>();private final AtomicBoolean run = new AtomicBoolean(true);private final String name;/*** 延迟秒数*/private final int delaySeconds;private SavingStrategy savingStrategy;public DelayContainer(String name, int delaySeconds, SavingStrategy savingStrategy) {this.name = name;this.delaySeconds = delaySeconds;this.savingStrategy = savingStrategy;}public void receive(Entity<?> entity) {if (!run.get()) {// 小店已经打烊了，恕不招待return;}String key = entity.getKey();if (savingQueue.contains(key)) {return;}Runnable task = () -> {savingStrategy.doSave(entity);queue.remove(key);};Node node = new Node();node.key = key;node.task = task;queue.put(key, node);service.schedule(node, delaySeconds, TimeUnit.SECONDS);}@Overridepublic void shutdownGraceful() {run.compareAndSet(true, false);Iterator<Map.Entry<String, Node>> iterator = queue.entrySet().iterator();while (iterator.hasNext()) {Map.Entry<String, Node> next = iterator.next();// 直接执行next.getValue().task.run();iterator.remove();}logger.info("db container [{}] close ok", name);}@Overridepublic SavingStrategy getSavingStrategy() {return savingStrategy;}private static class Node implements Runnable {String key;Runnable task;@Overridepublic void run() {task.run();}}}

2.5.周期策略

该策略不针对一个一个对象，而是周期性的批量处理所有对象，例如每隔一分钟。可以借助Quartz这个第三方库，或者spring的cron表达式，执行复杂的周期性任务。

import org.quartz.CronScheduleBuilder;
import org.quartz.Job;
import org.quartz.JobBuilder;
import org.quartz.JobDetail;
import org.quartz.JobExecutionContext;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.Trigger;
import org.quartz.TriggerBuilder;
import org.quartz.impl.StdSchedulerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;public class CronContainer implements PersistContainer {private static final Logger logger = LoggerFactory.getLogger(CronContainer.class);private final AtomicBoolean run = new AtomicBoolean(true);private final String name;// 存储待持久化的实体，以实体的唯一标识作为键private  ConcurrentHashMap<String, Entity<?>> entityQueue = new ConcurrentHashMap<>();// Quartz调度器private final Scheduler scheduler;private SavingStrategy savingStrategy;public CronContainer(String name, String cronExpression, SavingStrategy savingStrategy) {this.name = name;this.savingStrategy = savingStrategy;try {// 创建Quartz调度器实例并启动scheduler = StdSchedulerFactory.getDefaultScheduler();scheduler.start();// 创建定时任务并注册到调度器中，使用传入的cronExpression配置触发规则JobDetail jobDetail = JobBuilder.newJob(CronPersistJob.class).withIdentity("cronPersistJob", name).build();Trigger trigger = TriggerBuilder.newTrigger().withIdentity("cronTrigger", name).withSchedule(CronScheduleBuilder.cronSchedule(cronExpression)).build();scheduler.scheduleJob(jobDetail, trigger);// 将当前CronContainer实例存入调度器上下文，方便在任务执行时获取scheduler.getContext().put("cronContainer", this);} catch (SchedulerException e) {logger.error("Failed to initialize scheduler for CronContainer [{}]", name, e);throw new RuntimeException("Failed to initialize scheduler", e);}}@Overridepublic void receive(Entity<?> entity) {if (!run.get()) {return;}String key = entity.getKey();entityQueue.put(key, entity);}@Overridepublic void shutdownGraceful() {run.compareAndSet(true, false);try {// 关闭调度器，停止定时任务scheduler.shutdown(true);logger.info("Cron container [{}] close ok", name);} catch (SchedulerException e) {logger.error("Failed to shutdown scheduler for CronContainer [{}]", name, e);}}@Overridepublic SavingStrategy getSavingStrategy() {return savingStrategy;}// 内部类实现Job接口，定义定时任务执行时的逻辑，会按照cronExpression配置的规则触发public static class CronPersistJob implements Job {@Overridepublic void execute(JobExecutionContext context) {try {CronContainer container = (CronContainer) context.getScheduler().getContext().get("cronContainer");if (container != null && container.run.get()) {// 获取当前队列中所有待持久化的实体Map<String, Entity<?>> entities = container.entityQueue;// 重置引用，数据隔离container.entityQueue = new ConcurrentHashMap<>();Set<String> keys = entities.keySet();for (String key : keys) {Entity<?> entity = entities.get(key);if (entity != null) {// 对每个实体执行持久化操作container.savingStrategy.doSave(entity);}}}} catch (SchedulerException e) {logger.error("Error occurred during executing cron job for CronContainer", e);}}}
}

2.6.组合策略

当一种策略不满足需求的时候，我们可以考虑以组合的形式。例如，如果觉得玩家表一个线程跑不过来，我们也可以加多几个，形成一个容器组，类似下面的代码。

public class QueueContainerGroup extends QueueContainer {private QueueContainer[] group;public QueueContainerGroup(String name, SavingStrategy savingStrategy, int workers) {super(name, savingStrategy);group = new QueueContainer[workers];for (int i = 0; i < workers; i++) {QueueContainer work = new QueueContainer(name, savingStrategy);group[i] = work;}}@Overridepublic void receive(Entity<?> entity) {int index = entity.getId().hashCode() % group.length;group[index].receive(entity);}}

2.7.示例代码

/*** 异步持久化服务*/
@Service
public class AsyncDbService {private SavingStrategy savingStrategy = new MySavingStrategy();private PersistContainer playerWorker = new CronContainer("player", "0 0/1 * * * ?", savingStrategy);private PersistContainer commonWorker = new DelayContainer("common", 60, savingStrategy);/*** 调用此方法，将需要持久化的对象放入队列** @param entity*/public void saveToDb(BaseEntity<?> entity) {if (entity instanceof PlayerEnt) {playerWorker.receive(entity);} else {commonWorker.receive(entity);}}public void deleteFromDb(BaseEntity<?> entity) {entity.setDelete(true);saveToDb(entity);}public void shutDown() {playerWorker.shutdownGraceful();commonWorker.shutdownGraceful();}}

public class MySavingStrategy implements SavingStrategy {@SuppressWarnings("all")@Overridepublic void doSave(Entity<?> entity) {BaseEntity baseEntity = (BaseEntity) entity;// 优先处理删除操作if (baseEntity.isDelete()) {baseEntity.getCrudRepository().delete(entity);} else {baseEntity.getCrudRepository().save(entity);}}
}