游戏服务器业务处理线程管理

在游戏服务器的框架设计中，最重要的就是管理业务逻辑处理。当收到客户端的请求消息时，服务器如何辨认这个消息是什么请求，怎么样保证同一个用户请求处理的顺序性？怎么样提高并发性？这些都是在底层框架中要解决的问题。这里分享一种做法，有用者取之。

1，定义消息Id

给客户端与服务器交互的消息定义一个唯一的消息id,通过消息的id，处理对应的业务请求，比如1001 代表登陆，1002 代表签到，1003代表抽卡，等。

2，管理消息Id与处理方法的映射

当服务器解析出客户端的请求拿到消息号之后，怎么做相应处理呢？有的是这样做的：

public void dispatch(int id, JSONObject param, long userId) { if (id == 1001) { // 调用相关的处理方法
        } else if (id == 1002) { // 调用相关的处理方法
        } else if (id == 1003) { // 调用相关的处理方法
 } // 或者
        switch (id) { case 1001: // 调用相关的处理方法
                break; case 1002: // 调用相关的处理方法
                break; case 1003: // //调用相关的处理方法
                break; default: break; } }

这两种方法不是不可以，如果是请求不多还行，如果大一点的游戏，几十上百个请求，一个类中几百行if或switch，看起来就心烦，再加上团队间协作开发，多人修改导致提交git冲突等问题。

一种解决方法是，设计一个统一的接口，每个请求对应一个接口，把id和接口映射起来，放到一个map中。例如：

public class LogicManager { Map<Integer, IMessageProcess> messageMap = new HashMap<>(); public void init() { messageMap.put(1001, new LoginProcess()); //以下省略很多
 } public void dispatch(int id, JSONObject param, long userId) { IMessageProcess process = messageMap.get(id); if(process != null) { process.action(userId, param); } } }

这种方式只需要每增加一个消息，增加一个接口实现即可，但是有个问题是每次都需要手动添加消息号和接口的映射，很不爽，我们可以使用注解的反射解决这个问题。

我的项目是使用spring boot管理的，所以使用了spring的特性，从容器中获取实例，因为spring已经在启动的时候扫描了所有注解。

先看定义的注解:

import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; import org.springframework.stereotype.Service; import com.xinyue.interview.logiccontroller.RequestCode; @Target({ElementType.TYPE, ElementType.METHOD}) @Retention(RetentionPolicy.RUNTIME) @Service public @interface LogicRequestMapping { public RequestCode value() default RequestCode.DEFAULT; }

这个注解具体有spring @Service的特性。RequestCode是定义的消息号枚举，把它添加到接口的实现类上面。

@LogicRequestMapping public class LoginProcess implements IMessageProcess{ @Override public void action(long userId, JSONObject param) { } }

 

然后在服务器启动的时候，使用程序自动映射消息号和接口

package com.netty; import java.util.HashMap; import java.util.Map; import javax.annotation.PostConstruct; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.stereotype.Service; import com.alibaba.fastjson.JSONObject; @Service public class LogicManager { Map<Integer, IMessageProcess> messageMap = new HashMap<>(); @Autowired private ApplicationContext context; @PostConstruct public void init() { Map<String, Object> beanMap = context.getBeansWithAnnotation(LogicRequestMapping.class); beanMap.forEach((k,v)->{ LogicRequestMapping logicRequestMapping = v.getClass().getAnnotation(LogicRequestMapping.class); if(logicRequestMapping != null && v instanceof IMessageProcess) { int id = logicRequestMapping.value().getCode(); messageMap.put(id, (IMessageProcess) v); } }); } public void dispatch(int id, JSONObject param, long userId) { IMessageProcess process = messageMap.get(id); if (process != null) { process.action(userId, param); } } }

这样增加新请求的时候，只需要增加新接口实现即可，而旧的代码不需要修改，也没有长长的if和switch判断。如果觉得RequestCode也需要修改，可以去掉，换成接口，每个人可以定义自己的消息枚举。

如果觉得一个接口处理一个消息有点浪费的话，也可以像web的controller那样，把处理定位到方法上面。即一个方法对应一个处理的消息id.服务器启动的时候缓存方法：

package com.xinyue.interview.dispatch; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.util.HashMap; import java.util.Map; import javax.annotation.PostConstruct; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.stereotype.Service; import com.alibaba.fastjson.JSONObject; import com.xinyue.interview.controller.protocol.LogicMessage; import com.xinyue.interview.exception.ServerError; import com.xinyue.interview.logiccontroller.RequestCode; import com.xinyue.interview.utils.StringUtil; @Service public class LogicRequestMappingManager { private Logger logger = LoggerFactory.getLogger(LogicRequestMappingManager.class); private Map<Integer, LogicRequestMappingInfo> requestMap = new HashMap<>(); @Autowired private ApplicationContext context; @PostConstruct public void init() { Map<String, Object> beanMap = context.getBeansWithAnnotation(LogicRequestMapping.class); beanMap.forEach((k, v) -> { Method[] methods = v.getClass().getMethods(); for (Method method : methods) { LogicRequestMapping logicRequestMapping = method.getAnnotation(LogicRequestMapping.class); if (logicRequestMapping != null) { RequestCode requestCode = logicRequestMapping.value(); if(requestCode == null) { throw new IllegalArgumentException("请求消息标记为空"); } int commandId = requestCode.getCode(); if (requestMap.containsKey(commandId)) { String msg = StringUtil.format("业务处理方法的id有重复的，重复id:{},所在类：{}，方法名：{}", commandId, v.getClass().getName(), method.getName()); throw new IllegalArgumentException(msg); } LogicRequestMappingInfo logicRequestMappingInfo = new LogicRequestMappingInfo(commandId, v, method); requestMap.put(commandId, logicRequestMappingInfo); } } }); } public void callMethod(Integer commandId, LogicRequestContext ctx, JSONObject params) { LogicRequestMappingInfo requestMappingInfo = requestMap.get(commandId); if (requestMappingInfo == null) { LogicMessage logicMessage = new LogicMessage(ServerError.REQUEST_PARAM_ERROR); ctx.writeAndflush(logicMessage); logger.error("用户{}请求的消息id:{}不存在", ctx.getUserId(), commandId); return; } try { requestMappingInfo.call(ctx, params); } catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) { logger.error("系统异常", e); LogicMessage logicMessage = new LogicMessage(ServerError.SYSTEM_ERROR); ctx.writeAndflush(logicMessage); } } }

 

