Netty实战笔记

Netty

技术要点

同步/异步

阻塞/非阻塞

Reactor

封装成帧

固定长度

FixedLengthFrameDecoder

分隔符

DelimiterBasedFrameDecoder

设置固定长度字段表示消息体长度信息

LengthFieldBasedFrameDecoder

三种网络I/O模式

BIO阻塞

排队打翻

阻塞 同步

NIO

点单.等待被叫模式

非阻塞 非同步

AIO

包厢模式

异步

编码/解码

一次解码器

解码器

将网络上的数据解析为字节数组

ByteToMessageDecoder

io.netty.ByteBuf(原始数据流)-->io.netty.buffer.ByteBuf(用户数据)

编码器

二次解码器

j将bytebuf转成自定义的对象

MessageToMessageDecoder

io.netty.buffer.ByteByf-->user java object

编码方式

java序列化

xml

ProtoBuf

receiveMesage

ch.pipeline().addLast(new ProtobufVarint32FrameDecoder()); ch.pipeline().addLast(new ProtobufDecoder({自己实现的解析类}));

sendMessage

ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender()); ch.pipeline().addLast(new ProtobufEncoder());

json

messsagePack

keepalive

Idle监控

同步分析三要素

原子性

可见性

有序性

源码解析

启动过程

selector是在new NioEventLoopGroup()的时候创建

最终监听OP_ACCEPT是通过bind完成后的fireChannelActive()来触发的

NioEventLoop是通过Register操作的执行来完成启动的

连接过程

bossThread

NioEventLoop中的selector监听OP_ACCEPT

创建socketChannel

初始化socketChannel

从workerGroup中选择一个NioEventLoop

workerThread

将socketChannel注册到选择的NioEventLoop的selector

注册事件OP_READ到selector

关闭链接

本质

java.nio.channels.spi.AbstractInterruptibleChannel.close()

java.nio.channels.SelectionKey.cancle()

要点

Channel的关闭包含SectionKey的cancle

关闭服务

<netty实战>

第二章'

核心组件

Channel

硬件设备

文件

socket

其他可执行的组件

回调

callable

netty内部通过回调机制来处理事件

处理事件的是handler

Future

可视为是一个异步操作结果的占位符

可以注册ChannelFutureLinster

监听回调方法operationComplete(ChannelFuture future)

Channel channel = ...;// Does not blockChannelFuture future = channel.connect(new InetSocketAddress(\"192.168.0.1\

每个netty的出站IO都会返回一个ChannelFuture

意味着不会阻塞

event

日志记录

数据转换

流控制

引用程序逻辑

入站事件

连接状态

激活

失活

数据读取

用户事件

错误事件

出站事件

打开 /  关闭到远程节点的链接

将数据写或者冲刷到套接字

ChannelHandler

每个事件都可以分给channelHandler中的某个用户实现的方法

组合

每个channel都会分配一个eventloop

注册感兴趣的事件

将事件派发给channelhandler

安排进一步动作

顺序性

单线程

第三章-初探

对应关系

channel

EventLoop

ChannelFuture

异步通知

Channel接口

基本操作

bind()

connect()

read()

write()

子类

EmbeddedChannel；LocalServerChannel；NioDatagramChannel；NioSctpChannel；NioSocketChannel

eventloop接口

处理连接生命周期中所发生的事件

事件!

层次关系

eventloopgroup            包含N个      EventLoop

每个EventLoop                     生命周期内                      只绑定一个Thread

所有Eventloop处理的IO                                都在他自己专有的Thread上处理

一个Channel在他的生命中期内                    只注册到一个Eventloop

一个Eventloop                                                   分配给                      N个Channel

ChannelFuture接口

addLinstener()

获取结果通知

同一个Channel的操作保证他们以顺序执行

处理网络上事件

业务处理

logging

ChannelInboundHandler

处理入站事件

ChannelOutboundHandler

处理出站事件

