1.2

Netty在Java网络应用框架中的地位就好比#xff0c;Spring框架在JavaEE开发中的地位。

1.1

Netty是一个异步的基于事件驱动(即多路复用技术)的网络应用框架用于快速开发可维护、高性能的网络服务器和客户端。

1.2

Netty在Java网络应用框架中的地位就好比Spring框架在JavaEE开发中的地位。

Cassandra非关系型数据库Spark大数据分布式计算框架Hadoop大数据分布式存储框架RocketMQ阿里开源的消息队列ElasticSearch搜索引擎gRPCRPC框架DubboRPC框架Spring

5.xflux

api完全抛弃了tomcat使用netty作为服务器端Zookeeper分布式协调框架

1.3

传输问题如粘包、半包Linux多路复用的底层是epoll会存在空轮询导致

CPU

100%对应nio中Linux下不阻塞Netty兼容并解决该问题对

API

已废弃使用了AIO但是Linux的是伪AIO只有Win真正实现了AIO。

实际没有明显的性能提升却导致维护成本高

二、入门

dependencygroupIdio.netty/groupIdartifactIdnetty-all/artifactIdversion4.1.39.Final/version

/dependency使用Netty开发一个简单的服务器端和客户端

客户端向服务器端发送

ServerBootstrap()//启动器负责组装netty组件协调工作//一个selector和一个thread就叫EventLoop。

EventLoopGroup里面既有Boss处理连接也有Worker处理读写.group(new

NioEventLoopGroup())//选择服务器netty的ServerSocketChannel具体实现有4种自行看源码.channel(NioServerSocketChannel.class).childHandler(//child即Worker负责读写。

决定了Worker能执行哪些操作(Handler)new

ChannelInitializerNioSocketChannel()

{//代表和客户端进行数据读写的通道主要负责在initChannel里面添加其他的handlerOverrideprotected

void

{//添加具体的handlerchannel.pipeline().addLast(new

StringDecoder());//将netty的ByteBuf转换为字符串channel.pipeline().addLast(new

void

channelRead(ChannelHandlerContext

ctx,

{//此时这个msg就是上一步StringDecoder解码后的结果System.out.println(msg);}});}})//监听端口.bind(8080);}

}客户端

Bootstrap()//添加EventLoop.group(new

NioEventLoopGroup())//选择客户端channel实现.channel(NioSocketChannel.class)//添加处理器,连接建立后该初始化器被调用.handler(new

ChannelInitializerNioSocketChannel()

{Override//在连接建立后(accept)被调用protected

void

{channel.pipeline().addLast(new

StringEncoder());}})//连接到服务器.connect(localhost,

8080)//阻塞方法直到连接建立.sync()//代表着连接对象.channel()//向服务器发送数据.writeAndFlush(hello

world);}

对自己感兴趣的事件进行处理handler分为Inbound数据输入时走入站handler和Outbound数据输出时走出站handler两类

channel

可以被多个线程管理就会存在多个线程一起读写的情况防止出问题可能还要做串行操作工人既可以执行

操作也可以进行任务的处理。

每位工人有任务队列队列里可以存储该工人绑定的多个

channel

java.util.concurrent.ScheduledExecutorService

另一条线是继承自io.netty.util.concurrent.OrderedEventExecutor

boolean

一般我们不会直接使用EventLoop而是使用EventLoopGroup

3.2

io.netty.util.concurrent.EventExecutorGroup

Iterable

NioEventLoopGroup(2);//可以执行io事件、普通任务、定时任务//new

DefaultEventLoopGroup();//执行普通任务、定时任务//2.

获取下一个事件循环对象这个next底层就是轮询System.out.println(group.next());System.out.println(group.next());System.out.println(group.next());//3.

在netty中的意义就是执行一些比较耗时的任务group.next().execute(()

执行定时任务。

用于keepalive时连接的保活group.next().scheduleAtFixedRate(()

{log.info(定时任务);},

这时候是不应该让nioEventLoop阻塞在那里否则会影响后续其他channel的读写*

所以创建一个独立的EventLoopGroup用来执行那些耗时操作*/DefaultEventLoopGroup

new

AtomicInteger(0);Overridepublic

Thread

atomicInteger.incrementAndGet());}});new

划分为parent和child。

parent负责accept,

child负责read

AtomicInteger(0);Overridepublic

Thread

atomicInteger.incrementAndGet());}}),

new

AtomicInteger(0);Overridepublic

Thread