3，管理业务线程

 

要保证同一个用户的消息处理都是顺序性的，一般有几种方式，1，加锁，2，使用队列，生产者消费者模式，3，在同一个线程中处理。使用锁，在高并发的时候导致cpu上下文切换频繁，性能下降，我们可以采用netty的方式，同一个连接的所有消息都在同一个线程中处理。一服务器处理的线程并不是越多越好，见：https://www.cnblogs.com/wgslucky/p/9749990.html  所以在处理业务的时候最好不要有io操作。那不可避免的io操作，比如查数据库，更新数据，我们放在别的单独线程处理。这里先说业务处理，它操作的都是内存中的数据，所以速度很快，不会卡玩家。

首先创建一个线程处理器的类，用于管业务线程数：

package com.xinyue.interview.utils.concurrent; import io.netty.util.concurrent.DefaultEventExecutor; import io.netty.util.concurrent.EventExecutor; public class LogicEventExecutorGroup { private EventExecutor[] executors; public LogicEventExecutorGroup(int nthreads) { executors = new EventExecutor[nthreads]; for (int i = 0; i < nthreads; i++) { executors[i] = new DefaultEventExecutor(); } } public EventExecutor select(long userId) { int index = (int) (userId % executors.length); return executors[index]; } }

这里根据用户的userId，取余固定一个线程处理器，所以同一个用户的请求都放到这个线程处理器中。

在收到消息后：

package com.xinyue.interview.dispatch; import java.util.List; import java.util.concurrent.TimeUnit; import javax.annotation.PostConstruct; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import com.alibaba.fastjson.JSONObject; import com.xinyue.interview.controller.protocol.HeadInfo; import com.xinyue.interview.controller.protocol.LogicMessage; import com.xinyue.interview.exception.ServerError; import com.xinyue.interview.exception.ServerErrorException; import com.xinyue.interview.logic.manager.AccountManager; import com.xinyue.interview.logic.manager.EntityManagerFactory; import com.xinyue.interview.logic.manager.IEntityManager; import com.xinyue.interview.utils.concurrent.LogicEventExecutorGroup; import io.netty.util.concurrent.EventExecutor; import io.netty.util.concurrent.Future; import io.netty.util.concurrent.Promise; @Service public class LogicRequestDispatch { private Logger logger = LoggerFactory.getLogger(LogicRequestDispatch.class); @Autowired private LogicRequestMappingManager logicRequestMappingManager; private LogicEventExecutorGroup executorGroup; @PostConstruct public void init() { //初始化4个线程处理器处理业务逻辑
        int nthreads = 4; this.executorGroup = new LogicEventExecutorGroup(nthreads); } public void fireRead(LogicRequestContext ctx, JSONObject param) { long userId = ctx.getUserId(); EventExecutor executor = executorGroup.select(userId); executor.execute(() -> { try { // 检测是否登陆成功
                this.checkLogin(ctx.getHeadInfo()); // 调用业务方法
                Integer commandId = ctx.getCommandId(); logicRequestMappingManager.callMethod(commandId, ctx, param); } catch (Throwable e) { LogicMessage response = null; if (e instanceof ServerErrorException) { ServerErrorException errorException = (ServerErrorException) e; response = new LogicMessage(errorException); } else { response = new LogicMessage(ServerError.SYSTEM_ERROR); } ctx.writeAndflush(response); } }); } private void checkLogin(HeadInfo headInfo) { long userId = headInfo.getUserId(); AccountManager accountManager = EntityManagerFactory.getManager(userId, AccountManager.class); if (accountManager == null) { ServerError.throwError(ServerError.USER_NOT_LOGIN, "userId:{}", userId); return; } String token = headInfo.getToken(); if (!accountManager.checkToken(token)) { ServerError.throwError(ServerError.USER_NOT_LOGIN, "userId:{},token错误", userId); return; } } /** * * <p> * Description:初始化相应的数据管理类到对应的逻辑线程中 * </p> * * @param userId * @param managers * @author wgs * @date 2018年10月19日 下午4:39:31 * */
    public void initEntityManager(long userId, List<? extends IEntityManager> managers) { EventExecutor executor = executorGroup.select(userId); executor.execute(() -> { try { EntityManagerFactory.init(userId, managers); } catch (Exception e) { logger.error("初始化EntityManager出错", e); } }); executor.scheduleAtFixedRate(() -> { EntityManagerFactory.flushDB(); }, 5, 5, TimeUnit.MINUTES); } public Future<Object> fireEvent(long selectKey, Object param, Promise<Object> promise) { return promise; } }

这样就实现了业务处理保证在同一个线程中。

欢迎加群交流，QQ群：66728073，197321069，398808948 还可以扫描博客左上角二维码，关注游戏技术网公众号。