vb.nettyp类 vbnet class
netty3.6.2中写数据的过程,以及写数据写不出去后怎么处理
netty写数据的时候,会先放到一个缓存队列AbstractNioChannel.writeBufferQueue中,这个队列是WriteRequestQueue
让客户满意是我们工作的目标,不断超越客户的期望值来自于我们对这个行业的热爱。我们立志把好的技术通过有效、简单的方式提供给客户,将通过不懈努力成为客户在信息化领域值得信任、有价值的长期合作伙伴,公司提供的服务项目有:申请域名、网络空间、营销软件、网站建设、潞城网站维护、网站推广。
Java代码
public void eventSunk(
ChannelPipeline pipeline, ChannelEvent e) throws Exception {
if (e instanceof ChannelStateEvent) {
……
} else if (e instanceof MessageEvent) {
MessageEvent event = (MessageEvent) e;
NioSocketChannel channel = (NioSocketChannel) event.getChannel();
boolean offered = channel.writeBufferQueue.offer(event);//写到channel的writeBufferQueue
assert offered;
channel.worker.writeFromUserCode(channel);
}
}
WriteRequestQueue的offer方法中会根据缓存消息的总大小(字节数)判断是否超过了高水位线highWaterMark,如果第一次超过了超过高水位线,就会fireChannelInterestChanged;后边如果仍然一直往队列放数据,缓存的消息的大小持续超过高水位线的时候,不会再fireChannelInterestChanged。
Java代码
public boolean offer(MessageEvent e) {
boolean success = queue.offer(e);
assert success;
int messageSize = getMessageSize(e);
int newWriteBufferSize = writeBufferSize.addAndGet(messageSize);
int highWaterMark = getConfig().getWriteBufferHighWaterMark();
if (newWriteBufferSize = highWaterMark) {
if (newWriteBufferSize - messageSize highWaterMark) {
highWaterMarkCounter.incrementAndGet();
if (!notifying.get()) {
notifying.set(Boolean.TRUE);
fireChannelInterestChanged(AbstractNioChannel.this);
notifying.set(Boolean.FALSE);
}
}
}
return true;
}
fireChannelInterestChanged这个会调到SimpleChannelUpstreamHandler.handleUpstream,触发SimpleChannelUpstreamHandler.channelInterestChanged,可以通过继承这个方法来自定义做些事情。高水位的值可以通过Bootstrap设置,最终会调到DefaultNioSocketChannelConfig.setOption。writeBufferHighWaterMark默认值为64K
Java代码
public boolean setOption(String key, Object value) {
if (super.setOption(key, value)) {
return true;
}
if ("writeBufferHighWaterMark".equals(key)) {
setWriteBufferHighWaterMark0(ConversionUtil.toInt(value));
} else if ("writeBufferLowWaterMark".equals(key)) {
setWriteBufferLowWaterMark0(ConversionUtil.toInt(value));
} else if ("writeSpinCount".equals(key)) {
setWriteSpinCount(ConversionUtil.toInt(value));
} else if ("receiveBufferSizePredictorFactory".equals(key)) {
setReceiveBufferSizePredictorFactory((ReceiveBufferSizePredictorFactory) value);
} else if ("receiveBufferSizePredictor".equals(key)) {
setReceiveBufferSizePredictor((ReceiveBufferSizePredictor) value);
} else {
return false;
}
return true;
}
然后在write0的时候会从队列拉数据,拉数据的时候,如果发现本次拉的数据会导致缓存的数据大小(字节)从低水位writeBufferLowWaterMark之上,掉到了低水位之下,即跨过了低水位,会再次触发fireChannelInterestChanged事件。writeBufferLowWaterMark默认值为32K
Java代码
public MessageEvent poll() {
MessageEvent e = queue.poll();
if (e != null) {
int messageSize = getMessageSize(e);
int newWriteBufferSize = writeBufferSize.addAndGet(-messageSize);
int lowWaterMark = getConfig().getWriteBufferLowWaterMark();
if (newWriteBufferSize == 0 || newWriteBufferSize lowWaterMark) {
if (newWriteBufferSize + messageSize = lowWaterMark) {//本次拉取,是的缓存数据大小掉到了低水位之下
highWaterMarkCounter.decrementAndGet();
if (isConnected() !notifying.get()) {
notifying.set(Boolean.TRUE);
fireChannelInterestChanged(AbstractNioChannel.this);
notifying.set(Boolean.FALSE);
}
}
}
}
return e;
}
超过高水位和低于低水位都会触发fireChannelInterestChanged,怎么区分呢?通过AbstractChannel. isWritable(),如果channel的interestOps里边有注册过OP_WRITE,则是不可写的,否则是可写的
Java代码
public boolean isWritable() {
return (getInterestOps() OP_WRITE) == 0;
}
public int getInterestOps() {
if (!isOpen()) {
return Channel.OP_WRITE;
}
int interestOps = getRawInterestOps();
int writeBufferSize = this.writeBufferSize.get();
if (writeBufferSize != 0) {
if (highWaterMarkCounter.get() 0) {//还记得这个值,放数据到发送队列的时候值+=1,从队列拉数据出来的时候值-=1
int lowWaterMark = getConfig().getWriteBufferLowWaterMark();
if (writeBufferSize = lowWaterMark) {//缓存队列数据量,超过高水位,也超过了低水位,意味着高水位低水位,此时等于注册写操作
interestOps |= Channel.OP_WRITE;
} else {
interestOps = ~Channel.OP_WRITE;//缓存队列数据量,超过高水位但是低于低水位,意味着低水位高水位,此时等于没有注册写操作
}
} else {//超过高水位counter=0,意味着当前数据量小于高水位
int highWaterMark = getConfig().getWriteBufferHighWaterMark();
if (writeBufferSize = highWaterMark) {//这里,缓存数据量仍然高于高水位.....并发?按道理说channel的处理是单线程处理的,此时等于注册写操作
interestOps |= Channel.OP_WRITE;
} else {
interestOps = ~Channel.OP_WRITE;
}
}
} else {
interestOps = ~Channel.OP_WRITE;//写队列没数据,没有注册写操作
}
return interestOps;
}
即,如果超过高水位isWritable()==false,低于低水位isWritable()==true,低水位优先级高于高水位,即如果 当前水位低水位 则不可写,否则可写
如果在通过netty向某机器写数据,但是写很缓慢,则会导致数据都缓存到netty的发送队列中,如果不做控制,可能会导致full gc/cms gc频繁,甚至最终OOM。所以可以考虑用高水位和低水位的值来控制netty的缓存队列,即用AbstractChannel.isWritable来控制是否继续写,如果AbstractChannel.isWritable==false,则丢弃数据,或者记录发送数据的状态,待后续缓存数据队列水位下降到安全水位后再发送。
netty系列之:一口多用,使用同一端口运行不同协议
在之前的文章中,我们介绍了在同一个netty程序中支持多个不同的服务,它的逻辑很简单,就是在一个主程序中启动多个子程序,每个子程序通过一个BootStrap来绑定不同的端口,从而达到访问不同端口就访问了不同服务的目的。
但是多个端口虽然区分度够高,但是使用起来还是有诸多不便,那么有没有可能只用一个端口来统一不同的协议服务呢?
今天给大家介绍一下在netty中使用同一端口运行不同协议的方法,这种方法叫做port unification。
在讲解自定义port unification之前,我们来看下netty自带的port unification,比如SocksPortUnificationServerHandler。
我们知道SOCKS的主要协议有3中,分别是SOCKS4、SOCKS4a和SOCKS5,他们属于同一种协议的不同版本,所以肯定不能使用不同的端口,需要在同一个端口中进行版本的判断。
具体而言,SocksPortUnificationServerHandler继承自ByteToMessageDecoder,表示是将ByteBuf转换成为对应的Socks对象。
那他是怎么区分不同版本的呢?
在decode方法中,传入了要解码的ByteBuf in,首先获得它的readerIndex:
我们知道SOCKS协议的第一个字节表示的是版本,所以从in ByteBuf中读取第一个字节作为版本号:
有了版本号就可以通过不同的版本号进行处理,具体而言,对于SOCKS4a,需要添加Socks4ServerEncoder和Socks4ServerDecoder:
对于SOCKS5来说,需要添加Socks5ServerEncoder和Socks5InitialRequestDecoder两个编码和解码器:
这样,一个port unification就完成了,其思路就是通过传入的同一个端口的ByteBuf的首字节,来判断对应的SOCKS的版本号,从而针对不同的SOCKS版本进行处理。
在本例中,我们将会创建一个自定义的Port Unification,用来同时接收HTTP请求和gzip请求。
在这之前,我们先看一下两个协议的magic word,也就是说我们拿到一个ByteBuf,怎么能够知道这个是一个HTTP协议,还是传输的一个gzip文件呢?
先看下HTTP协议,这里我们默认是HTTP1.1,对于HTTP1.1的请求协议,下面是一个例子:
HTTP请求的第一个单词就是HTTP请求的方法名,具体而言有八种方法,分别是:
OPTIONS
返回服务器针对特定资源所支持的HTTP请求方法。也可以利用向Web服务器发送'*'的请求来测试服务器的功能性。
HEAD
向服务器索要与GET请求相一致的响应,只不过响应体将不会被返回。这一方法可以在不必传输整个响应内容的情况下,就可以获取包含在响应消息头中的元信息。
GET
向特定的资源发出请求。注意:GET方法不应当被用于产生“副作用”的操作中,例如在Web Application中。其中一个原因是GET可能会被网络蜘蛛等随意访问。
POST
向指定资源提交数据进行处理请求(例如提交表单或者上传文件)。数据被包含在请求体中。POST请求可能会导致新的资源的建立和/或已有资源的修改。
PUT
向指定资源位置上传其最新内容。
DELETE
请求服务器删除Request-URI所标识的资源。
TRACE
回显服务器收到的请求,主要用于测试或诊断。
CONNECT
HTTP/1.1协议中预留给能够将连接改为管道方式的代理服务器。
那么需要几个字节来区分这八个方法呢?可以看到一个字节是不够的,因为我们有POST和PUT,他们的第一个字节都是P。所以应该使用2个字节来作为magic word。
对于gzip协议来说,它也有特殊的格式,其中gzip的前10个字节是header,其中第一个字节是0x1f,第二个字节是0x8b。
这样我们用两个字节也能区分gzip协议。
这样,我们的handler逻辑就出来了。首先从byteBuf中取出前两个字节,然后对其进行判断,区分出是HTTP请求还是gzip请求:
对应的,我们还需要对其添加相应的编码和解码器,对于gzip来说,netty提供了ZlibCodecFactory:
对于HTTP来说,netty也提供了HttpRequestDecoder和HttpResponseEncoder还有HttpContentCompressor来对HTTP消息进行编码解码和压缩。
添加了编码和解码器之后,如果你想自定义一些操作,只需要再添加自定义的对应的消息handler即可,非常的方便。
本文的例子可以参考: learn-netty4
netty基本使用- socket通信
[netty 基本使用- 作为http服务器][gcssloop]
[gcssloop]:
ServerSocket.java
** ServerInitializer.java **
** ServerHandler.java 处理业务 **
** ClientSocket.java **
** Clientinitializer.java **
**ClientHandler.java 处理业务 **
** 可以多次和服务器端通信的写法 **
netty 常用的处理大数据分包传输问题的解决类。
编码类,自动将
+----------------+
| "HELLO, WORLD" |
+----------------+
格式的数据转换成
+--------+----------------+
+--------+----------------+
格式的数据
[netty 数据分包、组包、粘包处理机制][123]
[123]:
Netty实战6——Netty业务处理线程池的选择
根据我们前面分析的,接收到消息后,为了避免在I/O线程里执行耗时的操作,一般都会使用线程池来执行业务处理逻辑.
那是使用Netty提供给我们的方法,传入一个线程池还是使用我们自己定义的线程池好呢?
先来看Netty给我们提供的
即我们添加handler的时候可以传入一个线程池进去
DefaultEventExecutorGroup
它与NioEventLoop之间的区别又是什么?
其次
也就是说使用netty提供默认的,是绑定的.如下图
如果采用自定义线程池时,优化方向就是锁消除.
可以使用Disruptor或者使用ChannelId与业务线程池中的某个业务进行绑定
链接:
技巧分享-20周-Netty的closeFuture().sync()和close().sync()
Netty模块存在closeFuture().sync()和close().sync()关闭端口,但是使用起来是两种情况
例如f.channel().closeFuture().sync() 是等待服务端监听端口关闭
该方法进行阻塞,等待服务端链路关闭之后继续执行。
这种模式一般都是使用Netty模块主动向服务端发送请求,然后最后结束才使用
例如f.channel().close().sync() 则作为服务端启用Netty模块接收情况使用。
一般在Netty模块对应@PreDestroy方法里面使用该方法来结束服务
如果是用Netty模块作为服务端,在@PreDestroy方法中使用
f.channel().closeFuture().sync() 来停止服务时候,在weblogic部署时会出现停止不了应用情况。
解决这种情况就是将 f.channel().closeFuture().sync() 改 f.channel().close().sync() 即可
如何保证netty执行事件是顺序而且高效
netty实现多个handler顺序调用在netty中,一次数据交互,可以由多个handler去处理,例如handler1和handler2,那么,在前面那个handler的messageReceived的最后要加上ctx.sendUpstream(e);理论请见:AChannelEventcanbehandledbyeitheraChannelUpstreamHandleroraChannelDownstreamHandlerandbeforwardedtotheclosesthandlerbycallingChannelHandlerContext.sendUpstream(ChannelEvent)orChannelHandlerContext.sendDownstream(ChannelEvent).代码:复制代码publicclassHandler1extendsSimpleChannelUpstreamHandler{@OverridepublicvoidmessageReceived(ChannelHandlerContextctx,MessageEvente){System.out.println("1messagereceived");Stringa="11";Objecto=a;ctx.getChannel().write(a);ctx.sendUpstream(e);}@OverridepublicvoidexceptionCaught(ChannelHandlerContextctx,ExceptionEvente)throwsException{e.getChannel().close();}}复制代码复制代码publicclassHandler2extendsSimpleChannelUpstreamHandler{@OverridepublicvoidmessageReceived(ChannelHandlerContextctx,MessageEvente){System.out.println("2messagereceived");e.getChannel().close();}@OverridepublicvoidexceptionCaught(ChannelHandlerContextctx,ExceptionEvente)throwsException{e.getChannel().close();}}复制代码复制代码publicclassTcpServer{publicstaticvoidmain(String[]args){System.out.println("startingatcpserver");ServerBootstrapsb=newServerBootstrap(newNioServerSocketChannelFactory(Executors.newCachedThreadPool(),Executors.newCachedThreadPool()));sb.setPipelineFactory(newPKServerPipelineFactory());sb.setOption("child.tcpNoDelay",true);sb.setOption("child.keepAlive",true);sb.bind(newInetSocketAddress(9999));}}
本文标题:vb.nettyp类 vbnet class
文章来源:http://cdiso.cn/article/hgehid.html