Java游戏服业务线程模型一

肩负重担的感觉,就是一千个人全都翘首以盼,你说的每一个字。 --刀妹

Java游戏服的业务线程模型，是指从网络层接收到客户端的协议请求后，采用多线程（线程池）的方式，并行处理客户端的请求，以达到高效并发处理游戏业务的目的。

这时有人可能会想，随便整个线程池做这个事不就行了吗？

传统的线程池有以下几种：
newSingleThreadExecutor，
newFixedThreadPool，
newCachedThreadPool，
newWorkStealingPool，
newScheduledThreadPool，

newSingleThreadExecutor 单线程池，游戏中通常很少直接new一个Thread出来去做事情的（但是直接new出来也无可厚非），优雅一点的写法就是定义一个newSingleThreadExecutor去做，比如用它在游戏服运行时去做重新加载配置，就可以新开或用原有的一个单线程去做这事。

newFixedThreadPool，newCachedThreadPool 在游戏中用得比较多，用它们做游戏的业务线程模型是可以的，但是，使用时要注意它们的区别：
1）从线程数量上来说，newFixedThreadPool是固定数量的，newCachedThreadPool是可能最多达到Integer.MAX_VALUE数量的，而线程在应用中是一种稀缺的资源，在高并发请求下，如果使用newCachedThreadPool线程模型，是有可能一下创建非常多的线程数量的，而线程切换又会带来性能损耗，因此在游戏中用newFixedThreadPool会相对多一点。
2）从线程回收来说，newFixedThreadPool生成后的线程是不会被回收的，而newCachedThreadPool生成后的线程是会被回收的，默认为60s，60s空闲后即会被回收。
3）从任务队列来说，newFixedThreadPool默认使用的是LinkedBlockingQueue，是基于链表的阻塞队列，内部由两个ReentrantLock来实现出入队列的线程安全，默认大小为Integer.MAX_VALUE，当消息的增加速度大于消息的处理速度时，要小心内存可能溢出的情况。newCachedThreadPool采用的是SynchronousQueue同步队列，一个不存储元素的阻塞队列，有消息来了，如果线程没被回收，则用原有空闲线程执行，否则新开线程执行。

newFixedThreadPool因其线程数量固定，又不会频繁创建线程，因此在游戏中常用来做业务处理线程池，newCachedThreadPool在游戏中通常用于在明确知道不会一下创建很多个线程，但是又最好能有多个线程处理的情况，比如游戏中有时要处理平台的http请求，这类的请求不会同时很多且频繁的，就可以采用newCachedThreadPool来做，空闲了线程也被回收了，忙起来也不会开很多个。

newScheduledThreadPool 定时线程池，有一些定时操作要做的，常用到它，比如数据存储线程池，游戏中玩家可以每5分钟批量保存他已更新的数据，或下线超过半个小时把他的数据移出缓存，就可以用newScheduledThreadPool来做。

newWorkStealingPool JDK1.8新增的，这个什么窃取线程的，在游戏中用得不多，大家自行百度其用法。

我们再看线程池的处理逻辑：
newSingleThreadExecutor，newFixedThreadPool，newCachedThreadPool，newScheduledThreadPool其实都是对ThreadPoolExecutor的包装，在源码中最终都会来到这里，newSingleThreadExecutor，newFixedThreadPool，newCachedThreadPool，newScheduledThreadPool只是各个使用的参数不同而已：

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler ) 

线程池的处理逻辑如下图：

线程池处理逻辑（此图借用别人的，侵删）.png

上图中在使用不同线程池时有几个重要注意的点：
1）处理逻辑可概括为先判断核心线程，再判断任务队列是否已满，再判断最大线程是否已满，最后再判断拒绝策略；
2） 队列是否已满时，要注意使用的队列是不是无界队列，如果是无界队列，那它是永远也不会满的，但可能会内存溢出的，如果队列不会满，那么就只会用核心线程跑了。线程不会再增了，拒绝策略也没用了。此外，使用无界队列，一定要注意游戏业务不要有很耗时的操作，不然无界队列消息堆积太多，可能导致内存溢出，或导致其他玩家难以进入游戏，因为登录的消息堆积在消息队列中未能得到及时处理。

