阿里开源分布式事务组件 seata ：seata server 通信层解析

RPC ？

seata client 和 seata server 间是需要通过网络通信来传递信息的，client 发送请求消息给 server，server 根据实际的处理逻辑，可能会给 client 发送相应的响应消息，或者不响应任何消息。在 seata 中，客户端和服务端的通信实现，被抽象成来公共的模块，它的 package 位于 io.seata.core.rpc 中。

这个包名叫 rpc，这个包下的很多类名也有 rpc 相关的字眼，而实际上在我看来，这个通信框架并不是一个常规意义的 rpc 框架，如果硬要揪书本知识，那么 rpc 的解释如下：

远程过程调用（英语：Remote Procedure Call，缩写为 RPC）是一个计算机通信协议。该协议允许运行于一台计算机的程序调用另一台计算机的子程序，而程序员无需额外地为这个交互作用编程。如果涉及的软件采用面向对象编程，那么远程过程调用亦可称作远程调用或远程方法调用。

在以 dubbo 为代表的微服务时代下，dubbo 常规意义上我们都称之为 rpc 框架，rpc 的理论原则是：程序员无需额外地为这个交互作用编程。那么对于像 dubbo 这样的 rpc 实现，它能让 client 像调用本地代码 api 一样，来调用远程 server 上的某个 method。

在 client 这一层直接面向 interface 编程，通过动态代理的方式，对上层屏蔽掉通信细节，在底层，将方法调用，通过序列化方式，封装成一个二进制数据串发送给 server，server 层解析该消息，通过反射的方式，将 interface 对应的 implemention 执行起来，将执行结果，扁平化成一个二进制数据串，回送给 client，client 收到数据后，拼装成 interface api 所定义的返回值类型的一个实例，作为方法调用的返回值。整个底层的细节，应用层面并不需要了解，应用层只需要以 interface.method 的方式，就像代码在本地执行一样，就能把远端 interface_implemention.method 给调用起来。

而 seata 的 rpc 框架上，实际上仅仅是一个普通的基于 netty 的网络通信框架，client 与 server 之间通过发送 request 和 response 来达到相互通信的目的，在 seata 中的每个 request 和 response 类，都实现了如何把自己序列化的逻辑。

各种消息类型，都实现了 io.seata.core.protocol.MessageCodec 接口

public interface MessageCodec {
    /**
     * Gets type code.
     *
     * @return the type code
     */
    short getTypeCode();

    /**
     * Encode byte [ ].
     *
     * @return the byte [ ]
     */
    byte[] encode();

    /**
     * Decode boolean.
     *
     * @param in the in
     * @return the boolean
     */
    boolean decode(ByteBuf in);
}

以 io.seata.core.protocol.GlobalBeginRequest 为例，它都 decode 和 encode 实现如下所示：

@Override
public byte[] encode() {
    ByteBuffer byteBuffer = ByteBuffer.allocate(256);
    byteBuffer.putInt(timeout);

    if (this.transactionName != null) {
        byte[] bs = transactionName.getBytes(UTF8);
        byteBuffer.putShort((short)bs.length);
        if (bs.length > 0) {
            byteBuffer.put(bs);
        }
    } else {
        byteBuffer.putShort((short)0);
    }

    byteBuffer.flip();
    byte[] content = new byte[byteBuffer.limit()];
    byteBuffer.get(content);
    return content;
}

@Override
public void decode(ByteBuffer byteBuffer) {
    this.timeout = byteBuffer.getInt();

    short len = byteBuffer.getShort();
    if (len > 0) {
        byte[] bs = new byte[len];
        byteBuffer.get(bs);
        this.setTransactionName(new String(bs, UTF8));
    }
}

这意味着，发送方先对 message 做 encode 动作形成字节数组，将字节数组发往接收方，接收方收到字节数组后，对字节数组先判断 message type，再用对应的 message 类型对字节数组做 decode 动作。

类的组织形式

从 seata server 的入口类 io.seata.server.Server 分析，main 方法如下所示：