atomicInteger.incrementAndGet());}})).channel(NioServerSocketChannel.class).childHandler(new

ChannelInitializerNioSocketChannel()

{Overrideprotected

{ch.pipeline().addLast(handler1,

new

{Override//没有编解码那就是ByteBufpublic

void

channelRead(ChannelHandlerContext

ctx,

msg;log.info(buf.toString(Charset.defaultCharset())

ctx.channel().remoteAddress());ctx.fireChannelRead(msg);//将消息传递给下一个handler//上行做法通过直接调用父级该方法一样可以实现往下传递//super.channelRead(ctx,

msg);}}).addLast(defaultEventLoopGroup,

handler2,

channelRead(ChannelHandlerContext

ctx,

msg;TimeUnit.SECONDS.sleep(2L);log.info(buf.toString(Charset.defaultCharset())

ctx.channel().remoteAddress());}});}}).bind(8080);}

}客户端

Bootstrap()//创建启动器//添加EventLoop.group(new

NioEventLoopGroup())//选择客户端channel实现.channel(NioSocketChannel.class)//添加处理器,连接建立后该初始化器被调用.handler(new

ChannelInitializerNioSocketChannel()

{Override//在连接建立后(accept)被调用protected

void

{channel.pipeline().addLast(new

StringEncoder());}})//连接到服务器.connect(localhost,

8080);Channel

future//阻塞方法直到连接建立.sync().channel();//代表着连接对象while

(true)

{System.in.read();channel.writeAndFlush(hello

world);}}

}建立三个Client每个Client发送一次消息。

运行结果如图

由上图可知channel第一次创建时就与线程绑定了不管是处理读写的worker还是处理耗时的calc都是绑定的。

3.2.2

查看源码io.netty.channel.AbstractChannelHandlerContext中的invokeChannelRead

static

next.pipeline.touch(ObjectUtil.checkNotNull(msg,

msg),

返回下一个handler的eventLoop,这里是使用了多态的写法EventExecutor

executor

next.executor();//判断当前handler的线程是否和下一个handler的eventLoop是同一个线程//如果是直接调用否则,

else

{next.invokeChannelRead(m);}});}

}3.3

阻塞本线程阻塞直到channel成功建立连接addListener(回调对象)

添加回调其他线程执行channel监听到连接成功后执行回调对象

服务端代码保持不变客户端代码修改

{//带有Future、Promise的都是和异步方法配套使用目的是提高效率且正确处理结果ChannelFuture

future

Bootstrap()//创建启动器//添加EventLoop.group(new

NioEventLoopGroup())//选择客户端channel实现.channel(NioSocketChannel.class)//添加处理器,连接建立后该初始化器被调用.handler(new

ChannelInitializerNioSocketChannel()

{Override//在连接建立后(accept)被调用protected

void

{channel.pipeline().addLast(new

StringEncoder());}})//连接到服务器/***

异步非阻塞*

异步我(当前线程)只负责发起连接至于连接后的结果让别人(NioEventLoopGroup中线程)取*

非阻塞不等待直接下一步*/.connect(localhost,

8080);//方法一使用sync阻塞方法直到连接建立//future.sync();//直接获取channel其实获取到的是个尚未连接的channel//Channel

channel

future.channel();//代表着连接对象//for

(int

TimeUnit.SECONDS.sleep(1);//}//方法二使用addListener(回调对象)方法异步处理结果future.addListener(new

{Override//在nio线程连接建立好之后会调用operationCompletepublic

void

operationComplete(ChannelFuture

future)

{log.info(已连接);channel.writeAndFlush(hello

world);}

{log.info(未连接);}TimeUnit.SECONDS.sleep(1);}}});}

}3.3.2

NioEventLoopGroup();ChannelFuture

future

Bootstrap().group(group).channel(NioSocketChannel.class).handler(new

ChannelInitializerNioSocketChannel()

{Overrideprotected

LoggingHandler(LogLevel.DEBUG)).addLast(new

StringEncoder());}}).connect(localhost,

8080);Channel

通过添加LoggingHandler可以监视关闭的线程channel.close();log.info(调用close方法);break;}channel.writeAndFlush(line);}},

input).start();//方式一sync//ChannelFuture

closeFuture

channel.closeFuture();//closeFuture.sync();//log.info(成功关闭);//group.shutdownGracefully();//方式二监听channel.closeFuture().addListener(new

{Overridepublic

operationComplete(ChannelFuture

future)

{log.info(成功关闭后的回调);group.shutdownGracefully();}});}

}Netty可以通过LoggingHandler打印日志直观的查看Channel连接、收发、断开的过程

3.3.3

channel那样不是也可以吗非要用这么复杂的异步方式比如一个线程发起建立连接另一个线程去真正建立连接一个线程去关闭连接另一个线程真正去关闭连接。