Executors.newSingleThreadScheduledExecutor()这个游戏中也用得多，用它可开启一个线程定时或定频做多次，或只执行一次的操作。

好，背景知识介绍得差不多了，现在我们再回到文初的那个问题，游戏业务线程池可以随便使用newFixedThreadPool，newCachedThreadPool中的一个吗？

我们可以想象一下现实中的大型网游情景，成百上千的客户端同时连接着游戏服，游戏服在极小的时间段内可能有成千上万的协议请求，一个个请求封装成Runnable task给上述两个线程池去执行，这些Runnable task要么正在被某个线程处理，要么增加到任务队列后面等待处理……

那问题来了，

如果用上述两种线程池，同一个玩家的多个协议请求就有可能被多个线程处理的（如上图player1的agreeReq_1001请求可能被线程池的1号线程处理，deleteAllReq_1002请求可能被线程池的2号线程处理），这多个线程又有可能有的执行快，有的执行慢，就是有可能后面执行的线程反而比前面的线程更快执行完。这样，就导致了玩家的数据可能是不安全的。比如我先发同意好友申请的协议，再发删除所有好友的协议，理论上最终是一个好友都没有了，玩家预期也是这样的。但如果这两条协议，删除所有好友的“任务”被线程先执行完，同意好友申请的“任务”被另一个线程后执行完，结果就变成了还有个好友在我的好友列表里，这与玩家的预期是不一样的，这就导致了数据不安全。同样地，如果不加锁，两个线程同时获取同时处理一条数据，还会产生数据覆盖的可能性，这是极大的安全隐患。

那么有人可能又会说，我加锁不就行了吗？
一、游戏中很多的业务数据实际都是个人的数据操作，而游戏中又有很多的玩法数据，如果每个数据每个字段操作时都加锁，肯定影响效率。
二、就算加了锁，上述好友的操作问题仍会发生，因为可能先删好友的线程先得到锁先执行，结果仍不是我们想要的。

所以，游戏中的业务线程模型也并不是简单使用newFixedThreadPool，newCachedThreadPool那么简单！
它们的使用是有前提的，前提是保障玩家的数据安全及正确！

这便有了游戏服的业务线程模型设计。

玩家的游戏操作都是有序的，那么它们的操作处理也需要是有序的。如果打破这个有序，就可能引发数据安全问题。那么我们应该如何保障这个有序，保障数据安全呢？

用线程池处理时，玩家的请求消息本来就是放在任务队列里的，这本来就是有序的。但是，我们还要保证它们的执行是有序的，分先后的，针对单个玩家的消息，不能后面的先执行完，而必须前面的先执行完后面的才执行。

解决方案有多个，
1）维护一个总的消息队列，所有玩家的消息都放入这个队列，但是同一个玩家的消息总是由一个相同的线程去执行，根据玩家id取模线程池大小即可，在单线程执行时，所有的消息只能一个一个接着来了。但是这样手气不好的时候，有可能有的线程会很累，有的线程会很闲。

2）每个玩家都维护一个消息队列，任何一个玩家的消息队列有数据时，都会notify通知线程池的任意线程wakeUp去处理消息，单个线程一次处理完个人队列里所有的消息。

3）维护一个总的消息队列，所有玩家的消息都放入这个队列，有任何玩家消息来的时候，消息和玩家的session绑定放入这个队列，线程池中的任意线程得到消息执行时，先试着对这个session加锁（session里维护一个volatile的是否锁定状态标志），如果session已是锁定状态，说明有线程正在处理这个玩家的协议，那么这个玩家新来的消息将放在消息队列后面，否则认为没有线程正在处理该玩家的消息，该线程立即处理该玩家的消息。而传统的线程池是不支持这样的，因此这种方案还需 自定义线程池 实现。
4）……

在《 游戏架构方案 》一文中已提到过第三种方案，现在就细讲这种方案。其余的方案后续博文可能会再补其一二。