/**
 * The entry point of application.
 *
 * @param args the input arguments
 * @throws IOException the io exception
 */
public static void main(String[] args) throws IOException {
    RpcServer rpcServer = new RpcServer(WORKING_THREADS);

    int port = SERVER_DEFAULT_PORT;
    //server port
    if (args.length > 0) {
        try {
            port = Integer.parseInt(args[0]);
        } catch (NumberFormatException e) {
            System.err.println("Usage: sh services-server.sh $LISTEN_PORT $PATH_FOR_PERSISTENT_DATA");
            System.exit(0);
        }
    }
    rpcServer.setListenPort(port);

    //log store mode : file、db
    String storeMode = null;
    if (args.length > 1) {
        storeMode = args[1];
    }

    UUIDGenerator.init(1);
    SessionHolder.init(storeMode);

    DefaultCoordinator coordinator = new DefaultCoordinator(rpcServer);
    coordinator.init();
    rpcServer.setHandler(coordinator);
    // register ShutdownHook
    ShutdownHook.getInstance().addDisposable(coordinator);

    if (args.length > 2) {
        XID.setIpAddress(args[2]);
    } else {
        XID.setIpAddress(NetUtil.getLocalIp());
    }
    XID.setPort(rpcServer.getListenPort());

    rpcServer.init();

    System.exit(0);
}

可以看到 seata server 使用一个 RpcServer 类来启动它的服务监听端口，这个端口用来接收 seata client 的消息，RpcServer 这个类是通信层的实现分析的入口。

在这里，SessionHolder 用来做全局事务树的管理，DefaultCoordinator 用来处理事务执行逻辑，而 RpcServer 是这两者可以正常运行的基础，这篇文章的重点在于剖析 RpcServer 的实现，进而延伸到 seata 整个通信框架的细节。

如果先从 RpcServer 的类继承图看的话，那么我们能发现一些与常规思维不太一样的地方，类继承图如下：

褐色部分是 netty 的类，灰色部分是 seata 的类。

在一般常规的思维中，依赖 netty 做一个 server，大致的思路是：

1. 定义一个 xxx server 类
2. 在这个类中设置初始化 netty bootstrap，eventloop，以及设置相应的 ChannelHandler

在这种思维下，很容易想到，server 与 ChannelHandler 之间的关系应该是一个“组合”的关系，即在我们构建 server 的过程中，应该把 ChannelHandler 当作参数传递给 server，成为 server 类的成员变量。

没错，这是我们一般情况下的思维。不过 seata 在这方面却不那么“常规”，从上面的类继承图中可以看到，从 RpcServer 这个类开始向上追溯，发现它其实是 ChannelDuplexHandler 的一个子类或者实例。这种逻辑让人一时很困惑，一个问题在我脑海里浮现：“当我启动一个 RpcServer 的时候，我是真的在启动一个 server 吗？看起来我好像在启动一个 ChannelHandler，可是 ChannelHandler 怎么谈得上‘启动’呢？”

异步转同步的 Future 机制

首先分析 AbstractRpcRemoting 这个类，它直接继承自 ChannelDuplexHandler 类，而 ChannelDuplexHandler 是 netty 中 inbound handler 和 outbound handler 的结合体。

AbstractRpcRemoting 的 init 方法里，仅仅通过 Java 中的定时任务执行线程池启动了一个定时执行的任务：

/**
 * Init.
 */
public void init() {
    timerExecutor.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            List<MessageFuture> timeoutMessageFutures = new ArrayList<MessageFuture>(futures.size());

            for (MessageFuture future : futures.values()) {
                if (future.isTimeout()) {
                    timeoutMessageFutures.add(future);
                }
            }

            for (MessageFuture messageFuture : timeoutMessageFutures) {
                futures.remove(messageFuture.getRequestMessage().getId());
                messageFuture.setResultMessage(null);
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("timeout clear future : " + messageFuture.getRequestMessage().getBody());
                }
            }
            nowMills = System.currentTimeMillis();
        }
    }, TIMEOUT_CHECK_INTERNAL, TIMEOUT_CHECK_INTERNAL, TimeUnit.MILLISECONDS);
}