这个问题也很简单就比如多路复用的做法。

我只有4个线程来发起长连接如果一线程一长连接那种撑死只能建立4个长连接。

但是使用多路复用技术就能处理更多的长连接了。

这也就是Netty异步的核心思想了。

3.4

只能同步等待任务结束或成功、或失败才能得到结果。

比如get方法就是只能同步等待获取结果。

netty

Future

可以同步等待任务结束得到结果也可以异步方式得到结果比如CloseFuture的addListener但都是要等任务结束netty

Promise

FuturePromisecancel取消任务--isCanceled任务是否取消--isDone任务是否完成不能区分成功失败--get获取任务结果阻塞等待--getNow-获取任务结果非阻塞还未产生结果时返回

null-await-等待任务结束如果任务失败不会抛异常而是通过

isSuccess

判断-sync-等待任务结束如果任务失败抛出异常-isSuccess-判断任务是否成功-cause-获取失败信息非阻塞如果没有失败返回null-addLinstener-添加回调异步接收结果-setSuccess--设置成功结果setFailure--设置失败结果

3.4.2

Executors.newFixedThreadPool(2);FutureInteger

future

{log.info(calc..);TimeUnit.SECONDS.sleep(2);return

ThreadLocalRandom.current().nextInt(1,

10);}});log.info(waiting..);Integer

integer

future.get();log.info(received..{},

integer);}

{log.info(calc..);TimeUnit.SECONDS.sleep(2);return

ThreadLocalRandom.current().nextInt(1,

10);}});//方式一同步方式//log.info(waiting..);//Integer

integer

future.get();//log.info(received..{},

integer);//方式二异步方式future.addListener(new

GenericFutureListenerFutureInteger()

{Overridepublic

operationComplete(FutureInteger

future)

future.getNow();log.info(received..{},

integer);}});}

NioEventLoopGroup(1).next();//与future不同的是可以主动创建promise对象。

而不用像future一样通过提交任务获取对象。

//结果容器PromiseInteger

promise

{TimeUnit.SECONDS.sleep(2L);//int

0;//线程执行完毕后向promise填充结果promise.setSuccess(ThreadLocalRandom.current().nextInt(1,

10));}

{e.printStackTrace();promise.setFailure(e);}}).start();//接收结果log.info(waiting..);Integer

integer

promise.get();log.info(received..{},

integer);}

是原材料经过很多工序的加工先经过一道道入站工序再经过一道道出站工序最终变成产品

Slf4j

