Netty中的Channel之数据冲刷与线程安全(writeAndFlush)

37次阅读

共计 5509 个字符,预计需要花费 14 分钟才能阅读完成。

本文首发个人博客:猫叔的博客 | MySelf

GitHub 项目地址

InChat
一个轻量级、高效率的支持多端(应用与硬件 Iot)的异步网络应用通讯框架
前言
本文预设读者已经了解了一定的 Netty 基础知识,并能够自己构建一个 Netty 的通信服务(包括客户端与服务端)。那么你一定使用到了 Channel,这是 Netty 对传统 JavaIO、NIO 的链接封装实例。
那么接下来让我们来了解一下关于 Channel 的数据冲刷与线程安全吧。
数据冲刷的步骤
1、获取一个链接实例
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// 获取链接实例
Channel channel = ctx.channel();
}
我将案例放在初学者最熟悉的 channelRead 方法中,这是一个数据接收的方法,我们自实现 Netty 的消息处理接口时需要重写的方法。即客户端发送消息后,这个方法会被触发调用,所以我们在这个方法中进行本次内容的讲解。
由上一段代码,其实目前还是很简单,我们借助 ChannelHandlerContext(这是一个 ChannelHandler 与 ChannelPipeline 相交互并对接的一个对象。如下是源码的解释)来获取目前的链接实例 Channel。
/* Enables a {@link ChannelHandler} to interact with its {@link ChannelPipeline}
* and other handlers. Among other things a handler can notify the next {@link ChannelHandler} in the
* {@link ChannelPipeline} as well as modify the {@link ChannelPipeline} it belongs to dynamically.
*/
public interface ChannelHandlerContext extends AttributeMap, ChannelInboundInvoker, ChannelOutboundInvoker {
//……
}
2、创建一个持有数据的 ByteBuf
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// 获取链接实例
Channel channel = ctx.channel();
// 创建一个持有数据的 ByteBuf
ByteBuf buf = Unpooled.copiedBuffer(“data”, CharsetUtil.UTF_8);
}
ByteBuf 又是什么呢?
它是 Netty 框架自己封装的一个字符底层对象,是一个对 byte[] 和 ByteBuffer NIO 的抽象类,更官网的说就是“零个或多个字节的随机和顺序可访问的序列。”,如下是源码的解释
/**
* A random and sequential accessible sequence of zero or more bytes (octets).
* This interface provides an abstract view for one or more primitive byte
* arrays ({@code byte[]}) and {@linkplain ByteBuffer NIO buffers}.
*/
public abstract class ByteBuf implements ReferenceCounted, Comparable<ByteBuf> {
//……
}
由上一段源码可以看出,ByteBuf 是一个抽象类,所以我们不能通过 new 的形式来创建一个新的 ByteBuf 对象。那么我们可以通过 Netty 提供的一个 final 的工具类 Unpooled(你将其看作是一个创建 ByteBuf 的工具类就好了)。
/**
* Creates a new {@link ByteBuf} by allocating new space or by wrapping
* or copying existing byte arrays, byte buffers and a string.
*/
public final class Unpooled {
//……
}
这真是一个有趣的过程,那么接下来我们仅需要再看看 copiedBuffer 这个方法了。这个方法相对简单,就是我们将创建一个新的缓冲区,其内容是我们指定的 UTF- 8 字符集 编码指定的“data”,同时这个新的缓冲区的读索引和写索引分别是 0 和字符串的长度。
3、冲刷数据
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// 获取链接实例
Channel channel = ctx.channel();
// 创建一个持有数据的 ByteBuf
ByteBuf buf = Unpooled.copiedBuffer(“data”, CharsetUtil.UTF_8);
// 数据冲刷
channel.writeAndFlush(buf);
}
我相信大部分人都是直接这么写的,因为我们经常理所当然的启动测试,并在客户端接受到了这个“data”消息。那么我们是否应该注意一下,这个数据冲刷会返回一个什么值,我们要如何才能在服务端知道,这次数据冲刷是成功还是失败呢?
那么其实 Netty 框架已经考虑到了这个点,本次数据冲刷我们将得到一个 ChannelFuture。
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// 获取链接实例
Channel channel = ctx.channel();
// 创建一个持有数据的 ByteBuf
ByteBuf buf = Unpooled.copiedBuffer(“data”, CharsetUtil.UTF_8);
// 数据冲刷
ChannelFuture cf = channel.writeAndFlush(buf);
}
是的,他就是 Channel 异步 IO 操作的结果,它是一个接口,并继承了 Future<V>。(如下为源码的解释)
/**
* The result of an asynchronous {@link Channel} I/O operation.
*/
public interface ChannelFuture extends Future<Void> {
//……
}
既然如此,那么我们可以明显的知道我们可以对其添加对应的监听。
4、异步回调结果监听
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// 获取链接实例
Channel channel = ctx.channel();
// 创建一个持有数据的 ByteBuf
ByteBuf buf = Unpooled.copiedBuffer(“data”, CharsetUtil.UTF_8);
// 数据冲刷
ChannelFuture cf = channel.writeAndFlush(buf);
// 添加 ChannelFutureListener 以便在写操作完成后接收通知
cf.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
// 写操作完成,并没有错误发生
if (future.isSuccess()){
System.out.println(“successful”);
}else{
// 记录错误
System.out.println(“error”);
future.cause().printStackTrace();
}
}
});
}
好的,我们可以简单的从代码理解到,我们将通过对异步 IO 的结果监听,得到本次运行的结果。我想这才是一个相对完整的 数据冲刷(writeAndFlush)。
测试线程安全的流程
对于线程安全的测试,我们将模拟多个线程去执行数据冲刷操作,我们可以用到 Executor。
我们可以这样理解 Executor,是一种省略了线程启用与调度的方式,你只需要传递一个 Runnable 给它即可,你不再需要去 start 一个线程。(如下是源码的解释)
/**
* An object that executes submitted {@link Runnable} tasks. This
* interface provides a way of decoupling task submission from the
* mechanics of how each task will be run, including details of thread
* use, scheduling, etc. An {@code Executor} is normally used
* instead of explicitly creating threads. For example, rather than
* invoking {@code new Thread(new(RunnableTask())).start()} for each
* of a set of tasks, you might use:…
*/
public interface Executor {
//……
}
那么我们的测试代码,大致是这样的。
final Channel channel = ctx.channel();
// 创建要写数据的 ByteBuf
final ByteBuf buf = Unpooled.copiedBuffer(“data”,CharsetUtil.UTF_8).retain();
// 创建将数据写到 Channel 的 Runnable
Runnable writer = new Runnable() {
@Override
public void run() {
ChannelFuture cf = channel.writeAndFlush(buf.duplicate());
cf.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
// 写操作完成,并没有错误发生
if (future.isSuccess()){
System.out.println(“successful”);
}else{
// 记录错误
System.out.println(“error”);
future.cause().printStackTrace();
}
}
});
}
};