大家可先大致看下JDK中线程池ThreadPoolExecutor的实现，看下它继承的父类，它的属性及方法定义，然后结合线程池的逻辑，我们可仿照ThreadPoolExecutor实现如下自定义线程池：

public final class OrderedExecutorService extends AbstractExecutorService {
    private final int corePoolSize;//核心线程数
    private final int maxPoolSize;//最大线程数

    private final LinkedList<Runnable> tasks = new LinkedList<>(); //任务队列/消息队列

    private final Lock lock = new ReentrantLock(); //用于对消息队列加锁
    private final Condition cond = lock.newCondition();
    private final int keepAliveTime;//线程活跃时间限制
    private int active;//激活的线程数
    private final ThreadFactory factory;//线程工厂

    private volatile boolean stopped = false;//是否已停止
    private volatile boolean forceStopped = false;

    public OrderedExecutorService(int corePoolSize, int maxPoolSize, int keepAliveTime, ThreadFactory factory) {
        this.corePoolSize = corePoolSize;
        this.maxPoolSize = maxPoolSize;
        this.keepAliveTime = keepAliveTime;
        this.factory = factory;
    }
    ……
}

当在网络层收到客户端的消息请求时，

    @Override
    public void onReceive(IConnection conn, Packet packet) {
        PlayerSession<K> session = c2p.get(conn);
        if (session == null) {
            return;
        }
        pool.execute(new OrderedTask(session, packet) {//自定义的线程池
            @Override
            public void run() {
                try {
                    handlerManager.forward(getSession(), getPacket());//各模块处理消息
                } catch (Throwable t) {
                    onException(session, new HandleException(packet.getCmd(), t));
                }
            }
        });
    }

在OrderedTask里，会把packet和session结合，

public abstract class OrderedTask implements OrderedRunnable{
    private final Packet packet;
    private final PlayerSession<?> session;

    protected OrderedTask(PlayerSession<?> session, Packet packet) {
        this.packet = packet;
        this.session = session;
    }
}

此后，在线程池中线程执行时，

@Override
    public void execute(Runnable command) {
        lock.lock();
        try {
            if (stopped)
                return;
            if (active < corePoolSize) {
                if (((OrderedRunnable) command).tryLock()) {//如果该session能够锁定，说明该玩家没有正在处理的消息
                    addWorker(command, true);
                } else {//如果该session已是锁定状态，说明正在处理该玩家的消息，他的消息需要放在消息队列里等待正在处理的消息完成了才会得到处理
                    tasks.addLast(command);
                    cond.signal();
                }
            } else if (active < maxPoolSize) {
                if (((OrderedRunnable) command).tryLock()) {
                    addWorker(command, false);
                } else {
                    tasks.addLast(command);
                    cond.signal();
                }
            } else {
                tasks.addLast(command);
                cond.signal();
            }
        } finally {
            lock.unlock();
        }
    }

session中定义一个是否锁定的状态标志，

volatile boolean locked = false;

通过判断该玩家的session是否能够锁定，如果能锁，说明该玩家没有正在处理的消息，则可以马上帮他处理这条消息；否则，说明正在处理该玩家的消息，他的消息需要放在消息队列里等待正在处理的消息完成了才会得到处理。

addWorker方法：

    private void addWorker(Runnable task, boolean core) {
        active++;
        factory.newThread(new Worker(task, core)).start();
    }

worker的run方法：

public void run() {
    Runnable task = firstTask;
    firstTask = null;
    for (;;) {
        if (task == null)
            task = getTask();
        if (task == null) {
            break;
        } else {
            task.run();//在这里处理业务消息
            lock.lock();
            ((OrderedRunnable) task).unlock();
            cond.signal();
            lock.unlock();
            task = null;
        }
    }
    lock.lock();
    active--;
    cond.signalAll();
    lock.unlock();
}

如此，就实现了解决方案3）中的业务线程模型，因为要实现玩家同一时刻只能处理他的一条协议，所以这里采用了自定义的线程池，以保障数据安全。