Swift的网络框架采用了jboss的netty库，没有采用后期的netty标准库，它实现了纯异步的Thrift服务器和客户端框架，本篇主要介绍服务端的源码。

1、以一个简单的例子开始

我们以一个简单的服务器端例子开始：

public static void main(String[] args) {
        //SwiftScribe scribeService = new SwiftScribe();

        ExecutorService taskWorkerExecutor;
        ThriftServer server;
        CountDownLatch latch;
        ExecutorService bossExecutor;
        ExecutorService ioWorkerExecutor;


        ThriftServiceProcessor processor = new ThriftServiceProcessor(
                new ThriftCodecManager(),
                ImmutableList.<ThriftEventHandler>of(),
                new SwiftScribe()
        );

        taskWorkerExecutor = newFixedThreadPool(1);

        ThriftServerDef serverDef = ThriftServerDef.newBuilder()
                                                   .listen(8899)
                                                   .withProcessor(processor)
                                                   .using(taskWorkerExecutor)
                                                   .build();

        bossExecutor = newCachedThreadPool();
        ioWorkerExecutor = newCachedThreadPool();

        NettyServerConfig serverConfig = NettyServerConfig.newBuilder()
                                                          .setBossThreadExecutor(bossExecutor)
                                                          .setWorkerThreadExecutor(ioWorkerExecutor)
                                                          .build();


        server = new ThriftServer(serverConfig, serverDef) ;
        server.start();
    }

我们首先封装了ThriftServerDef和NettyServerConfig，由这两个参数来构造一个ThriftServer对象，然后启动Thrift服务。对外就可以提供端口号为8899的高并发的Thrift协议服务能力。

该服务对外提供的服务协议在SwiftScribe中，SwfitScribe的框架类是通过scribe.thrift文件生成，

namespace java com.facebook.swift.service.scribe

enum ResultCode
{
  OK,
  TRY_LATER
}

struct LogEntry
{
  1:  string category,
  2:  string message
}

service scribe
{
  ResultCode Log(1: list<LogEntry> messages);
}

生成框架的命令

java -jar .\swift-generator-cli-0.19.2-standalone.jar scribe.thrift -out ..\java

2、类说明

• ThriftCodecManager类

ThriftCodecManager包含了所有已知的ThriftCodec的索引，并且可以按需为未知类型创建编解码器。既然编解码器的创建耗费资源， 所以每个类只有一个实例会被创建，并被ThriftCodecManager类管理起来。

目前包括的类型有：

public enum ThriftProtocolType
{
    UNKNOWN((byte) 0),
    BOOL((byte) 2),
    BYTE((byte) 3),
    DOUBLE((byte) 4),
    I16((byte) 6),
    I32((byte) 8),
    I64((byte) 10),
    STRING((byte) 11),
    STRUCT((byte) 12),
    MAP((byte) 13),
    SET((byte) 14),
    LIST((byte) 15),
    ENUM((byte) 8), // same as I32 type
    BINARY((byte) 11); // same as STRING type

    private final byte type;
}

在ThriftCodecManager中，通过typeCodecs管理所有数据类型的协议编解码功能。

ThriftCodecManager通过Guawa库中的LoadingCache来缓存每种Thrift类型的编解码器，LoadingCache的原理这里大概说一下，如果能从缓存中找到到数据，那么就直接返回，否则会调用构造器中传入的Callable对象的call方法，来根据key值计算出value，并缓存起来。然后才返回数据。

        addBuiltinCodec(new BooleanThriftCodec());
        addBuiltinCodec(new ByteThriftCodec());
        addBuiltinCodec(new ShortThriftCodec());
        addBuiltinCodec(new IntegerThriftCodec());
        addBuiltinCodec(new LongThriftCodec());
        addBuiltinCodec(new DoubleThriftCodec());
        addBuiltinCodec(new ByteBufferThriftCodec());
        addBuiltinCodec(new StringThriftCodec());
        addBuiltinCodec(new VoidThriftCodec());
        addBuiltinCodec(new BooleanArrayThriftCodec());
        addBuiltinCodec(new ShortArrayThriftCodec());
        addBuiltinCodec(new IntArrayThriftCodec());
        addBuiltinCodec(new LongArrayThriftCodec());
        addBuiltinCodec(new DoubleArrayThriftCodec());

