远程通讯——Netty3
目标：介绍基于netty3的来实现的远程通信、介绍dubbo-remoting-netty内的源码解析。
前言
现在dubbo默认的网络传输Transport接口默认实现的还是基于netty3实现的网络传输，不过马上后面默认实现就要改为netty4了。由于netty4对netty3对兼容性不是很好，所以保留了两个版本的实现。
下面是包结构：
源码分析
（一）NettyChannel
该类继承了AbstractChannel类，是基于netty3实现的通道。
1.属性
/**
* 通道集合
*/
private static final ConcurrentMap<org.jboss.netty.channel.Channel, NettyChannel> channelMap = new ConcurrentHashMap<org.jboss.netty.channel.Channel, NettyChannel>();
/**
* 通道
*/
private final org.jboss.netty.channel.Channel channel;
/**
* 属性集合
*/
private final Map<String, Object> attributes = new ConcurrentHashMap<String, Object>();
2.getOrAddChannel
static NettyChannel getOrAddChannel(org.jboss.netty.channel.Channel ch, URL url, ChannelHandler handler) {
if (ch == null) {
return null;
}
// 首先从集合中取通道
NettyChannel ret = channelMap.get(ch);
// 如果为空，则新建
if (ret == null) {
NettyChannel nc = new NettyChannel(ch, url, handler);
// 如果通道连接着
if (ch.isConnected()) {
// 加入集合
ret = channelMap.putIfAbsent(ch, nc);
}
if (ret == null) {
ret = nc;
}
}
return ret;
}
该方法是获得通道，当通道在集合中没有的时候，新建一个通道。
3.removeChannelIfDisconnected
static void removeChannelIfDisconnected(org.jboss.netty.channel.Channel ch) {
if (ch != null && !ch.isConnected()) {
channelMap.remove(ch);
}
}
该方法是当通道没有连接的时候，从集合中移除它。
4.send
@Override
public void send(Object message, boolean sent) throws RemotingException {
super.send(message, sent);
boolean success = true;
int timeout = 0;
try {
// 写入数据，发送消息
ChannelFuture future = channel.write(message);
// 如果已经发送过
if (sent) {
// 获得超时时间
timeout = getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
// 等待timeout的连接时间后查看是否发送成功
success = future.await(timeout);
}
// 看是否有异常
Throwable cause = future.getCause();
if (cause != null) {
throw cause;
}
} catch (Throwable e) {
throw new RemotingException(this, "Failed to send message " + message + " to " + getRemoteAddress() + ", cause: " + e.getMessage(), e);
}
if (!success) {
throw new RemotingException(this, "Failed to send message " + message + " to " + getRemoteAddress()
+ "in timeout(" + timeout + "ms) limit");
}
}
该方法是发送消息，其中用到了channe.write方法传输消息，并且通过返回的future来判断是否发送成功。
5.close
@Override
public void close() {
try {
super.close();
} catch (Exception e) {
logger.warn(e.getMessage(), e);
}
try {
// 如果通道断开，则移除该通道
removeChannelIfDisconnected(channel);
} catch (Exception e) {
logger.warn(e.getMessage(), e);
}
try {
// 清空属性
attributes.clear();
} catch (Exception e) {
logger.warn(e.getMessage(), e);
}
try {
if (logger.isInfoEnabled()) {
logger.info("Close netty channel " + channel);
}
// 关闭通道
channel.close();
} catch (Exception e) {
logger.warn(e.getMessage(), e);
}
}
该方法是关闭通道，做了三个操作，分别是从集合中移除、清除属性、关闭通道。
其他实现方法比较简单，我就讲解了。
（二）NettyHandler
该类继承了SimpleChannelHandler类，是基于netty3的通道处理器，而该类被加上了@Sharable注解，也就是说该处理器可以从属于多个ChannelPipeline
1.属性
/**
* 通道集合，key是主机地址 ip:port
*/
private final Map<String, Channel> channels = new ConcurrentHashMap<String, Channel>(); // <ip:port, channel>
/**
* url对象
*/
private final URL url;
/**
* 通道
*/
private final ChannelHandler handler;
该类的属性比较简单，并且该类中实现的方法都是调用了属性handler的方法，我举一个例子来讲，其他的可以自己查看源码，比较简单。
@Override
public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
// 获得通道实例
NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
try {
if (channel != null) {
// 保存该通道，加入到集合中
channels.put(NetUtils.toAddressString((InetSocketAddress) ctx.getChannel().getRemoteAddress()), channel);
}
// 连接
handler.connected(channel);
} finally {
NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
}
}
该方法是通道连接的方法，其中先获取了通道实例，然后吧该实例加入到集合中，最好带哦用handler.connected来进行连接。
（三）NettyClient
该类继承了AbstractClient，是基于netty3实现的客户端类。
1.属性
private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);
// ChannelFactory's closure has a DirectMemory leak, using static to avoid
// https://issues.jboss.org/browse/NETTY-424
/**
* 通道工厂，用static来避免直接缓存区的一个OOM问题
*/
private static final ChannelFactory channelFactory = new NioClientSocketChannelFactory(Executors.newCachedThreadPool(new NamedThreadFactory("NettyClientBoss", true)),
Executors.newCachedThreadPool(new NamedThreadFactory("NettyClientWorker", true)),
Constants.DEFAULT_IO_THREADS);
/**
* 客户端引导对象
*/
private ClientBootstrap bootstrap;
/**
* 通道
*/
private volatile Channel channel; // volatile, please copy reference to use
上述属性中ChannelFactory用了static修饰，为了避免netty3中会有直接缓冲内存泄漏的现象，具体的讨论可以访问注释中的讨论。
2.doOpen
@Override
protected void doOpen() throws Throwable {
// 设置日志工厂
NettyHelper.setNettyLoggerFactory();
// 实例化客户端引导类
bootstrap = new ClientBootstrap(channelFactory);
// config
// @see org.jboss.netty.channel.socket.SocketChannelConfig
// 配置选择项
bootstrap.setOption("keepAlive", true);
bootstrap.setOption("tcpNoDelay", true);
bootstrap.setOption("connectTimeoutMillis", getConnectTimeout());
// 创建通道处理器
final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
// 设置责任链路
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
/**
* 获得通道
* @return
*/
@Override
public ChannelPipeline getPipeline() {
// 新建编解码
NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this);
// 获得管道
ChannelPipeline pipeline = Channels.pipeline();
// 设置解码器
pipeline.addLast("decoder", adapter.getDecoder());
// 设置编码器
pipeline.addLast("encoder", adapter.getEncoder());
// 设置通道处理器
pipeline.addLast("handler", nettyHandler);
// 返回通道
return pipeline;
}
});
}
该方法是创建客户端，并且打开，其中的逻辑就是用netty3的客户端引导类来创建一个客户端，如果对netty不熟悉的朋友可以先补补netty知识。
3.doConnect
@Override
protected void doConnect() throws Throwable {
long start = System.currentTimeMillis();
// 用引导类连接
ChannelFuture future = bootstrap.connect(getConnectAddress());
try {
// 在超时时间内是否连接完成
boolean ret = future.awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS);
if (ret && future.isSuccess()) {
// 获得通道
Channel newChannel = future.getChannel();
// 异步修改此通道
newChannel.setInterestOps(Channel.OP_READ_WRITE);
try {
// Close old channel 关闭旧的通道
Channel oldChannel = NettyClient.this.channel; // copy reference
if (oldChannel != null) {
try {
if (logger.isInfoEnabled()) {
logger.info("Close old netty channel " + oldChannel + " on create new netty channel " + newChannel);
}
// 关闭
oldChannel.close();
} finally {
// 移除通道
NettyChannel.removeChannelIfDisconnected(oldChannel);
}
}
} finally {
// 如果客户端关闭
if (NettyClient.this.isClosed()) {
try {
if (logger.isInfoEnabled()) {
logger.info("Close new netty channel " + newChannel + ", because the client closed.");
}
// 关闭通道
newChannel.close();
} finally {
NettyClient.this.channel = null;
NettyChannel.removeChannelIfDisconnected(newChannel);
}
} else {
NettyClient.this.channel = newChannel;
}
}
} else if (future.getCause() != null) {
throw new RemotingException(this, "client(url: " + getUrl() + ") failed to connect to server "
+ getRemoteAddress() + ", error message is:" + future.getCause().getMessage(), future.getCause());
} else {
throw new RemotingException(this, "client(url: " + getUrl() + ") failed to connect to server "
+ getRemoteAddress() + " client-side timeout "
+ getConnectTimeout() + "ms (elapsed: " + (System.currentTimeMillis() - start) + "ms) from netty client "
+ NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion());
}
} finally {
// 如果客户端没有连接
if (!isConnected()) {
// 取消future
future.cancel();
}
}
}
该方法是客户端连接服务器的方法。其中调用了bootstrap.connect。后面的逻辑是用来检测是否连接，最后如果未连接，则会取消该连接任务。
4.doClose
@Override
protected void doClose() throws Throwable {
/*try {
bootstrap.releaseExternalResources();
} catch (Throwable t) {
logger.warn(t.getMessage());
}*/
}
在这里不能关闭是因为channelFactory 是静态属性，被多个 NettyClient 共用。所以不能释放资源。
（四）NettyServer
该类继承了AbstractServer，实现了Server，是基于netty3实现的服务器类。
1.属性
/**
* 连接该服务器的通道集合
*/
private Map<String, Channel> channels; // <ip:port, channel>
/**
* 服务器引导类对象
*/
private ServerBootstrap bootstrap;
/**
* 通道
*/
private org.jboss.netty.channel.Channel channel;
2.doOpen
@Override
protected void doOpen() throws Throwable {
// 设置日志工厂
NettyHelper.setNettyLoggerFactory();
// 创建线程池
ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));
ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));
// 新建通道工厂
ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
// 新建服务引导类对象
bootstrap = new ServerBootstrap(channelFactory);
// 新建通道处理器
final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
// 获得通道集合
channels = nettyHandler.getChannels();
// https://issues.jboss.org/browse/NETTY-365
// https://issues.jboss.org/browse/NETTY-379
// final Timer timer = new HashedWheelTimer(new NamedThreadFactory("NettyIdleTimer", true));
// 禁用nagle算法，将数据立即发送出去。纳格算法是以减少封包传送量来增进TCP/IP网络的效能
bootstrap.setOption("child.tcpNoDelay", true);
// 设置管道工厂
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
/**
* 获得通道
* @return
*/
@Override
public ChannelPipeline getPipeline() {
// 新建编解码器
NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
// 获得通道
ChannelPipeline pipeline = Channels.pipeline();
/*int idleTimeout = getIdleTimeout();
if (idleTimeout > 10000) {
pipeline.addLast("timer", new IdleStateHandler(timer, idleTimeout / 1000, 0, 0));
}*/
// 设置解码器
pipeline.addLast("decoder", adapter.getDecoder());
// 设置编码器
pipeline.addLast("encoder", adapter.getEncoder());
// 设置通道处理器
pipeline.addLast("handler", nettyHandler);
// 返回通道
return pipeline;
}
});
// bind 绑定地址，也就是启用服务器
channel = bootstrap.bind(getBindAddress());
}
该方法是创建服务器，并且打开服务器。同样创建服务器的方式跟正常的用netty创建服务器方式一样，只是新加了编码器和解码器。还有一个注意点就是这里ServerBootstrap 的可选项。
3.doClose
@Override
protected void doClose() throws Throwable {
try {
if (channel != null) {
// unbind.关闭通道
channel.close();
}
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
}
try {
// 获得所有连接该服务器的通道集合
Collection<com.alibaba.dubbo.remoting.Channel> channels = getChannels();
if (channels != null && !channels.isEmpty()) {
// 遍历通道集合
for (com.alibaba.dubbo.remoting.Channel channel : channels) {
try {
// 关闭通道连接
channel.close();
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
}
}
}
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
}
try {
if (bootstrap != null) {
// release external resource. 回收资源
bootstrap.releaseExternalResources();
}
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
}
try {
if (channels != null) {
// 清空集合
channels.clear();
}
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
}
}
该方法是关闭服务器，一系列的操作很清晰，我就不多说了。
4.getChannels
@Override
public Collection<Channel> getChannels() {
Collection<Channel> chs = new HashSet<Channel>();
for (Channel channel : this.channels.values()) {
// 如果通道连接，则加入集合，返回
if (channel.isConnected()) {
chs.add(channel);
} else {
channels.remove(NetUtils.toAddressString(channel.getRemoteAddress()));
}
}
return chs;
}
该方法是返回连接该服务器的通道集合，并且用了HashSet保存，不会重复。
（五）NettyTransporter
public class NettyTransporter implements Transporter {
public static final String NAME = "netty";
@Override
public Server bind(URL url, ChannelHandler listener) throws RemotingException {
// 创建一个NettyServer
return new NettyServer(url, listener);
}
@Override
public Client connect(URL url, ChannelHandler listener) throws RemotingException {
// 创建一个NettyClient
return new NettyClient(url, listener);
}
}
该类就是基于netty3的Transporter实现类，同样两个方法也是分别创建了NettyServer和NettyClient。
（六）NettyHelper
该类是设置日志的工具类，其中基于netty3的InternalLoggerFactory实现类一个DubboLoggerFactory。这个我就不讲解了，比较好理解，不理解也无伤大雅。
（七）NettyCodecAdapter
该类是基于netty3实现的编解码类。
1.属性
/**
* 编码者
*/
private final ChannelHandler encoder = new InternalEncoder();
/**
* 解码者
*/
private final ChannelHandler decoder = new InternalDecoder();
/**
* 编解码器
*/
private final Codec2 codec;
/**
* url对象
*/
private final URL url;
/**
* 缓冲区大小
*/
private final int bufferSize;
/**
* 通道对象
*/
private final com.alibaba.dubbo.remoting.ChannelHandler handler;
InternalEncoder和InternalDecoder属性是该类的内部类，分别掌管着编码和解码
2.构造方法
public NettyCodecAdapter(Codec2 codec, URL url, com.alibaba.dubbo.remoting.ChannelHandler handler) {
this.codec = codec;
this.url = url;
this.handler = handler;
int b = url.getPositiveParameter(Constants.BUFFER_KEY, Constants.DEFAULT_BUFFER_SIZE);
// 如果缓存区大小在16字节以内，则设置配置大小，如果不是，则设置8字节的缓冲区大小
this.bufferSize = b >= Constants.MIN_BUFFER_SIZE && b <= Constants.MAX_BUFFER_SIZE ? b : Constants.DEFAULT_BUFFER_SIZE;
}
你会发现对于缓存区大小的规则都是一样的。
3.InternalEncoder
@Sharable
private class InternalEncoder extends OneToOneEncoder {
@Override
protected Object encode(ChannelHandlerContext ctx, Channel ch, Object msg) throws Exception {
// 动态分配一个1k的缓冲区
com.alibaba.dubbo.remoting.buffer.ChannelBuffer buffer =
com.alibaba.dubbo.remoting.buffer.ChannelBuffers.dynamicBuffer(1024);
// 获得通道对象
NettyChannel channel = NettyChannel.getOrAddChannel(ch, url, handler);
try {
// 编码
codec.encode(channel, buffer, msg);
} finally {
NettyChannel.removeChannelIfDisconnected(ch);
}
// 基于buteBuffer创建一个缓冲区，并且写入数据
return ChannelBuffers.wrappedBuffer(buffer.toByteBuffer());
}
}
该内部类实现类编码的逻辑，主要调用了codec.encode。
4.InternalDecoder
private class InternalDecoder extends SimpleChannelUpstreamHandler {
private com.alibaba.dubbo.remoting.buffer.ChannelBuffer buffer =
com.alibaba.dubbo.remoting.buffer.ChannelBuffers.EMPTY_BUFFER;
@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent event) throws Exception {
Object o = event.getMessage();
// 如果消息不是一个ChannelBuffer类型
if (!(o instanceof ChannelBuffer)) {
// 转发事件到与此上下文关联的处理程序最近的上游
ctx.sendUpstream(event);
return;
}
ChannelBuffer input = (ChannelBuffer) o;
// 如果可读数据不大于0，直接返回
int readable = input.readableBytes();
if (readable <= 0) {
return;
}
com.alibaba.dubbo.remoting.buffer.ChannelBuffer message;
if (buffer.readable()) {
// 判断buffer是否是动态分配的缓冲区
if (buffer instanceof DynamicChannelBuffer) {
// 写入数据
buffer.writeBytes(input.toByteBuffer());
message = buffer;
} else {
// 需要的缓冲区大小
int size = buffer.readableBytes() + input.readableBytes();
// 动态生成缓冲区
message = com.alibaba.dubbo.remoting.buffer.ChannelBuffers.dynamicBuffer(
size > bufferSize ? size : bufferSize);
// 把buffer数据写入message
message.writeBytes(buffer, buffer.readableBytes());
// 把input数据写入message
message.writeBytes(input.toByteBuffer());
}
} else {
// 否则 基于ByteBuffer通过buffer来创建一个新的缓冲区
message = com.alibaba.dubbo.remoting.buffer.ChannelBuffers.wrappedBuffer(
input.toByteBuffer());
}
NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
Object msg;
int saveReaderIndex;
try {
// decode object.
do {
saveReaderIndex = message.readerIndex();
try {
// 解码
msg = codec.decode(channel, message);
} catch (IOException e) {
buffer = com.alibaba.dubbo.remoting.buffer.ChannelBuffers.EMPTY_BUFFER;
throw e;
}
// 拆包
if (msg == Codec2.DecodeResult.NEED_MORE_INPUT) {
message.readerIndex(saveReaderIndex);
break;
} else {
// 如果已经到达读索引，则没有数据可解码
if (saveReaderIndex == message.readerIndex()) {
buffer = com.alibaba.dubbo.remoting.buffer.ChannelBuffers.EMPTY_BUFFER;
throw new IOException("Decode without read data.");
}
//
if (msg != null) {
// 将消息发送到指定关联的处理程序最近的上游
Channels.fireMessageReceived(ctx, msg, event.getRemoteAddress());
}
}
} while (message.readable());
} finally {
// 如果消息还有可读数据，则丢弃
if (message.readable()) {
message.discardReadBytes();
buffer = message;
} else {
buffer = com.alibaba.dubbo.remoting.buffer.ChannelBuffers.EMPTY_BUFFER;
}
NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
ctx.sendUpstream(e);
}
}
该内部类实现了解码的逻辑，其中大部分逻辑都在对数据做读写，关键的解码调用了codec.decode。
（八）NettyBackedChannelBufferFactory
该类是创建缓冲区的工厂类。它实现了ChannelBufferFactory接口，也就是实现类它的三种获得缓冲区的方法。
public class NettyBackedChannelBufferFactory implements ChannelBufferFactory {
/**
* 单例
*/
private static final NettyBackedChannelBufferFactory INSTANCE = new NettyBackedChannelBufferFactory();
public static ChannelBufferFactory getInstance() {
return INSTANCE;
}
@Override
public ChannelBuffer getBuffer(int capacity) {
return new NettyBackedChannelBuffer(ChannelBuffers.dynamicBuffer(capacity));
}
@Override
public ChannelBuffer getBuffer(byte[] array, int offset, int length) {
org.jboss.netty.buffer.ChannelBuffer buffer = ChannelBuffers.dynamicBuffer(length);
buffer.writeBytes(array, offset, length);
return new NettyBackedChannelBuffer(buffer);
}
@Override
public ChannelBuffer getBuffer(ByteBuffer nioBuffer) {
return new NettyBackedChannelBuffer(ChannelBuffers.wrappedBuffer(nioBuffer));
}
}
可以看到，都是创建了一个NettyBackedChannelBuffer，下面讲解NettyBackedChannelBuffer。
（九）NettyBackedChannelBuffer
该类是基于netty3的buffer重新实现的缓冲区，它实现了ChannelBuffer接口，并且有一个属性：
private org.jboss.netty.buffer.ChannelBuffer buffer;
那么其中的几乎所有方法都是调用了这个buffer的方法，因为我在dubbo源码解析（十一）远程通信——Buffer中写到ChannelBuffer接口方法定义跟netty中的缓冲区定义几乎一样，连注释都几乎一样。所有知识单纯的调用了buffer的方法。具体的代码可以查看我的GitHub
后记
该部分相关的源码解析地址：github.com/CrazyHZM/in…
该文章讲解了基于netty3的来实现的远程通信、介绍dubbo-remoting-netty内的源码解析，关键需要对netty有所了解。下一篇我会讲解基于netty4实现远程通信部分。
