netty4+protobuf3最佳实践

我就是我 2023-10-11 12:17 5阅读 0赞

> 本文要点：

1.  netty4+protobuf3多类型传输实现
2.  优雅的实现消息分发

做后台服务经常有这样的流程:

![format_png][]

2eeae9b5-4a34-4171-b0e8-4f695c29d2d9.png

如何优雅的完成这个过程呢？下面分享下基于`netty+protobuf`的方案:

首先要解决的是如何在`netty+protobuf`中传输多个protobuf协议，这里采取的方案是使用一个类来做为描述协议的方案，也就是需要二次解码的方案，IDL文件如下:

syntax = "proto3";
    option java_package = "com.nonpool.proto";
    option java_multiple_files = true;
    
    message Frame {
       string messageName = 1;
       
       bytes payload = 15;
    }
    
    message TextMessage {
       string  text = 1;
    }

Frame为描述协议，所有消息在发送的时候都序列化成byte数组写入Frame的payload，**`messageName`约定为要发送的`message的类名`**，生成的时候设置`java_multiple_files = true`可以让类分开生成，更清晰些,也更方便后面利用反射来获取这些类.

生成好了protobuf，我们解包的过程就应该是这样的:

![format_png 1][]

71d1665d-db17-48bf-a7b2-86f4c2faf652.png

其中protobuf的序列化和反序列化netty已经为我们编写好了对应的解码/编码器，直接调用即可,我们只需要编写二次解码/编码器即可:

public class SecondProtobufCodec extends MessageToMessageCodec<Frame, MessageLite> {
    
        @Override
        protected void encode(ChannelHandlerContext ctx, MessageLite msg, List<Object> out) throws Exception {
        
            out.add(Frame.newBuilder()
                    .setMessageType(msg.getClass().getSimpleName())
                    .setPayload(msg.toByteString())
                    .build());
        }
    
        @Override
        protected void decode(ChannelHandlerContext ctx, Frame msg, List<Object> out) throws Exception {
            out.add(ParseFromUtil.parse(msg));
        }
    
    }

public abstract class ParseFromUtil {
    
        private final static ConcurrentMap<String, Method> methodCache = new ConcurrentHashMap<>();
    
        static {
            //找到指定包下所有protobuf实体类
            List<Class> classes = ClassUtil.getAllClassBySubClass(MessageLite.class, true, "com.nonpool.proto");
            classes.stream()
                    .filter(protoClass -> !Objects.equals(protoClass, Frame.class))
                    .forEach(protoClass -> {
                        try {
                            //反射获取parseFrom方法并缓存到map
                            methodCache.put(protoClass.getSimpleName(), protoClass.getMethod("parseFrom", ByteString.class));
                        } catch (NoSuchMethodException e) {
                            throw new RuntimeException(e);
                        }
                    });
        }
        
        /**
         * 根据Frame类解析出其中的body
         *
         * @param msg
         * @return
         */
        public static MessageLite parse(Frame msg) throws InvocationTargetException, IllegalAccessException {
            String type = msg.getMessageType();
            ByteString body = msg.getPayload();
    
            Method method = methodCache.get(type);
            if (method == null) {
                throw new RuntimeException("unknown Message type :" + type);
            }
            return (MessageLite) method.invoke(null, body);
        }
    }

至此，我们收发数据的解码/编码已经做完配合自带的解码/编码器，此时`pipeline`的处理链是这样的:

public void initChannel(SocketChannel ch) throws Exception {
                                ch.pipeline()
                                        .addLast(new ProtobufVarint32FrameDecoder())
                                        .addLast(new ProtobufDecoder(Frame.getDefaultInstance()))
                                        .addLast(new ProtobufVarint32LengthFieldPrepender())
                                        .addLast(new ProtobufEncoder())
                                        .addLast(new SecondProtobufCodec())
                                ;

数据收发完成，接下来就是把消息分发到对应的处理方法。处理方法也利用多态特性+泛型+注解优雅的实现分发。首先定义一个泛型接口:`interface DataHandler<T extends MessageLite>`,该接口上定义一个方法`void handler(T t, ChannelHandlerContext ctx)`,然后每一个类型的处理类都使用自己要处理的类型实现该接口，并使用自定义注解映射处理类型。在项目启动的扫描实现该接口的所有类，使用跟上述解析Message类型相同的方法来缓存这些处理类(有一点不同的是这里只需要缓存一个处理类的实例而不是方法，因为`parseFrom`是`static`的无法统一调用),做完这些我们就可以编写我们的`pipeline`上的最后一个处理器:

public class DispatchHandler extends ChannelInboundHandlerAdapter {
    
        @Override
        @SuppressWarnings("unchecked")
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            HandlerUtil.getHandlerInstance(msg.getClass().getSimpleName()).handler((MessageLite) msg,ctx);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
        }
    }

public abstract class HandlerUtil {
    
        private final static ConcurrentMap<String,DataHandler> instanceCache = new ConcurrentHashMap<>();
    
        static {
            try {
                List<Class> classes = ClassUtil.getAllClassBySubClass(DataHandler.class, true,"com.onescorpion");
                for (Class claz : classes) {
                    HandlerMapping annotation = (HandlerMapping) claz.getAnnotation(HandlerMapping.class);
                    instanceCache.put(annotation.value(), (DataHandler) claz.newInstance());
                }
                System.out.println("handler init success handler Map: " +  instanceCache);
            } catch (Exception e) {
                e.printStackTrace();
            }
    
        }
    
        public static DataHandler getHandlerInstance(String name) {
            return instanceCache.get(name);
        }
    }

这样一个优雅的处理分发就完成了。  
由于代码规律性极强，所以所有handler类均可以使用模版来生成，完整代码请看[这里][Link 1]

ps：其实本例中由于使用了protobuf还有比较强约定性，所以理论上来说每个消息处理器上的自定义注解是不需要的，通过获取泛型的真实类型即可，但是注解可以大大增加handler的灵活性，如果采用其他方案(例如json)也更有借鉴的价值。

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8zMTg5MzAxLTMxNThjMzE5Mzk5NWVkZTAucG5nP2ltYWdlTW9ncjIvYXV0by1vcmllbnQvc3RyaXB8aW1hZ2VWaWV3Mi8yL3cvNTY2L2Zvcm1hdC93ZWJw?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8zMTg5MzAxLWZjNDk2MGNjNTExMzYxZjUucG5nP2ltYWdlTW9ncjIvYXV0by1vcmllbnQvc3RyaXB8aW1hZ2VWaWV3Mi8yL3cvNjEzL2Zvcm1hdC93ZWJw?x-oss-process=image/format,png
[Link 1]: https://link.jianshu.com/?t=https%3A%2F%2Fgithub.com%2Fnonpool%2Ftypedispatchdemo