以上是内在的类型以及相应的编解码器，所以内在的类型是不需要计算，对于自定义的复杂的类型，就需要计算了，如：enum, set, map, struct等，由于初始化的时候，这些类型都是没有放入编解码器缓存的，所以第一次查询肯定查询不到。
这时候就需要计算：

 typeCodecs = CacheBuilder.newBuilder().build(new CacheLoader<ThriftType, ThriftCodec<?>>()
        {
            public ThriftCodec<?> load(ThriftType type)
                    throws Exception
            {
                try {
                    // When we need to load a codec for a type the first time, we push it on the
                    // thread-local stack before starting the load, and pop it off afterwards,
                    // so that we can detect recursive loads.
                    stack.get().push(type);

                    switch (type.getProtocolType()) {
                        case STRUCT: {
                            return factory.generateThriftTypeCodec(ThriftCodecManager.this, type.getStructMetadata());
                        }
                        case MAP: {
                            return new MapThriftCodec<>(type, getElementCodec(type.getKeyTypeReference()), getElementCodec(type.getValueTypeReference()));
                        }
                        case SET: {
                            return new SetThriftCodec<>(type, getElementCodec(type.getValueTypeReference()));
                        }
                        case LIST: {
                            return new ListThriftCodec<>(type, getElementCodec(type.getValueTypeReference()));
                        }
                        case ENUM: {
                            return new EnumThriftCodec<>(type);
                        }
                        default:
                            if (type.isCoerced()) {
                                ThriftCodec<?> codec = getCodec(type.getUncoercedType());
                                TypeCoercion coercion = catalog.getDefaultCoercion(type.getJavaType());
                                return new CoercionThriftCodec<>(codec, coercion);
                            }
                            throw new IllegalArgumentException("Unsupported Thrift type " + type);
                    }
                }
                finally {
                    ThriftType top = stack.get().pop();
                    checkState(type.equals(top),
                               "ThriftCatalog circularity detection stack is corrupt: expected %s, but got %s",
                               type,
                               top);
                }

            }
        });

从ThriftCodecManager中获取编解码器主要有两种方法：getCodec和getCodecIfpresent，前者不存在会调用前面的builder方法构建，后者不存在返回null。

ThriftCodecManager包含一个 ThriftCatalog対像，ThriftCatalog包含了所有已知的structs, enums,和类型隐含转换器。因为元数据抽取是非常消耗资源的，所有catalog是单实例。

ThriftCatalog的构造函数如下，它采用DefaultJavaCorecions类来加载隐式类型转换器。

    @VisibleForTesting
    public ThriftCatalog(Monitor monitor)
    {
        this.monitor = monitor;
        addDefaultCoercions(DefaultJavaCoercions.class);
    }

可以用来转换一些原始数据类型：int,long,double,float,byte,byte[]
在DefaultJavaCoercions类中，包含了一些方法，用来转换原始类型和对应的封装类型，这些方法都有注解，注解有两种类型：FromThrift和ToThrift。
在addDefaultCoercions中，根据不同的注解，会把相应的方法，放到
Map<ThriftType, Method> toThriftCoercions = new HashMap<>();
Map<ThriftType, Method> fromThriftCoercions = new HashMap<>();
两个结构体中。最后把每种类型的的FromThrift方法和ToThrift方法封装成一个类型转换器，存放到隐式转换映射中。

        Map<Type, TypeCoercion> coercions = new HashMap<>();
        for (Map.Entry<ThriftType, Method> entry : toThriftCoercions.entrySet()) {
            ThriftType type = entry.getKey();
            Method toThriftMethod = entry.getValue();
            Method fromThriftMethod = fromThriftCoercions.get(type);
            TypeCoercion coercion = new TypeCoercion(type, toThriftMethod, fromThriftMethod);
            coercions.put(type.getJavaType(), coercion);
        }
        this.coercions.putAll(coercions);

• ThriftServiceProcessor类

用来封装一个Thrift服务。

    private final Map<String, ThriftMethodProcessor> methods;
    private final List<ThriftEventHandler> eventHandlers;

methods存储每个方法的处理器对象，而eventHandlers则是用于给外部监控用，类的定义如下所示：

public abstract class ThriftEventHandler
{
    public Object getContext(String methodName, RequestContext requestContext)
    {
        return null;
    }