常用适配器类

 ChannelHandlerAdapter ChannelInboundHandlerAdapter ChannelOutboundHandlerAdapter ChannelDuplexHandler

入站消息

出站消息

SimpleChannelInboundHanler<T>

ChannelPipeLine

是ChannelHandler的容器

定义用于该链上传播入站/出站事件流的API

Channel被创建时                 会自动分配到所属的              ChannelPioleLine

责任链模式.按照顺序传递事件

引导-Bootstrap

客户端引导

连接到远程主机和端口

eventloopgroup  1个

服务器引导

绑定到一个本地端口

eventloopgroup 1~2 个

有2组channel

第一组表示绑定到本地的一个端口上正在监听的socket--             serverChannel

第二组表示体精闯将的用来处理传入客户端的链接channel             每个客户端都有一个

第四章-传输

传输类型

OIO

阻塞

Epoll

Local

JVM内部的异步

Embedded

测试ChannelHandler

将数据从一种格式转换为另一种格式；提供异常的通知；提供 Channel 变为活动的或者非活动的通知；提供当 Channel 注册到 EventLoop 或者从 EventLoop 注销时的通知；提供有关用户自定义事件的通知

持有一个ChannleConfig

持有一个ChannlePipeLine

主要方法

eventLoop

pipeline()

isActive()

localAddress

remoteAddress

write

flush

将之前写入的数据冲刷到底层传输

writeAndFlush

selector

OP_ACCEPT

请求在新连接创建Channel时获得通知

OP_CONNECT

请求创建一个新连接时获得通知

OP_READ

OP_WRITE

请求可以向channel中写更多的数据时获得通知

第五章 ByteBuf

初识

ByteBuf 是 netty 的数据容器

组件

abstract class ByteBuf

interface ByteBufHolder

用于存储各种属性值

字节

HTTP状态码

cookie

支持池化

优点

它可以被用户自定义的缓冲区类型扩展；通过内置的复合缓冲区类型实现了透明的零拷贝；容量可以按需增长（类似于 JDK 的 StringBuilder）；在读和写这两种模式之间切换不需要调用 ByteBuffer 的 flip()方法；读和写使用了不同的索引；支持方法的链式调用；支持引用计数；支持池化

ByteBuf

两个指针

读指针

写指针

一些方法

get/set

不移动指针

read/write

移动指针

三个区域

可写字节

可读字节

读过的部分

ByteBuf使用模式

堆缓冲区

HeapBuf

存储在JVM管理的内存中

支撑数组    backing array

直接缓冲区

DirectBuf

存储在JVM管理之外

复合缓冲区

CompositeByteBuf

为多个ByteBuf提供一个聚合的视图

字节级操作

随机访问

getByte(index)

第六章 ChannelHandler 和ChannelPipeline

ChannelHandlers

channel生命周期

channelunregistered

channelRegistered

已经注册到EventLoop

channelActive

channel处于活动状态

channelInactive

channel没有连接到远程节点

@Shareable

可以将handler加入到不同的pipeline中

ChannelHandler生命周期

handlerAdded

把channelhandler添加到channelPipeline中时调用

handlerRemoved

从channelpipeline中移除时调用

exceptionCaugth

发生异常时调用

ChannelInboundHandler接口

入站

channelRegistered                        当 Channel 已经注册到它的 EventLoop 并且能够处理 I/O 时被调用channelUnregistered                    当 Channel 从它的 EventLoop 注销并且无法处理任何 I/O 时被调用channelActive                               当 Channel 处于活动状态时被调用； Channel 已经连接/绑定并且已经就绪channelInactive                             当 Channel 离开活动状态并且不再连接它的远程节点时被调用channelReadComplete                 当Channel上的一个读操作完成时被调用 ①channelRead                                 当从 Channel 读取数据时被调用ChannelWritabilityChanged          当 Channel 的可写状态发生改变时被调用。用户可以确保写操作不会完成 得太快（以避免发生 OutOfMemoryError）                                                                   或者可以在 Channel 变为再次可写时恢复写入。可以通过调用 Channel 的 isWritable()方法来检测Channel 的                                                                     可写性。 与可写性相关的阈值可以通过 Channel.config().setWriteHighWaterMark()和                                                                  Channel.config().setWriteLowWaterMark()方法来设置userEventTriggered                      当 ChannelnboundHandler.fireUserEventTriggered()方 法被调用时被调用.因为一个POJO被传经了ChannelPipeline