1. send request message
2. 为业务的请求构建一个业务层面的 future 实例
3. 阻塞等待在这个 future 上
4. 当收到对应的 response message 后，唤醒上面的 future，阻塞等待在这个 future 上的线程继续执行
5. 拿到结果，request 方法结束

AbstractRpcRemoting 定义了几个数据发送相关的方法，分别是：

/**
 * Send async request with response object.
 *
 * @param address the address
 * @param channel the channel
 * @param msg     the msg
 * @return the object
 * @throws TimeoutException the timeout exception
 */
protected Object sendAsyncRequestWithResponse(String address, Channel channel, Object msg) throws TimeoutException;

/**
 * Send async request with response object.
 *
 * @param address the address
 * @param channel the channel
 * @param msg     the msg
 * @param timeout the timeout
 * @return the object
 * @throws TimeoutException the timeout exception
 */
protected Object sendAsyncRequestWithResponse(String address, Channel channel, Object msg, long timeout) throws
    TimeoutException;

/**
 * Send async request without response object.
 *
 * @param address the address
 * @param channel the channel
 * @param msg     the msg
 * @return the object
 * @throws TimeoutException the timeout exception
 */
protected Object sendAsyncRequestWithoutResponse(String address, Channel channel, Object msg) throws
    TimeoutException;

这几个方法就符合上面说到的 发送一个请求，并等待相应 这样的使用场景，上面这三个方法，其实都委托给了 sendAsyncRequest 来实现，这个方法的代码是这样子的：

private Object sendAsyncRequest(String address, Channel channel, Object msg, long timeout)
    throws TimeoutException {
    if (channel == null) {
        LOGGER.warn("sendAsyncRequestWithResponse nothing, caused by null channel.");
        return null;
    }
    final RpcMessage rpcMessage = new RpcMessage();
    rpcMessage.setId(RpcMessage.getNextMessageId());
    rpcMessage.setAsync(false);
    rpcMessage.setHeartbeat(false);
    rpcMessage.setRequest(true);
    rpcMessage.setBody(msg);

    final MessageFuture messageFuture = new MessageFuture();
    messageFuture.setRequestMessage(rpcMessage);
    messageFuture.setTimeout(timeout);
    futures.put(rpcMessage.getId(), messageFuture);

    if (address != null) {
        ConcurrentHashMap<String, BlockingQueue<RpcMessage>> map = basketMap;
        BlockingQueue<RpcMessage> basket = map.get(address);
        if (basket == null) {
            map.putIfAbsent(address, new LinkedBlockingQueue<RpcMessage>());
            basket = map.get(address);
        }
        basket.offer(rpcMessage);
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("offer message: " + rpcMessage.getBody());
        }
        if (!isSending) {
            synchronized (mergeLock) {
                mergeLock.notifyAll();
            }
        }
    } else {
        ChannelFuture future;
        channelWriteableCheck(channel, msg);
        future = channel.writeAndFlush(rpcMessage);
        future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) {
                if (!future.isSuccess()) {
                    MessageFuture messageFuture = futures.remove(rpcMessage.getId());
                    if (messageFuture != null) {
                        messageFuture.setResultMessage(future.cause());
                    }
                    destroyChannel(future.channel());
                }
            }
        });
    }
    if (timeout > 0) {
        try {
            return messageFuture.get(timeout, TimeUnit.MILLISECONDS);
        } catch (Exception exx) {
            LOGGER.error("wait response error:" + exx.getMessage() + ",ip:" + address + ",request:" + msg);
            if (exx instanceof TimeoutException) {
                throw (TimeoutException)exx;
            } else {
                throw new RuntimeException(exx);
            }
        }
    } else {
        return null;
    }
}

先抛开方法的其它细节，比如说同步写还是异步写，以及发送频率控制。我们可以发现，这个方法其实从大角度来划分，就是如下的步骤：

