示例工程代码
可从附件下载
具体的说明和用法在后面介绍
需求与目的
一个游戏服务端需要处理各种业务逻辑,每一种业务逻辑都对应着一个请求消息和一个响应消息。那么服务端需要把这些不同的消息自动分发到对应的业务逻辑中处理。
最简单的处理方式就是根据请求消息中的type字段,使用switch case来进行分别处理,但这种方式随着消息的增多,显现了一些坏味道:长长的一大坨不太好看;如果要添加新的消息、新的逻辑,或者去掉新的消息、新的逻辑,在代码上不但要修改这些消息和逻辑,还不得不修改这长长的一坨swtich case,这样的修改显得很多余。
所以我们的目的就是把消息分发这块的代码自动化,在增加、修改、删除消息和逻辑的时候不需要再对消息分发的代码再做修改,从而使得修改的代码最小化。
实现原理
在实现中,使用了注解(annotation)
package com.company.game.dispatcher.annotation; import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; /** * 修饰消息类和业务逻辑执行类 * msgType指定对应的类型,从1开始计数 * @author xingchencheng * */ @Target(ElementType.TYPE) @Retention(RetentionPolicy.RUNTIME) public @interface UserMsgAndExecAnnotation { short msgType(); }
唯一的字段msgType代表了消息类型,这是客户端与服务端的约定,这里我们从1开始计数。
当我们要增加一个加法消息时,就使用这个注解来修饰我们的请求消息类:
package com.company.game.dispatcher.msg; import com.company.game.dispatcher.annotation.UserMsgAndExecAnnotation; /** * 加法请求消息类 * * @author xingchencheng * */ @UserMsgAndExecAnnotation(msgType = MsgType.ADD) public class UserAddRequest extends RequestMsgBase { private double leftNumber; private double RightNumber; public UserAddRequest() { super(MsgType.ADD); } public double getLeftNumber() { return leftNumber; } public void setLeftNumber(double leftNumber) { this.leftNumber = leftNumber; } public double getRightNumber() { return RightNumber; } public void setRightNumber(double rightNumber) { RightNumber = rightNumber; } }
为什么要这样修饰呢?先从服务端的解码(decode)说起,实例代码中,一个请求消息是这样规定的:
0-1字节表示整个消息的长度(单位:字节)
2-3字节代表消息类型,对应annotation的msgType
余下的是消息的json字符串(UTF-8编码)
我们需要根据2-3字节表示的msgType得到对应请求消息类的class对象,用这个class对象来序列化json字符串,得到具体的请求对象。那么怎么根据msgType得到class对象呢?这就是为什么要使用annotation的原因。
在服务端程序启动前,会执行下面的处理:
// msgType->请求、响应类的class对象 private static Map<Short, Class<?>> typeToMsgClassMap; // 根据类型得到对应的消息类的class对象 public static Class<?> getMsgClassByType(short type) { return typeToMsgClassMap.get(type); } /** * 初始化typeToMsgClassMap * 遍历包com.company.game.dispatcher.msg * 取得消息类的class文件 * * @throws ClassNotFoundException * @throws IOException */ public static void initTypeToMsgClassMap() throws ClassNotFoundException, IOException { Map<Short, Class<?>> tmpMap = new HashMap<Short, Class<?>>(); Set<Class<?>> classSet = getClasses("com.company.game.dispatcher.msg"); if (classSet != null) { for (Class<?> clazz : classSet) { if (clazz.isAnnotationPresent(UserMsgAndExecAnnotation.class)) { UserMsgAndExecAnnotation annotation = clazz.getAnnotation(UserMsgAndExecAnnotation.class); tmpMap.put(annotation.msgType(), clazz); } } } typeToMsgClassMap = Collections.unmodifiableMap(tmpMap); }
程序初始化了一个映射,在指定的包找到请求的消息类的class,读取class上的annotation,保存到一个Map中,这样在后续就可以根据这个Map来根据msgType得到class对象了。
再给出解码器的实现:
package com.company.game.dispatcher.codec; import java.util.List; import com.company.game.dispatcher.util.ClassUtil; import com.company.game.dispatcher.util.GsonUtil; import com.google.gson.Gson; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.ByteToMessageDecoder; /** * 解码器 * 客户端和服务端均有使用 * 0-1字节表示整个消息的长度(单位:字节) * 2-3字节代表消息类型,对应annotation * 余下的是消息的json字符串(UTF-8编码) * * @author xingchencheng * */ public class MsgDecoder extends ByteToMessageDecoder { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> list) throws Exception { if (buf.readableBytes() < 2) { return; } Gson gson = GsonUtil.getGson(); short jsonBytesLength = (short) (buf.readShort() - 2); short type = buf.readShort(); byte[] tmp = new byte[jsonBytesLength]; buf.readBytes(tmp); String json = new String(tmp, "UTF-8"); Class<?> clazz = ClassUtil.getMsgClassByType(type); Object msgObj = gson.fromJson(json, clazz); list.add(msgObj); } }
解码完成后,程序进入到服务端的handler中:
@Override protected void channelRead0(ChannelHandlerContext ctx, Object msgObject) throws Exception { // 分发消息给对应的消息处理器 Dispatcher.submit(ctx.channel(), msgObject); }
Dispatcher代码如下:
package com.company.game.dispatcher; import io.netty.channel.Channel; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import com.company.game.dispatcher.exec.BusinessLogicExecutorBase; import com.company.game.dispatcher.msg.RequestMsgBase; import com.company.game.dispatcher.util.ClassUtil; /** * 抽象了分发器 * 多线程执行 * 某个消息对象msgObject指定某个业务逻辑对象executor * submit到线程池中 * @author xingchencheng * */ public class Dispatcher { private static final int MAX_THREAD_NUM = 50; private static ExecutorService executorService = Executors.newFixedThreadPool(MAX_THREAD_NUM); public static void submit(Channel channel, Object msgObject) throws InstantiationException, IllegalAccessException { RequestMsgBase msg = (RequestMsgBase) msgObject; Class<?> executorClass = ClassUtil.getExecutorClassByType(msg.getType()); BusinessLogicExecutorBase executor = (BusinessLogicExecutorBase) executorClass.newInstance(); executor.setChannel(channel); executor.setMsgObject(msgObject); executorService.submit(executor); } }
我们看到,在代码中也是根据msgType取得了对应的一个class对象,并new了一个对象出来,交给了线程池进行并发执行,这个对象就是业务逻辑处理器对象,它实现了Runnable接口,进行一些业务逻辑上的处理。根据msgType取得class对象的映射过程跟前面提到的映射原理是相同的,可以参见代码。贴出业务逻辑处理器对象的代码:
package com.company.game.dispatcher.exec; import com.company.game.dispatcher.annotation.UserMsgAndExecAnnotation; import com.company.game.dispatcher.msg.MsgType; import com.company.game.dispatcher.msg.UserAddRequest; import com.company.game.dispatcher.msg.UserAddResponse; /** * 具体的业务逻辑 * 实现加法 * * @author xingchencheng * */ @UserMsgAndExecAnnotation(msgType = MsgType.ADD) public class UserAddExecutor extends BusinessLogicExecutorBase { public void run() { UserAddResponse response = new UserAddResponse(); if (this.msgObject instanceof UserAddRequest) { UserAddRequest request = (UserAddRequest) this.msgObject; double result = request.getLeftNumber() + request.getRightNumber(); response.setResult(result); response.setSuccess(true); } else { response.setSuccess(false); } System.out.println("服务端处理结果:" + response.getResult()); channel.writeAndFlush(response); } }
注意,它也得用annotation来修饰。
思路大致就是如此,如果要增加一个请求,在示例代码中,需要做3件事情:
- 在MsgType添加一个类型
- 添加请求相应消息类
- 添加业务逻辑处理器类
而不需要修改消息分发的代码。
示例项目的说明和使用
- 工程可在文章开头的github中或附件得到
- 项目使用Maven3构建,构建的结果是一个jar,可通过命令行分别运行服务端和客户端
- 仅仅是个示例,并没有过多的考虑异常处理,性能等方面
- 没有单元测试和其他测试
提供了命令行工具,帮助信息如下:
服务端启动命令:
客户端启动命令:
结语
本文的描述未必清晰,更好的方法是直接看代码。
关于消息分发想必还有更好的方法,这里只是抛砖引玉,希望路过的各位能提供更好的方法一起参考。
相关推荐
使用Netty4实现多线程的消息分发,这是一个基于netty4做的一个异步通信模型。
mqttserver,基于netty 4.1.1,可解码http、mqtt协议请求。 项目包括: 1.基于netty绑定端口监听,对于mqtt消息和http请求消息分别绑定不同的...4.HttpServerHandler类实现对http消息的自定义处理。该handle类包含以下
项目包括: 1.基于netty绑定端口监听,对于mqtt消息和http请求消息分别绑定不同的...4.HttpServerHandler类实现对http消息的自定义处理。该handle类包含以下内容: 对一次完整的http请求进行解码然后处理该请求。
遗嘱消息, 保留消息及消息分发重试 心跳机制 MQTT连接认证(可选择是否开启) SSL方式连接(可选择是否开启) 主题过滤(支持单主题订阅如 test_topic /mqtt/test --不能以/结尾, 通配符订阅 # /mqtt/# --以#结尾) ...
使用Netty+WebSocket实现聊天服务器,实现用户与店家的交流,并发送数据到mq,保存聊天消息和图片 使用Zookeeper作为分布式锁,保证库存扣减的一致性(本项目作用不大) 配置一主一从的MySQL进行读写分离、主从同步,...
详解RabbitMQ消息分发机制及主题消息分发 RabbitMQ消息路由机制分析 RabbitMQ消息确认机制 Redis redis数据结构分析 Redis主从复制原理及无磁盘复制分析 Redis管道模式详解 Redis缓存与数据库一致性问题解决...
趁早(quickearly)B2C早餐外卖微信小程序,理想经营范围为人口密集的小区。用户可以提前预定,选择预定时间,也可以立即...使用Netty+WebSocket实现聊天服务器,实现用户与店家的交流,并发送数据到mq,保存聊天消息和图
还实现了对应的自动扫描接口,并生产对应的代理类注入到对应的ioc容器中,并加上对应的自动DI操作,服务端对应的消息的分发模式,代码更加简单明白,网络部分,加上了对应的tcp粘包,json编解码
weixin4j 微信开发工具包 weixin4j是一个用Java编写针对微信...netty服务器&消息分发 weixin4j-example 示例工程,包含了如何构建weixin4j-server服务器和与spring集成,建议看看 接下来 微信门店 微信摇一摇 微信连WI-
使用Netty+WebSocket实现聊天服务器,实现用户与店家的交流,并发送数据到mq,保存聊天消息和图片 使用Zookeeper作为分布式锁,保证库存扣减的一致性(本项目作用不大) 配置一主一从的MySQL进行读写分离、主从同步,...
Message Pipe是基于Redis实现的顺序消息管道,由于内部引入了Redisson分布式锁所以它是线程安全的,多线程情况下也会按照写入管道的顺序执行消费。...消息分发 顺序消费 读写分离 线程安全 负载均衡 自动剔除
Message Pipe采用Client、Server概念进行设计,内部通过grpc-netty来建立消息通道相互通信的长连接,消息的分发由Server负责,而每一个管道内的消息在分发时会通过LoadBalance(负载均衡)的方式来获取
Message Pipe采用Client、Server概念进行设计,内部通过grpc-netty来建立消息通道相互通信的长连接,消息的分发由Server负责,而每一个管道内的消息在分发时会通过LoadBalance(负载均衡)的方式来获取
《Linux多线程服务端编程:使用muduo C++网络库》主要讲述采用现代C++在x86-64 Linux上编写多线程TCP网络服务程序的主流常规技术,重点讲解一种适应性较强的多线程服务器的编程模型,即one loop per thread。...