    public void preRead(Object context, String methodName) {}
    public void postRead(Object context, String methodName, Object[] args) {}
    public void preWrite(Object context, String methodName, Object result) {}
    public void preWriteException(Object context, String methodName, Throwable t) {}
    public void postWrite(Object context, String methodName, Object result) {}
    public void postWriteException(Object context, String methodName, Throwable t) {}
    public void declaredUserException(Object o, String methodName, Throwable t, ThriftCodec<?> exceptionCodec) {}
    public void undeclaredUserException(Object o, String methodName, Throwable t) {}
    public void done(Object context, String methodName) {}
}

这里说明一下process函数
1）读取Thrift的消息开始信息
2）获取Thrift方法名和SequenceId（这两个信息都在开始信息里面，还包括信息类型是CALL还是ONEWAY）
3）根据Thrift方法名，获取对应的处理器:ThriftMethodProcessor
4）生成ContextChain对象
5）调用方法处理器进行处理。
6）添加回调。这里的回调是基于google的Futures来做的。
但是这个回调感觉对于那些需要直接返回的函数，会增加额外的开销，因为回调的处理，需要放到另外一个执行器中处理（线程池），然后才会返回。

• ThriftServiceMetadata类

成员说明：

    private final String name;  // 服务名，通过@ThriftService("名字")注解读取出来的名字。
    private final Map<String, ThriftMethodMetadata> methods;  // 类所有的Thrift方法，包括父类和所有接口的

Thrift方法

    private final Map<String, ThriftMethodMetadata> declaredMethods; // 类本身自己定义的Thrift方法
    private final ImmutableList<ThriftServiceMetadata> parentServices;  // 父类或者接口定义的元数据
    private final ImmutableList<String> documentation;   // 通过ThriftDocumentation注解的文档

• ThriftMethodMetadata类

成员说明：

    private final String name; // 方法名
    private final String qualifiedName; // 服务名.方法名
    private final ThriftType returnType;
    private final List<ThriftFieldMetadata> parameters;
    private final Method method;
    private final ImmutableMap<Short, ThriftType> exceptions;
    private final ImmutableList<String> documentation;
    private final boolean oneway;

