摘要：
Rocketmq使用netty实现网络连接。发现多个线程无法使用通道连接，所以这是线程安全的吗？使用Netty编程时，我们通常从用户线程而不是Netty线程池启动写操作，因为我们无法在Netty事件回调中执行大量耗时的操作。那么问题来了——1.writeAndFlush线程是否安全？
rocketmq用netty实现的网络连接，发现它多个线程掉用一个channel连接，所以这个是线程安全的？
使用Netty编程时，我们经常会从用户线程，而不是Netty线程池发起write操作，因为我们不能在netty的事件回调中做大量耗时操作。那么问题来了 –
1， writeAndFlush是线程安全的吗？
2， 是否使用了锁，导致并发性能下降呢
我们来看代码 – 在DefaultChannelHandlerContext中
@Override public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) { DefaultChannelHandlerContext next; next = findContextOutbound(MASK_WRITE); ReferenceCountUtil.touch(msg, next); next.invoker.invokeWrite(next, msg, promise); next = findContextOutbound(MASK_FLUSH); next.invoker.invokeFlush(next); return promise; }
在DefaultChannelHandlerInvoker.java中
@Override public void invokeWrite(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { if (msg == null) { throw new NullPointerException("msg"); } if (!validatePromise(ctx, promise, true)) { // promise cancelled ReferenceCountUtil.release(msg); return; } if (executor.inEventLoop()) { invokeWriteNow(ctx, msg, promise); } else { AbstractChannel channel = (AbstractChannel) ctx.channel(); int size = channel.estimatorHandle().size(msg); if (size > 0) { ChannelOutboundBuffer buffer = channel.unsafe().outboundBuffer(); // Check for null as it may be set to null if the channel is closed already if (buffer != null) { buffer.incrementPendingOutboundBytes(size); } } safeExecuteOutbound(WriteTask.newInstance(ctx, msg, size, promise), promise, msg); } }
private void safeExecuteOutbound(Runnable task, ChannelPromise promise, Object msg) { try { executor.execute(task); } catch (Throwable cause) { try { promise.setFailure(cause); } finally { ReferenceCountUtil.release(msg); } } }
可见，writeAndFlush如果在Netty线程池内执行，则是直接write；否则，将作为一个task插入到Netty线程池执行。
《Netty权威指南》写到
通过调用NioEventLoop的execute(Runnable task)方法实现，Netty有很多系统Task，创建他们的主要原因是：当I/O线程和用户线程同时操作网络资源时，为了防止并发操作导致的锁竞争，将用户线程的操作封装成Task放入消息队列中，由I/O线程负责执行，这样就实现了局部无锁化。