ReferenceCountUtil.release(msg);

出站

bind\tconnect\tdisconnect\tclose\tderegister\tflush\twrite\twriteAndFlush\tread   

以上方法皆调用ChannelPipeline 中下一个ChannelOutboundHandler的对应方法

ChannelHandlerContext

ChannelHandlerContext 有很多的方法，其中一些方法也存在于 Channel 和 ChannelPipeline 本身上，但是有一点重要的不同。如果调用 Channel 或者 ChannelPipeline 上的这些方法，它们将沿着整个 ChannelPipeline 进行传播。而调用位于 ChannelHandlerContext上的相同方法，则将从当前所关联的 ChannelHandler 开始，并且只会传播给位于该ChannelPipeline 中的下一个能够处理该事件的 ChannelHandler。

使用

异常处理

实现方法  exceptionCaught

通过 ChannelFuture

每个出站的事件都会传入一个ChannelPromise

方法一    加入监听器

new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture f) {    if (!f.isSuccess()) {        f.cause().printStackTrace();        f.channel().close();    }}

方法二    setFailure(Throwable cause)

第七章 eventloop               和          线程模型

线程模型要素

操作系统

编程语言

框架 / 应用程序的上下文 的线程管理的关键方面

事件循环---处理连接的生命周期内的发生的事件

while (!terminated) {    List<Runnable> readyEvents = blockUntilEventsReady();    for (Runnable ev: readyEvents) {        ev.run();    }}

仅有一个新定义的方法              EventLoopGroup parent();

事件/任务的执行顺序

本质是队列 所以是 FIFO

 所有的I/O操作和事件都由已经被分配给了EventLoop的那个Thread来处理 

任务调度

JDK 1.5 - 

Timer

JDK 1.5+ 

ScheduledExecutorService

executor.shutdown() 释放资源

 netty

扩展JDK

实现细节

线程管理

任务

在eventloop中?

是

执行

否

入队

eventloop/线程的分配

异步传输

具有三个eventloop的eventloopgroup

eventloop

channe

.....

阻塞传输

eventloopgroup

第八章引导

Bootstrap

在调用bind()

创建连接

调用 connect()

确定传输方式

生成serverbootstrap

new serverbootstrap

channel类型

指定要应用到新创建的 [Server]Channel 的 ChannelConfig 的 ChannelOption

option

serverChannel

childOption

设置[server]channel属性

attr

childAttr

处理流

childHandler

绑定端口

bind

服务器作为客户端访问其他服务器

channelActive() 中创建新的Bootstrap.  (ChannelHandler中)

连接远程端口 connect()

将当前服务器的eventloop设置到新建的Bootstrap中

bootstrap.group(ctx.eventloop())

交给一套eventloop管理

缓存future

在引导中加入多个ChannelHandler

实现ChannelInitializer

init()

执行init之后会将当前handler在pipeline中移除

netty的channelOption属性