NioEventLoopGroup()).channel(NioServerSocketChannel.class).childHandler(new

ChannelInitializerNioSocketChannel()

{Overrideprotected

{//通过channel拿到pipelineChannelPipeline

pipeline

ch.pipeline();//添加处理器。

netty会自动添加两个handler分别为head和tail//addLast并不是加到最后而是加到tail之前//channel的执行流程head-In_1-In_2-In_3-Out_4-Out_5-Out_6-tailpipeline.addLast(In_1,

new

channelRead(ChannelHandlerContext

ctx,

{log.info(In_1);super.channelRead(ctx,

msg);//唤醒下一个入站处理器}});pipeline.addLast(In_2,

new

channelRead(ChannelHandlerContext

ctx,

{log.info(In_2);super.channelRead(ctx,

msg);//唤醒下一个入站处理器}});pipeline.addLast(In_3,

new

channelRead(ChannelHandlerContext

ctx,

{log.info(In_3);super.channelRead(ctx,

msg);//唤醒下一个入站处理器。

此处已经结尾了所以无所谓。

//ctx.writeAndFlush(hello

world);//表示从当前处理器往前找

处理器ctx.channel().writeAndFlush(hello

new

ChannelOutboundHandlerAdapter()

{Overridepublic

{log.info(Out_4);super.write(ctx,

msg,

promise);}});pipeline.addLast(Out_5,

new

ChannelOutboundHandlerAdapter()

{Overridepublic

{log.info(Out_5);super.write(ctx,

msg,

promise);}});pipeline.addLast(Out_6,

new

ChannelOutboundHandlerAdapter()

{Overridepublic

{log.info(Out_6);super.write(ctx,

msg,

NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new

ChannelInitializerNioSocketChannel()

{Overrideprotected

channelRead(ChannelHandlerContext

ctx,

ChannelOutboundHandlerAdapter()

{Overridepublic

promise);}});}}).connect(localhost,

(true)

{System.in.read();channel.writeAndFlush(hello

world);}}

}可以看到ChannelInboundHandlerAdapter

addLast

channelRead(ChannelHandlerContext

ctx,

{log.info(h1);super.channelRead(ctx,

msg);}};ChannelInboundHandlerAdapter

new

channelRead(ChannelHandlerContext

ctx,

{log.info(h2);super.channelRead(ctx,

msg);}};ChannelOutboundHandlerAdapter

new

ChannelOutboundHandlerAdapter()

{Overridepublic

promise);}};ChannelOutboundHandlerAdapter

new

ChannelOutboundHandlerAdapter()

{Overridepublic

h4);//测试入站log.info(测试入站);embeddedChannel.writeInbound(inbound);Object

embeddedChannel.readInbound();log.info(测试出站);embeddedChannel.writeOutbound(outbound);Object

embeddedChannel.readOutbound();}

}3.6

io.netty.buffer.ByteBuf是对java.nio.ByteBuffer的增强。

支持动态扩容。

最大容量不超过Integer最大值。

池化思想。

对直接内存影响最大保证享受了直接内存的高读写的同时又能有效避免重复开辟内存造成的性能损失。

读写指针分离。

内部使用两套指针标识读和写。

与ByteBuffer相比就能减少不必要的来回切换。

零拷贝。

比如slice/duplicate/compositeByteBuf方便开发者高效编写。

比如链式调用。

3.6.2

该组成结构使得ByteBuf在使用上比ByteBuffer(如下图所示)方便许多因为节省了人为频繁切换指针位置的操作。

3.6.3

类似的内存分配算法提升分配效率高并发时池化功能更节约内存减少内存溢出的可能

-Dio.netty.allocator.type{unpooled|pooled}4.1

Android

ByteBufAllocator.DEFAULT.heapBuffer(10);也可以使用下面的代码来创建池化基于直接内存的

ByteBuf

ByteBufAllocator.DEFAULT.directBuffer(10);直接内存

堆内存

直接内存创建和销毁的代价昂贵但读写性能高少一次内存复制适合配合池化功能一起用直接内存对

压力小因为这部分内存不受

io.netty.buffer.ByteBufUtil.appendPrettyHexDump;

import

io.netty.util.internal.StringUtil.NEWLINE;public

class

index:).append(buffer.readerIndex()).append(

write

index:).append(buffer.writerIndex()).append(

capacity:).append(buffer.capacity()).append(NEWLINE);if

(pretty)

buffer);}System.out.println(buf.toString());}

}创建

ByteBuf能自动扩容初始值256最大值为Integer最大范围

public

默认创建大小256个字节的最大为Integer最大值ByteBuf

buffer

ByteBufAllocator.DEFAULT.buffer();System.out.println(buffer.getClass());DebugByteBuf.log(buffer);//初始256StringBuilder

new

{sb.append(a);}buffer.writeBytes(sb.toString().getBytes(StandardCharsets.UTF_8));DebugByteBuf.log(buffer);//由于超过了容量256自动扩容到512}

}写入

Endian(小端存储)是两种不同的字节存储方式用于表示一个多字节数据类型在内存中的存储顺序。

计算机中用来表示内存储器容量大小的基本单位是字节Byte此处是讲多字节数据类型的存储顺序。

Big

ByteBuffer.allocate(Long.BYTES).order(ByteOrder.BIG_ENDIAN).putLong(num).array();//

long

ByteBuffer.allocate(Long.BYTES).order(ByteOrder.LITTLE_ENDIAN).putLong(num).array();//

打印大端和小端字节数组System.out.println(Big-endian

dec:

Arrays.toString(bigEndian));System.out.println(Little-endian

dec:

Arrays.toString(littleEndian));String[]

bigEndianHex

Integer.toHexString(bigEndian[i]);}String[]

littleEndianHex

String[littleEndian.length];for

(int

Integer.toHexString(littleEndian[i]);}System.out.println(Big-endian

hex:

Arrays.toString(bigEndianHex));System.out.println(Little-endian

hex:

Arrays.toString(littleEndianHex));}public

static

ByteBufAllocator.DEFAULT.buffer(3);buffer.writeBytes(new

byte[]{1,2});DebugByteBuf.log(buffer,true);buffer.writeInt(250);DebugByteBuf.log(buffer,true);buffer.writeIntLE(250);DebugByteBuf.log(buffer,true);}

ByteBuf

若容量小于16则扩容后16若容量大于16小于64则扩容后64若容量大于64则每次扩容为当前容量的2倍扩容不能超过

max

readByte()读取一个字节会向后移动读指针ByteBuf

markReaderIndex()将当前位置定义为读标记。

默认是0ByteBuf

resetReaderIndex()重置到读标记getXXX读取不会改变读指针

3.6.4