1. 构造请求 message
2. 为这个请求 message 构造一个 message future
3. 发送数据
4. 阻塞等待在 message future

不过 AbstractRpcRemoting 也定义了方法用于 仅发送消息，不接收响应 的使用场景，如下所示：

/**
 * Send request.
 *
 * @param channel the channel
 * @param msg     the msg
 */
protected void sendRequest(Channel channel, Object msg) {
    RpcMessage rpcMessage = new RpcMessage();
    rpcMessage.setAsync(true);
    rpcMessage.setHeartbeat(msg instanceof HeartbeatMessage);
    rpcMessage.setRequest(true);
    rpcMessage.setBody(msg);
    rpcMessage.setId(RpcMessage.getNextMessageId());
    if (msg instanceof MergeMessage) {
        mergeMsgMap.put(rpcMessage.getId(), (MergeMessage)msg);
    }
    channelWriteableCheck(channel, msg);
    if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("write message:" + rpcMessage.getBody() + ", channel:" + channel + ",active?"
            + channel.isActive() + ",writable?" + channel.isWritable() + ",isopen?" + channel.isOpen());
    }
    channel.writeAndFlush(rpcMessage);
}

/**
 * Send response.
 *
 * @param msgId   the msg id
 * @param channel the channel
 * @param msg     the msg
 */
protected void sendResponse(long msgId, Channel channel, Object msg) {
    RpcMessage rpcMessage = new RpcMessage();
    rpcMessage.setAsync(true);
    rpcMessage.setHeartbeat(msg instanceof HeartbeatMessage);
    rpcMessage.setRequest(false);
    rpcMessage.setBody(msg);
    rpcMessage.setId(msgId);
    channelWriteableCheck(channel, msg);
    if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("send response:" + rpcMessage.getBody() + ",channel:" + channel);
    }
    channel.writeAndFlush(rpcMessage);
}

这样的场景就不需要引入 future 机制，直接调用 netty 的 api 把数据发送出去就完事了。

分析思路回到有 future 的场景，发送数据后，要在 future 上进行阻塞等待，即调用 get 方法，那 get 方法什么返回呢，我们上面说到 future 被唤醒的时候，我们先不讨论 future 的实现细节，一个 future 什么时候被唤醒呢，在这种 请求-响应 的模式下，显然是收到了响应的时候。所以我们需要查看一下 AbstractRpcRemoting 的 channelRead 方法

@Override
public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception {
    if (msg instanceof RpcMessage) {
        final RpcMessage rpcMessage = (RpcMessage)msg;
        if (rpcMessage.isRequest()) {
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug(String.format("%s msgId:%s, body:%s", this, rpcMessage.getId(), rpcMessage.getBody()));
            }
            try {
                AbstractRpcRemoting.this.messageExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            dispatch(rpcMessage.getId(), ctx, rpcMessage.getBody());
                        } catch (Throwable th) {
                            LOGGER.error(FrameworkErrorCode.NetDispatch.getErrCode(), th.getMessage(), th);
                        }
                    }
                });
            } catch (RejectedExecutionException e) {
                LOGGER.error(FrameworkErrorCode.ThreadPoolFull.getErrCode(),
                    "thread pool is full, current max pool size is " + messageExecutor.getActiveCount());
                if (allowDumpStack) {
                    String name = ManagementFactory.getRuntimeMXBean().getName();
                    String pid = name.split("@")[0];
                    int idx = new Random().nextInt(100);
                    try {
                        Runtime.getRuntime().exec("jstack " + pid + " >d:/" + idx + ".log");
                    } catch (IOException exx) {
                        LOGGER.error(exx.getMessage());
                    }
                    allowDumpStack = false;
                }
            }
        } else {
            MessageFuture messageFuture = futures.remove(rpcMessage.getId());
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug(String
                    .format("%s msgId:%s, future :%s, body:%s", this, rpcMessage.getId(), messageFuture,
                        rpcMessage.getBody()));
            }
            if (messageFuture != null) {
                messageFuture.setResultMessage(rpcMessage.getBody());
            } else {
                try {
                    AbstractRpcRemoting.this.messageExecutor.execute(new Runnable() {
                        @Override
                        public void run() {
                            try {
                                dispatch(rpcMessage.getId(), ctx, rpcMessage.getBody());
                            } catch (Throwable th) {
                                LOGGER.error(FrameworkErrorCode.NetDispatch.getErrCode(), th.getMessage(), th);
                            }
                        }
                    });
                } catch (RejectedExecutionException e) {
                    LOGGER.error(FrameworkErrorCode.ThreadPoolFull.getErrCode(),
                        "thread pool is full, current max pool size is " + messageExecutor.getActiveCount());
                }
            }
        }
    }
}