// 获取到线程池的 Executor 的引用
Executor executor = Executors.newCachedThreadPool();

// 提交到某个线程中执行
executor.execute(writer);

// 提交到另一个线程中执行
executor.execute(writer);
这里,我们需要注意的是:
创建 ByteBuf 的时候,我们使用了 retain 这个方法,他是将我们生成的这个 ByteBuf 进行保留操作。
在 ByteBuf 中有这样的一种区域:非保留和保留派生缓冲区。
这里有点复杂,我们可以简单的理解,如果调用了 retain 那么数据就存在派生缓冲区中,如果没有调用,则会在调用后,移除这一个字符数据。(如下是 ByteBuf 源码的解释)
/*<h4>Non-retained and retained derived buffers</h4>
*
* Note that the {@link #duplicate()}, {@link #slice()}, {@link #slice(int, int)} and {@link #readSlice(int)} does NOT
* call {@link #retain()} on the returned derived buffer, and thus its reference count will NOT be increased. If you
* need to create a derived buffer with increased reference count, consider using {@link #retainedDuplicate()},
* {@link #retainedSlice()}, {@link #retainedSlice(int, int)} and {@link #readRetainedSlice(int)} which may return
* a buffer implementation that produces less garbage.
*/
好的,我想你可以自己动手去测试一下,最好再看看源码,加深一下实现的原理印象。
这里的线程池并不是现实线程安全,而是用来做测试多线程的,Netty 的 Channel 实现是线程安全的,所以我们可以存储一个到 Channel 的引用,并且每当我们需要向远程节点写数据时,都可以使用它,即使当时许多线程都在使用它,消息也会被保证按顺序发送的。
结语
最后,介绍一下,个人的一个基于 Netty 的开源项目:InChat
一个轻量级、高效率的支持多端(应用与硬件 Iot)的异步网络应用通讯框架

参考资料:《Netty 实战》

正文完
 0