netty4+protobuf3最佳實踐

本文要點:java

  1. netty4+protobuf3多類型傳輸實現
  2. 優雅的實現消息分發

作後臺服務常常有這樣的流程: git

2eeae9b5-4a34-4171-b0e8-4f695c29d2d9.png
如何優雅的完成這個過程呢?下面分享下基於 netty+protobuf的方案:

首先要解決的是如何在netty+protobuf中傳輸多個protobuf協議,這裏採起的方案是使用一個類來作爲描述協議的方案,也就是須要二次解碼的方案,IDL文件以下:github

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可讓類分開生成,更清晰些,也更方便後面利用反射來獲取這些類.json

生成好了protobuf,咱們解包的過程就應該是這樣的: 數組

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類型相同的方法來緩存這些處理類(有一點不一樣的是這裏只須要緩存一個處理類的實例而不是方法,由於parseFromstatic的沒法統一調用),作完這些咱們就能夠編寫咱們的pipeline上的最後一個處理器:bash

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類都可以使用模版來生成,完整代碼請看這裏app

ps:其實本例中因爲使用了protobuf還有比較強約定性,因此理論上來講每一個消息處理器上的自定義註解是不須要的,經過獲取泛型的真實類型便可,可是註解能夠大大增長handler的靈活性,若是採用其餘方案(例如json)也更有借鑑的價值。ide

相關文章
相關標籤/搜索