可以看到调用了 messageFuture 当 setResultMessage() 方法，设置 future 的结果，也就是说，唤醒了 future，那么阻塞在 future 的 get 方法上的线程就被唤醒了，得到结果，继续往下执行。

接下来我们讨论 MessageFuture 的实现细节，其实 seata 里面有很多种 future 相关的类，实现方式也不太一样，不过都大同小异，有的是基于 CompletableFuture 实现，有的是基于 CountDownLatch 实现。比如说，MessageFuture 就是基于 CompletableFuture 实现的，先看看它的成员变量：

private RpcMessage requestMessage;
private long timeout;
private long start = System.currentTimeMillis();
private transient CompletableFuture origin = new CompletableFuture();

CompletableFuture 是它的一个成员变量，它被利用来阻塞当前线程。MessageFuture 的 get 方法，依赖于 CompletableFuture 的 get 方法，来实现有一定时间限制的等待，直到另一个线程唤醒 CompletableFuture。如下所示：

/**
 * Get object.
 *
 * @param timeout the timeout
 * @param unit    the unit
 * @return the object
 * @throws TimeoutException the timeout exception
 * @throws InterruptedException the interrupted exception
 */
public Object get(long timeout, TimeUnit unit) throws TimeoutException,
    InterruptedException {
    Object result = null;
    try {
        result = origin.get(timeout, unit);
    } catch (ExecutionException e) {
        throw new ShouldNeverHappenException("Should not get results in a multi-threaded environment", e);
    } catch (TimeoutException e) {
        throw new TimeoutException("cost " + (System.currentTimeMillis() - start) + " ms");
    }

    if (result instanceof RuntimeException) {
        throw (RuntimeException)result;
    } else if (result instanceof Throwable) {
        throw new RuntimeException((Throwable)result);
    }

    return result;
}

/**
 * Sets result message.
 *
 * @param obj the obj
 */
public void setResultMessage(Object obj) {
    origin.complete(obj);
}

既然说到了 future 机制，这里也顺便把 io.seata.config.ConfigFuture 提一下，它就是上面提到的基于 CountDownLatch 实现的一种 future 机制，虽然实现方式两者不一样，但完成的功能和作用是一样的。

private final CountDownLatch latch = new CountDownLatch(1);

/**
 * Get object.
 *
 * @param timeout the timeout
 * @param unit    the unit
 * @return the object
 * @throws InterruptedException the interrupted exception
 */
public Object get(long timeout, TimeUnit unit) {
    this.timeoutMills = unit.toMillis(timeout);
    try {
        boolean success = latch.await(timeout, unit);
        if (!success) {
            LOGGER.error(
                "config operation timeout,cost:" + (System.currentTimeMillis() - start) + " ms,op:" + operation
                    .name()
                    + ",dataId:" + dataId);
            return getFailResult();
        }
    } catch (InterruptedException exx) {
        LOGGER.error("config operate interrupted,error:" + exx.getMessage());
        return getFailResult();
    }
    if (operation == ConfigOperation.GET) {
        return result == null ? content : result;
    } else {
        return result == null ? Boolean.FALSE : result;
    }
}

/**
 * Sets result.
 *
 * @param result the result
 */