类的构造过程：(以例子中的Scribe.log为例进行说明）
1）获取参数编号，一般为index+1
2）获取参数名称
3）根据返回值和参数的Java类型获取（或者按需生成）对应的ThriftType【ThriftType thriftType = catalog.getThriftType(parameterType);】
获取的逻辑：如果缓存中有，则从缓存中获取，否则调用 buildThriftType-->buildThriftTypeInternal来生成。
buildThriftTypeInternal的逻辑大致为：
i）获取类型的裸类型Class<?> rawType = TypeToken.of(javaType).getRawType();
关于裸类型与元素类型：
举个栗子：java.util.List<String>的裸类型是java.util.List，元素类型是：String
ii）如果裸类型是原始类型返回对应的ThriftType，如：DOUBLE，STRING，I32， I64等。
iii）如果是ByteBuffer返回BINARY
iv）如果是Enum生成相应的ThriftEnumMetadata，并据此创建对应的Enum的Thrift类型，具体逻辑略
v）如果是数组，则生成具体类型的数组thrift类型，具体逻辑略
vi）如果是Map，获取key和value的具体类型，生成相应的thrifttype，在根据这两thrifttype生成map Thrifttype。
vii）如果是void，返回VOID
viii）如果是结构
ix）如果是Future类型（牛逼，THrift居然有这种东东）
x）其他情况，这时候，将采用隐式转换类型，来获取相应的ThriftType。
关于隐式转换前面已经提到了，系统在初始化时，把DefaultJavaCoercions类作为隐式转换的初始化对象。主要是一些java原始类型与其包装类之间的相互转换。另外在ThriftType中有对应的
这里细节先就先不提了，以后另外专门分一个章节来讲解。
4）构建异常信息
5）单方向信息

• ThriftMethodProcessor类

该类负责执行一个客户端的Thrift请求，它封装了该方法的元数据和编解码器相关信息。

这里先看看他的process方法的主要流程：
1）读取参数信息
2）读取消息结束标志
【注意：读取消息开始，以及读取消息名，都在前面ThriftServiceProcessor类的process函数中处理过了】
3） 调用方法 : Object response = method.invoke(service, args); 这个很简单。
在这里，它对于立即返回和Future返回值的函数做了统一处理。

      try {
            Object response = method.invoke(service, args);
            if (response instanceof ListenableFuture) {
                return (ListenableFuture<?>) response;
            }
            return Futures.immediateFuture(response);
        }
        catch (IllegalAccessException | IllegalArgumentException e) {
            // These really should never happen, since the method metadata should have prevented it
            return Futures.immediateFailedFuture(e);
        }
        catch (InvocationTargetException e) {
            Throwable cause = e.getCause();
            if (cause != null) {
                return Futures.immediateFailedFuture(cause);
            }

            return Futures.immediateFailedFuture(e);
        }

4）返回数据
注意，这里有两块：成功的和失败的。
成功的情况下，会返回一个

• ThriftServer类

负责服务器的相关元素的封装和服务器的启动，它主要依赖于NettyServerTransport类来完成服务器的相关功能。

1、首先看看编解码工厂类

有两块：
1）、DEFAULT_FRAME_CODEC_FACTORIES 存放了两个基于Framed的编解码：buffered和framed
2）、DEFAULT_PROTOCOL_FACTORIES 存放了两种协议解析binary和compact，
binary：简单的二进制编码
compact：以一种更加紧凑的编码方式编码，

注意：Thrift也可以使用Json来编码，不过Swfit不支持

• NettyServerTransport类

这是一个核心的通道类，用于解析framed 的thrift消息，分到给指定的TProcessor，然后把返回的消息打包返回都Thrift栈。

1、启动服务

    public void start(ServerChannelFactory serverChannelFactory)
    {
        bootstrap = new ServerBootstrap(serverChannelFactory);
        bootstrap.setOptions(nettyServerConfig.getBootstrapOptions());
        bootstrap.setPipelineFactory(pipelineFactory);
        serverChannel = bootstrap.bind(new InetSocketAddress(requestedPort));
        InetSocketAddress actualSocket = (InetSocketAddress) serverChannel.getLocalAddress();
        actualPort = actualSocket.getPort();
        Preconditions.checkState(actualPort != 0 && (actualPort == requestedPort || requestedPort == 0));
        log.info("started transport %s:%s", def.getName(), actualPort);
    }

2、协议栈

       this.pipelineFactory = new ChannelPipelineFactory()
        {
            @Override
            public ChannelPipeline getPipeline()
                    throws Exception
            {
                ChannelPipeline cp = Channels.pipeline();
                TProtocolFactory inputProtocolFactory = def.getDuplexProtocolFactory().getInputProtocolFactory();
                NiftySecurityHandlers securityHandlers = def.getSecurityFactory().getSecurityHandlers(def, nettyServerConfig);
                cp.addLast("connectionContext", new ConnectionContextHandler());
                cp.addLast("connectionLimiter", connectionLimiter);
                cp.addLast(ChannelStatistics.NAME, channelStatistics);
                cp.addLast("encryptionHandler", securityHandlers.getEncryptionHandler());
                cp.addLast("frameCodec", def.getThriftFrameCodecFactory().create(def.getMaxFrameSize(),
                                                                                 inputProtocolFactory));
                if (def.getClientIdleTimeout() != null) {
                    // Add handlers to detect idle client connections and disconnect them
                    cp.addLast("idleTimeoutHandler", new IdleStateHandler(nettyServerConfig.getTimer(),
                                                                          def.getClientIdleTimeout().toMillis(),
                                                                          NO_WRITER_IDLE_TIMEOUT,
                                                                          NO_ALL_IDLE_TIMEOUT,
                                                                          TimeUnit.MILLISECONDS));
                    cp.addLast("idleDisconnectHandler", new IdleDisconnectHandler());
                }

                cp.addLast("authHandler", securityHandlers.getAuthenticationHandler());
                cp.addLast("dispatcher", new NiftyDispatcher(def, nettyServerConfig.getTimer()));
                cp.addLast("exceptionLogger", new NiftyExceptionLogger());
                return cp;
            }
        };

1）ConnectionContextHandler 负责在连接建立成功的时候，创建NiftyConnectionContext上线文对象，附着在ctx中【ConnectionHandlerContext】
2）ConnectionLimiter 负责对会话打开和关闭是对连接数进行计数，并判断是否超过最大连接数，超出的会直接关闭连接。
3）ChannelStatistics 负责统计打开的channel数、上下行流量。
4）安全方面的Handler，这里先不分析了。
5）DefaultThriftFrameCodec，负责Thrift的协议头的解析。把字节数据变成ThriftMessage对象。从而会触发到NiftyDispatcher处理器。
对于Framed的消息，Thrift是4个字节的长度+对应长度的消息体。
6）IdleStateHandler用来检查读写或者两者一起Idle的情况，如果发生了超时现象，就会向上汇报IDLE事件。
7）IdleDisconnectHandler用于处理Idle事件，收到Idle事件后，IdleDisconnectHandler负责关闭连接。
8）认证处理器
9）NiftyDispatcher负责消息的分发
10）NiftyExceptionLogger：负责记录网络中的异常数据到日志中。

**NiftyDispatcher类
负责Thrift消息的处理、分发和返回处理。

1、消息处理部分

前面在协议栈已经分析过了，在DefaultThriftFrameCodec处理器处理过后，就会把字节数据变成ThriftMessage，从而触发NiftyDispatcher的处理，如下所示：

    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
            throws Exception
    {
        if (e.getMessage() instanceof ThriftMessage) {
            ThriftMessage message = (ThriftMessage) e.getMessage();
            message.setProcessStartTimeMillis(System.currentTimeMillis());
            checkResponseOrderingRequirements(ctx, message);

            TNiftyTransport messageTransport = new TNiftyTransport(ctx.getChannel(), message);
            TTransportPair transportPair = TTransportPair.fromSingleTransport(messageTransport);
            TProtocolPair protocolPair = duplexProtocolFactory.getProtocolPair(transportPair);        // 活的双线协议处理器，这里会根据TThriftServerDef中定义的，缺省值为：this.duplexProtocolFactory = TDuplexProtocolFactory.fromSingleFactory(new TBinaryProtocol.Factory(true, true));

            TProtocol inProtocol = protocolPair.getInputProtocol();
            TProtocol outProtocol = protocolPair.getOutputProtocol();

            processRequest(ctx, message, messageTransport, inProtocol, outProtocol);
        }
        else {
            ctx.sendUpstream(e);
        }
    }

从上面代码中，协议的出入栈打包器在这里创建了出来，TNiftyTransport类似乎继承和实现了原生Thrift的TTransport类。（注意看ThriftTransport类的in和out成员，in成员是传入的TThriftMessage的消息体，而out则是新城成的一个ChannelBuffer）

2、消息分发

NiftyDispatcher的exec成员是一个Executor对象，它负责分发消息执行，收到的ThriftMessage会被放到工作线程池中执行，而不是在io线程中。

           exe.execute(new Runnable() {
                @Override
                public void run() {
                    ListenableFuture<Boolean> processFuture;
                    final AtomicBoolean responseSent = new AtomicBoolean(false);
                    // Use AtomicReference as a generic holder class to be able to mark it final
                    // and pass into inner classes. Since we only use .get() and .set(), we don't
                    // actually do any atomic operations.
                    final AtomicReference<Timeout> expireTimeout = new AtomicReference<>(null);

                    try {
                        try {
                            // 超时处理
                            long timeRemaining = 0;
                            long timeElapsed = System.currentTimeMillis() - message.getProcessStartTimeMillis();
                            if (queueTimeoutMillis > 0) {
                                if (timeElapsed >= queueTimeoutMillis) {
                                    TApplicationException taskTimeoutException = new TApplicationException(
                                            TApplicationException.INTERNAL_ERROR,
                                            "Task stayed on the queue for " + timeElapsed +
                                                    " milliseconds, exceeding configured queue timeout of " + queueTimeoutMillis +
                                                    " milliseconds."
                                    );
                                    sendTApplicationException(taskTimeoutException, ctx, message, requestSequenceId, messageTransport,
                                        inProtocol, outProtocol);
                                    return;
                                }
                            } else if (taskTimeoutMillis > 0) {
                                if (timeElapsed >= taskTimeoutMillis) {
                                    TApplicationException taskTimeoutException = new TApplicationException(
                                            TApplicationException.INTERNAL_ERROR,
                                            "Task stayed on the queue for " + timeElapsed +
                                                    " milliseconds, exceeding configured task timeout of " + taskTimeoutMillis +
                                                    " milliseconds."
                                    );
                                    sendTApplicationException(taskTimeoutException, ctx, message, requestSequenceId, messageTransport,
                                            inProtocol, outProtocol);
                                    return;
                                } else {
                                    timeRemaining = taskTimeoutMillis - timeElapsed;
                                }
                            }

                            if (timeRemaining > 0) {
                                expireTimeout.set(taskTimeoutTimer.newTimeout(new TimerTask() {
                                    @Override
                                    public void run(Timeout timeout) throws Exception {
                                        // The immediateFuture returned by processors isn't cancellable, cancel() and
                                        // isCanceled() always return false. Use a flag to detect task expiration.
                                        if (responseSent.compareAndSet(false, true)) {  // 只能被发送一次返回消息
                                            TApplicationException ex = new TApplicationException(
                                                    TApplicationException.INTERNAL_ERROR,
                                                    "Task timed out while executing."
                                            );
                                            // Create a temporary transport to send the exception
                                            ChannelBuffer duplicateBuffer = message.getBuffer().duplicate();
                                            duplicateBuffer.resetReaderIndex();
                                            TNiftyTransport temporaryTransport = new TNiftyTransport(
                                                    ctx.getChannel(),
                                                    duplicateBuffer,
                                                    message.getTransportType());
                                            TProtocolPair protocolPair = duplexProtocolFactory.getProtocolPair(
                                                    TTransportPair.fromSingleTransport(temporaryTransport));
                                            sendTApplicationException(ex, ctx, message,
                                                    requestSequenceId,
                                                    temporaryTransport,
                                                    protocolPair.getInputProtocol(),
                                                    protocolPair.getOutputProtocol());
                                        }
                                    }
                                }, timeRemaining, TimeUnit.MILLISECONDS));
                            }

                            ConnectionContext connectionContext = ConnectionContexts.getContext(ctx.getChannel());
                            RequestContext requestContext = new NiftyRequestContext(connectionContext, inProtocol, outProtocol, messageTransport);
                            RequestContexts.setCurrentContext(requestContext);
                            processFuture = processorFactory.getProcessor(messageTransport).process(inProtocol, outProtocol, requestContext);
                        } finally {
                            // RequestContext does NOT stay set while we are waiting for the process
                            // future to complete. This is by design because we'll might move on to the
                            // next request using this thread before this one is completed. If you need
                            // the context throughout an asynchronous handler, you need to read and store
                            // it before returning a future.
                            RequestContexts.clearCurrentContext();
                        }

                        Futures.addCallback(
                                processFuture,
                                new FutureCallback<Boolean>() {
                                    @Override
                                    public void onSuccess(Boolean result) {
                                        deleteExpirationTimer(expireTimeout.get());
                                        try {
                                            // Only write response if the client is still there and the task timeout
                                            // hasn't expired.
                                            if (ctx.getChannel().isConnected() && responseSent.compareAndSet(false, true)) {  // 只能被发送一次返回消息，防止跟超时消息搞重
                                                ThriftMessage response = message.getMessageFactory().create(
                                                        messageTransport.getOutputBuffer());
                                                writeResponse(ctx, response, requestSequenceId,
                                                        DispatcherContext.isResponseOrderingRequired(ctx));
                                            }
                                        } catch (Throwable t) {
                                            onDispatchException(ctx, t);
                                        }
                                    }

                                    @Override
                                    public void onFailure(Throwable t) {
                                        deleteExpirationTimer(expireTimeout.get());
                                        onDispatchException(ctx, t);
                                    }
                                }
                        );
                    } catch (TException e) {
                        onDispatchException(ctx, e);
                    }
                }
            });

3、流程说明

1、初始化ThriftServiceProcessor
这个过程充分利用了java的反射原理，基于一个Thrift的服务处理类来构建Thrift服务处理器。
逐步步骤为：
1）生成ThriftCodecManager对象，用于对Thrift协议中的各种数据类型进行编解码处理。
这个步骤有点特殊，将独立说明，见：ThriftCodecManager的构建。
2）构建一个服务器处理对象，例子中是一个SwfitScribe対象
3）分析服务器处理対像，提出处理函数，绑定到服务处理器中。

2、协议解析以及消息分发机制