使用Netty4實現基本的消息分發

本文轉載自查看原文 2017-11-13 16:54 1523 netty

示例工程代碼

可從附件下載

具體的說明和用法在后面介紹

需求與目的

一個游戲服務端需要處理各種業務邏輯，每一種業務邏輯都對應着一個請求消息和一個響應消息。那么服務端需要把這些不同的消息自動分發到對應的業務邏輯中處理。

最簡單的處理方式就是根據請求消息中的type字段，使用switch case來進行分別處理，但這種方式隨着消息的增多，顯現了一些壞味道：長長的一大坨不太好看；如果要添加新的消息、新的邏輯，或者去掉新的消息、新的邏輯，在代碼上不但要修改這些消息和邏輯，還不得不修改這長長的一坨swtich case，這樣的修改顯得很多余。

所以我們的目的就是把消息分發這塊的代碼自動化，在增加、修改、刪除消息和邏輯的時候不需要再對消息分發的代碼再做修改，從而使得修改的代碼最小化。

實現原理

在實現中，使用了注解（annotation）

Java代碼

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開始計數。

當我們要增加一個加法消息時，就使用這個注解來修飾我們的請求消息類：

Java代碼

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的原因。

在服務端程序啟動前，會執行下面的處理：

Java代碼

// 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對象了。

再給出解碼器的實現：

Java代碼

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中：

Java代碼

@Override
protected void channelRead0(ChannelHandlerContext ctx, Object msgObject)
throws Exception {
// 分發消息給對應的消息處理器
Dispatcher.submit(ctx.channel(), msgObject);
}

Dispatcher代碼如下：

Java代碼

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對象的映射過程跟前面提到的映射原理是相同的，可以參見代碼。貼出業務邏輯處理器對象的代碼：

Java代碼