阿里开源分布式事务组件 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 给调用起来。

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);

}

这个定时任务的逻辑也比较简单：扫描   ConcurrentHashMap<Long, MessageFuture> futures   这个成员变量里的 MessageFuture，如果这个 Future 超时了，就将 Future 的结果设置为 null。逻辑虽然简单，但这个功能涉及到了异步通信里一个很常见的功能，即 异步转同步 的功能。  

在 netty 这种基于 NIO 的通信方式中，数据的发送，接收，全部是非阻塞的，因此判断一个动作完成与否，并不能像传统的 Java 同步代码一样，代码执行完了就认为相应的动作也真正完成了，例如，在 netty 中，如果通过 channel.write(); 方法往对端发送一个数据，这个方法执行完了，并不代表数据发送出去了，channel.write() 方法会返回一个 future，应用代码应该利用这个 future ，通过这个 future 可以知道数据到底发送出去了没有，也可以为这个 future 添加动作完成后的回调逻辑，也可以阻塞等待这个 future 所关联的动作执行完毕。  

在 seata 中，存在着 发送一个请求，并等待相应 这样的使用场景，上层的 api 可能是这么定义的：  

public Response request(Request request) {}  

而基于 nio 的底层数据发送逻辑却是这样的：

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());

}

}

}

}

上述的 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 提到的什么时候该用组合，什么时候该用继承。

本文到此结束。