public void setResult(Object result) {
    this.result = result;
    latch.countDown();
}

阻塞操作调用了 CountDownLatch 的 await 方法，而唤醒操作则调用 countDown 方法，核心在于需要把 CountDownLatch 的 latch 值设置为 1。

实际上，Java 语言本身已经提供了 java.util.concurrent.Future 这个类来提供 Future 机制，但 Java 原生的 Future 机制功能过于单一，比如说不能主动设置 future 的结果，也不能为它添加 listener，所有有许多像 seata 这样的软件，会选择去重新实现一种 future 机制来满足异步转同步的需求。也有像 netty 这样的软件，它不会借助类似于 countdownlatch 来实现，而是直接扩展 java.util.concurrent.Future，在它的基础上添加功能。

防洪机制

在 AbstractRpcRemoting 中，往外发数据的时候，它都会先进行一个检查，即检查当前的 channel 是否可写。

private void channelWriteableCheck(Channel channel, Object msg) {
    int tryTimes = 0;
    synchronized (lock) {
        while (!channel.isWritable()) {
            try {
                tryTimes++;
                if (tryTimes > NettyClientConfig.getMaxNotWriteableRetry()) {
                    destroyChannel(channel);
                    throw new FrameworkException("msg:" + ((msg == null) ? "null" : msg.toString()),
                        FrameworkErrorCode.ChannelIsNotWritable);
                }
                lock.wait(NOT_WRITEABLE_CHECK_MILLS);
            } catch (InterruptedException exx) {
                LOGGER.error(exx.getMessage());
            }
        }
    }
}

这要从 netty 的内部机制说起，当调用 ChannelHandlerContext 或者 Channel 的 write 方法时，netty 只是把要写的数据放入了自身的一个环形队列里面，再由后台线程真正往链路上发。如果接受方的处理速度慢，也就是说，接收的速度慢，那么根据 tcpip 协议的滑动窗口机制，它也会导致发送方发送得慢。

我们可以把 netty 的环形队列想像成一个水池，调用 write 方法往池子里加水，netty 通过后台线程，慢慢把池子的水流走。这就有可能出现一种情况，由于池子水流走的速度远远慢于往池子里加水的速度，这样会导致池子的总水量随着时间的推移越来越多。所以往池子里加水时应该考虑当前池子里的水量，否则最终会导致应用的内存溢出。

netty 对于水池提供了两个设置，一个是 高水位 ，一个是 低水位 ，当池子里的水高于高水位时，这个时候 channel.isWritable() 返回 false，并且直到水位慢慢降回到低水位时，这个方法才会返回 true。

上述的 channelWriteableCheck 方法，发现channel 不可写的时候，进入循环等待，等待的目的是让池子的水位下降到 low water mark，如果等待超过最大允许等待的时间，那么将会抛出异常并关闭连接。

消息队列

在 AbstractRpcRemoting 中，发送数据有两种方式，一种是直接调用 channel 往外写，另一种是先把数据放进“数据篮子”里，它实际上是一个 map， key 为远端地址，value为一个消息队列。数据放队列后，再由其它线程往外发。下面是 sendAsycRequest 方法的一部分代码，显示了这种机制：

ConcurrentHashMap<String, BlockingQueue<RpcMessage>> map = basketMap;
BlockingQueue<RpcMessage> basket = map.get(address);
if (basket == null) {
    map.putIfAbsent(address, new LinkedBlockingQueue<RpcMessage>());
    basket = map.get(address);
}
basket.offer(rpcMessage);
if (LOGGER.isDebugEnabled()) {
    LOGGER.debug("offer message: " + rpcMessage.getBody());
}
if (!isSending) {
    synchronized (mergeLock) {
        mergeLock.notifyAll();
    }
}

但我们在 AbstractRpcRemoting 里面没有看有任何额外的线程在晴空这个 basketMap。回顾一下上面的 RpcServer 的类继承体系，接下来我们要分析一下，AbstractRpcRemotingServer 这个类。

