Netty中的Channel之数据冲刷与线程安全（writeAndFlush）

本文首发于本博客，如需转载，请申明出处.

GitHub项目地址

InChat

一个轻量级、高效率的支持多端（应用与硬件Iot）的异步网络应用通信框架

前言

本文预设读者已经了解了必定的Netty基础知识，并可以本身构建一个Netty的通讯服务（包括客户端与服务端）。那么你必定使用到了Channel，这是Netty对传统JavaIO、NIO的连接封装实例。

那么接下来让咱们来了解一下关于Channel的数据冲刷与线程安全吧。

数据冲刷的步骤

一、获取一个连接实例

@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 { //...... } 

二、建立一个持有数据的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和字符串的长度。

三、冲刷数据

@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。（以下为源码的解释）

/** * The result of an asynchronous {@link Channel} I/O operation. */ public interface ChannelFuture extends Future<Void> { //...... } 

既然如此，那么咱们能够明显的知道咱们能够对其添加对应的监听。

四、异步回调结果监听

@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）的异步网络应用通信框架