使用的就是直接内存了需要特殊的方法来回收内存手动释放PooledByteBuf

和它的子类使用了池化机制需要更复杂的规则来回收内存手动释放还给内存池

ByteBuf

计数为0后的释放逻辑在io.netty.buffer.AbstractReferenceCountedByteBuf#deallocate方法中

谁来负责

在pipeline中head与tail两个处理器可以自动做收尾工作

入站msgtail

如果用户在某个handler中并没有将ByteBuf往后面的处理器传这时候收尾的head与tail就失去了作用因为你根本没把资源传递给我我咋释放啊

tail源码

tail只处理入站写出都是通过outHandler执行的所以跟tail也没啥关系。

因此代码中也只实现入站处理器。

final

{}查看io.netty.channel.DefaultChannelPipeline.TailContext#channelRead方法往下跟找到如下代码

protected

onUnhandledInboundMessage(Object

msg)

{ReferenceCountUtil.release(msg);}

//ReferenceCountUtil

AbstractChannelHandlerContextimplements

ChannelInboundHandler{}查看io.netty.channel.DefaultChannelPipeline.HeadContext#write方法往下跟找到如下代码

public

{assertEventLoop();ChannelOutboundBuffer

outboundBuffer

https://github.com/netty/netty/issues/2362safeSetFailure(promise,

newClosedChannelException(initialCloseCause));//

release

resource-leakReferenceCountUtil.release(msg);return;}int

size;try

filterOutboundMessage(msg);size

pipeline.estimatorHandle().size(msg);if

(size

t);ReferenceCountUtil.release(msg);return;}outboundBuffer.addMessage(msg,

size,

ByteBufAllocator.DEFAULT.buffer(5);buf1.writeBytes(new

byte[]{1,

buf1.copy();//验证深拷贝buf2.setByte(0,

10);log(buf1,

切片后的ByteBuf底层是SlicedByteBuf再写内容对原始数据有影响因此SlicedByteBuf禁止写入。

验证并没有发生数据复制

ByteBufAllocator.DEFAULT.buffer(10);buf.writeBytes(new

byte[]{1,

true);//使用如下代码验证并没有进行数据复制//注意toString中的hashCode并非表示地址值只是哈希值而已buf1.setByte(0,

10);log(buf1,

true);System.out.println(引用次数buf.refCnt());//4}

}浅拷贝duplicate

ByteBufAllocator.DEFAULT.buffer(5);buf1.writeBytes(new

byte[]{1,

ByteBufAllocator.DEFAULT.buffer(5);buf2.writeBytes(new

byte[]{6,

ByteBufAllocator.DEFAULT.buffer(10);//buf.writeBytes(buf1).writeBytes(buf2);//log(buf,

true);//不发生数据复制的合并CompositeByteBuf

buf

ByteBufAllocator.DEFAULT.compositeBuffer(10);buf.addComponents(true,

buf1,

buf2);buf1.retain();//添加引用防止buf1和buf2被释放buf2.retain();//添加引用防止buf1和buf2被释放log(buf,

true);//验证未发生数据合并buf1.setByte(0,

10);log(buf,

true);System.out.println(buf1.refCnt());System.out.println(buf2.refCnt());}

}实际编写代码时需要添加引用和手动释放

ByteBufAllocator.DEFAULT.buffer(5);

buf1.writeBytes(new

ByteBufAllocator.DEFAULT.buffer(5);

buf2.writeBytes(new

System.out.println(ByteBufUtil.prettyHexDump(buf3));输出

-------------------------------------------------|

-------------------------------------------------------------------------

|00000000|

-------------------------------------------------------------------------也可以用来包装普通字节数组底层也不会有拷贝操作

ByteBuf

System.out.println(buf4.getClass());

System.out.println(ByteBufUtil.prettyHexDump(buf4));输出

class

io.netty.buffer.CompositeByteBuf-------------------------------------------------|

-------------------------------------------------------------------------

|00000000|

-------------------------------------------------------------------------T.buffer(5);

buf1.writeBytes(new

ByteBufAllocator.DEFAULT.buffer(5);

buf2.writeBytes(new

System.out.println(ByteBufUtil.prettyHexDump(buf3));

-------------------------------------------------|

|±-------±------------------------------------------------±---------------

|00000000|

±-------±------------------------------------------------±---------------

ByteBuf

System.out.println(buf4.getClass());

System.out.println(ByteBufUtil.prettyHexDump(buf4));输出

class

io.netty.buffer.CompositeByteBuf-------------------------------------------------|

-------------------------------------------------------------------------

|00000000|

-------------------------------------------------------------------------