AbstractRpcRemotingServer 这个类主要定义了于netty 启动一个 server bootstrap 相关的类，可见真正启动服务监听端口的是在这个类，先看一下它的start方法

@Override
public void start() {
    this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupWorker)
        .channel(nettyServerConfig.SERVER_CHANNEL_CLAZZ)
        .option(ChannelOption.SO_BACKLOG, nettyServerConfig.getSoBackLogSize())
        .option(ChannelOption.SO_REUSEADDR, true)
        .childOption(ChannelOption.SO_KEEPALIVE, true)
        .childOption(ChannelOption.TCP_NODELAY, true)
        .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSendBufSize())
        .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketResvBufSize())
        .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK,
            new WriteBufferWaterMark(nettyServerConfig.getWriteBufferLowWaterMark(),
                nettyServerConfig.getWriteBufferHighWaterMark()))
        .localAddress(new InetSocketAddress(listenPort))
        .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) {
                ch.pipeline().addLast(new IdleStateHandler(nettyServerConfig.getChannelMaxReadIdleSeconds(), 0, 0))
                    .addLast(new MessageCodecHandler());
                if (null != channelHandlers) {
                    addChannelPipelineLast(ch, channelHandlers);
                }

            }
        });

    if (nettyServerConfig.isEnableServerPooledByteBufAllocator()) {
        this.serverBootstrap.childOption(ChannelOption.ALLOCATOR, NettyServerConfig.DIRECT_BYTE_BUF_ALLOCATOR);
    }

    try {
        ChannelFuture future = this.serverBootstrap.bind(listenPort).sync();
        LOGGER.info("Server started ... ");
        RegistryFactory.getInstance().register(new InetSocketAddress(XID.getIpAddress(), XID.getPort()));
        initialized.set(true);
        future.channel().closeFuture().sync();
    } catch (Exception exx) {
        throw new RuntimeException(exx);
    }
}

这个类很常规，就是遵循 netty 的使用规范，用合适的配置启动一个 server，并调用注册中心 api 把自己作为一个服务发布出去。

我们可以看到，配置中确实也出现了我们上文中提到过的上下水位的配置。

另外，channelpipeline 中，除了添加一个保持链路有效性探测的 IdleStateHandler，和一个 MessageCodec，处理事务逻辑相关的 Handler 还需要由参数传入。

接下来我们看 RpcServer 这个类，从它的 init 方法里，我们可以看到，它把自己做为一个 ChannelHandler，加入到了 channel pipeline 中

/**
 * Init.
 */
@Override
public void init() {
    super.init();
    setChannelHandlers(RpcServer.this);
    DefaultServerMessageListenerImpl defaultServerMessageListenerImpl = new DefaultServerMessageListenerImpl(
        transactionMessageHandler);
    defaultServerMessageListenerImpl.init();
    defaultServerMessageListenerImpl.setServerMessageSender(this);
    this.setServerMessageListener(defaultServerMessageListenerImpl);
    super.start();

}

RpcServer 自身也实现了 channelRead 方法，但它只处理心跳相关的信息和注册相关的信息，其它的业务消息，它交给父类处理，而先前我们也已经看到，父类的channelRead

方法里，反过来会调用 dispatch 这个抽象方法去做消息的分发，而 RpcServer 类实现了这个抽象方法，在接收到不同的消息类型是，采取不同的处理流程。

关于事务的处理流程的细节，本篇文章暂不涉及，后续文章再慢慢分析。

行文至此，回想我们先前提到的一个疑惑：

“当我启动一个 RpcServer 的时候，我是真的在启动一个 server 吗？看起来我好像在启动一个 ChannelHandler，可是 ChannelHandler 怎么谈得上‘启动’呢？”

是的，我们既在启动一个 server，这个 server 也实现了事务处理逻辑，它同时也是个 ChannelHandler。

没有一定的事实标准去衡量这样写的代码是好是坏，我们也没必要去争论 Effective Java 提到的什么时候该用组合，什么时候该用继承。

本文到此结束。