final AttributeKey<Integer> id = new AttributeKey<Integer>(\"ID\"); 

创建

存储

Integer idValue = ctx.channel().attr(id).get();

获取

引导DatagramChannel

不调用connect()             改为调用        bind()

关闭

shutdownGracefully()

关闭 EventLoopGroup， 它将处理任何挂起的事件和任务，并且随后释放所有活动的线程。这就是调用 EventLoopGroup.shutdownGracefully()方法的作用

第九章.单元测试

EmbeddedChannel

第十章编解码器框架

定义

将消息对象转成适合网络传输的数据流

outbound

将网络数据流转成程序的可以使用的对象

inbound

解码器  实现了InboundHandler

字节 -->  消息

将读取的对象放入out.

然后会自动交给下一个InboundHandler处理

ReplayingDecoder

消息  -->  消息

TooLongFrameException

为了避免消息过大导致的OOM

编码器  实现了OutboundHandler

消息  --> 字节

MessageToByteEncoder

消息 --> 消息

MessageToMessageEncoder

抽象的编解码器类

同时实现了解码器和编码器

ByteToMessageCodec

MessageToMessageCodec

CombinedChannelDuplexHandler

ChannelInboundHandler 和 ChannelOutboundHandler的容器

第十一章预置的ChannelHandler      和      编解码器

通过 SSL/TLS 保护 Netty 应用程序

SslHandler

设置和获取超时时间， 超时之后， 握手ChannelFuture 将会被通知失败

设置和获取超时时间， 超时之后，将会触发一个关闭通知并关闭连接。这也将会导致通知该 ChannelFuture 失败

handshakeFuture()

返回一个在握手完成后将会得到通知的ChannelFuture。如果握手先前已经执行过了，则返回一个包含了先前的握手结果的 ChannelFuture

发送 close_notify 以请求关闭并销毁底层的 SslEngine

构建基于 Netty 的 HTTP/HTTPS 应用程序

组成部分

request

HttpRequest

HttpContent

LastHttpContent

response

HttpResponse

FullHttpRequest

编解码器

HttpRequestEncoder

HTTP 压缩

客户端使用  HttpContentDecompressor

处理服务器发来的压缩消息

服务器使用  HttpContentCompressor

需要客户端支持

使用HTTPS

需要将SslHandler加入pipeline

websocket

在客户端和服务器之间提供双向数据交换

websocketFrame类型

BinaryWebSocketFrame

数据帧:二进制数据

TextWebSocketFrame

数据帧:文本

ContinuationWebSocketFrame

数据帧:属于上一个BinaryWebSocket或者TextWebSocketFrame的文本或者二进制数据

接着上一个

CloseWebSocketFrame

PingWebSocketFrame

控制帧:请求一个PongWebSocketFrame

PongWebSocketFrame

控制帧:对PingWebSocketFrame请求的响应

服务器使用

前三个要按照顺序

HttpServerCodec()

HttpObjectAggregator(65536)

为握手提供聚合的httprequest

WebSocketServerProtocolHandler(\"/websocket\")

如果请求的端点是/websocket.则处理该升级握手

TextFrameHandler()

自己实现

BinaryFrameHandler()

ContinuationFrameHandler()

空闲 & 超时

IdleStateHandler

ReadTimeoutHandler

WriteTimeoutHandler

基于分隔符和长度的协议解码

基于分隔符

LineBasedFrameDecoder(final int maxLength)

单行最大长度

基于长度

FixedLengthFrameDecoder(int frameLength)

根据编码进帧头部中的长度值提取帧；该字段的偏移量以及长度在构造函数中指定

长度  +  内容

写大型数据

FileRegion

暂不论

序列化数据

JDK序列化

JBoss Marshalling

Protocol Buffers

ProtobufDecoder

使用 protobuf 对消息进行解码

ProtobufEecoder

使用 protobuf 对消息进行编码

ProtobufVarint32FrameDecoder

根据消息中的 Google Protocol Buffers 的“Base 128 Varints”整型长度字段值动态地分割所接收到的 ByteBuf

ProtobufVarint32LengthFieldPrepender

向 ByteBuf 前追加一个 Google Protocal Buffers 的“Base128 Varints”整型的长度字段值

private final MessageLite lite

Handlers

new ProtobufVarint32FrameDecoder()

用来做帧分割

new ProtobufEncoder()

new ProtobufDecoder(lite)

new ObjectHandler()

自己实现的消息解析器