欢迎访问 生活随笔!

生活随笔

当前位置: 首页 > 编程资源 > 编程问答 >内容正文

编程问答

java的网络工具netty简介

发布时间:2024/4/13 编程问答 57 豆豆
生活随笔 收集整理的这篇文章主要介绍了 java的网络工具netty简介 小编觉得挺不错的,现在分享给大家,帮大家做个参考.

2019独角兽企业重金招聘Python工程师标准>>>

java的网络工具netty简介

         Netty是一个NIO的客服端服务器框架,它可以简单、快速的搭建器一个协议包客服端服务器的应用程序。它极大的简化了TCP和UDP这类的网络编程。

   “快速”和“简单”并不意味着会让你的最终应用产生维护性或性能上的问题。Netty 是一个吸收了多种协议的实现经验,这些协议包括FTP,SMTP,HTTP,各种二进制,文本协议,并经过相当精心设计的项目,最终,Netty 成功的找到了一种方式,在保证易于开发的同时还保证了其应用的性能,稳定性和伸缩性。

        这里简单记录下学习要点,详细的讲解。可以看官网(github:https://github.com/netty/netty )或者查看李林锋的的系列文章http://ifeve.com/author/linfeng/ 。

        体系结构图:

        由李林锋讲解的易懂的架构图:

    1、两个selector线程:mainReactor处理accpet事件、subReactor处理connection、read、send事件

    2、业务处理线程池:包括编码、解码、业务处理。


1、官网案例

    a、处理bytes的serverhandler

/*** @see 进入的channel:用于处理接受时候的事件处理*/ public class TimeServerHandler extends ChannelInboundHandlerAdapter {/*** @see 当一个channel准备好的时候,发送一个32位的数字*/public void channelActive(final ChannelHandlerContext ctx) {// ByteBuf:没有了flip()。它只有2个功能:读、写// 读:// 写:当你写的时候,如果读取下标没有改变,则继续增长final ByteBuf time = ctx.alloc().buffer(4);time.writeInt((int) (System.currentTimeMillis() / 1000L + 2208988800L));final ChannelFuture f = ctx.writeAndFlush(time);// 当写如完成的时候,执行f.addListener(new ChannelFutureListener() {public void operationComplete(ChannelFuture future) throws Exception {// TODO Auto-generated method stubassert f == future;ctx.close();}});}public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close();} }

    sever的启动部分

public class TimeServer {private int port;public TimeServer() {this.port = port;}public void runn() throws Exception {EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap b = new ServerBootstrap();b.group(bossGroup, workerGroup);b.channel(NioServerSocketChannel.class);b.childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {// TODO Auto-generated method stubch.pipeline().addLast(new TimeServerHandler());}});b.option(ChannelOption.SO_BACKLOG, 128);b.childOption(ChannelOption.SO_KEEPALIVE, true);ChannelFuture f = b.bind(port).sync();f.channel().closeFuture().sync();} finally {workerGroup.shutdownGracefully();bossGroup.shutdownGracefully();}}public static void main(String[] args) {try {new TimeServer().runn();} catch (Exception e) {e.printStackTrace();}} }


    b、client部分:处理字节

public class TimeDecoder extends ByteToMessageDecoder {/*** @see 定义一个回调的数据累加buff* @see 如果有out,则表示解析成功。*/@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {if (in.readableBytes() < 4) {return;}out.add(in.readBytes(4));}}

   client的 channel处理类

public class TimeClientHandler extends ChannelInboundHandlerAdapter {public void channelRead(ChannelHandlerContext ctx, Object msg) {ByteBuf buf = (ByteBuf) msg;try {long currentTimeMillis = (buf.readUnsignedInt() - 2208988800L) * 1000L;System.out.println(new Date(currentTimeMillis));ctx.close();} catch (Exception e) {e.printStackTrace();} finally {buf.release();}}public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close();} }

    client的启动类:

public class TimeClient {public static void main(String[] args) throws InterruptedException {String host = args[0];int port = Integer.parseInt(args[1]);EventLoopGroup workerGroup = new NioEventLoopGroup();try {Bootstrap b = new Bootstrap();// 启动客服端连接b.group(workerGroup);// 同时用于主线程和工作线程b.channel(NioSocketChannel.class);// 客服端需要的channelb.option(ChannelOption.SO_KEEPALIVE, true); // socketChannel没有父类b.handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {// TODO Auto-generated method stubch.pipeline().addLast(new TimeClientHandler());}});ChannelFuture f = b.connect(host, port).sync();f.channel().closeFuture().sync();} finally {workerGroup.shutdownGracefully();}} }


2、stream处理

    小型buffer的socket传送流传输依据TCP/IP,接受的数据是储存在一个接受的socket的buffer中。但是,传送的buffer不是一个队列包、而是一个队列btyes。这就意味着,即使你使用两个包去传送两端信息,系统不会将它们视为两端信息,而是作为一串bytes。因此,这不能保证你读去的数据是你远程写入的数据。例如,我们需要使用系统的TCP/IP栈接受到3个数据包。

     

因为根据流协议,你很有可能在你的应用中读取到你下面的部分

因次,在服务器和客服端的接受部分,对接受数据必须定义一个协议的框架(处理方式),这个框架能够被应用程序使用。接收到的部分必须是下面这种方式。

    a、第一种解决方式:

        在TIME client的实例中。我们同样是有一个相似的问题,一个非常小的32位bit的整数数据,它不太可能分散。然而,随着流量的增加,问题是它会碎片化。

        简单的解决方式,增加一个内部的累加buffer,然后将接受的4bytes传输到这个buffer中。在TimeClientHandler直接修改

public class TimeClientHandler2 extends ChannelInboundHandlerAdapter {private ByteBuf buf;@Overridepublic void handlerAdded(ChannelHandlerContext ctx) {buf = ctx.alloc().buffer(4);}@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) {buf.release();buf = null;}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {ByteBuf m = (ByteBuf) msg;buf.writeBytes(m);m.release();if (buf.readableBytes() >= 4) {long currentTimeMillis = (buf.readUnsignedInt() - 2208988800L) * 1000L;System.out.println(new Date(currentTimeMillis));ctx.close();}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close();} }


    b、第二种就是前面的实例方式、将decode分离出来处理。看起来清晰、方便

3、编解object数据

    object

public class UnixTime {private final long value;public UnixTime() {this(System.currentTimeMillis() / 1000L + 2208988800L);}public UnixTime(long value) {this.value = value;}public long value() {return this.value;}public String toString() {return new Date((value() - 2208988800L) * 1000L).toString();}}

    object:decode

public class TimeDecoder2 extends ByteToMessageDecoder {/*** @see 定义一个回调的数据累加buff* @see 如果有out,则表示解析成功。*/@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {if (in.readableBytes() < 4) {return;}out.add(new UnixTime(in.readUnsignedInt()));} }

    objec:encode

public class TimeEncoder extends ChannelOutboundHandlerAdapter {public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {UnixTime m = (UnixTime) msg;ByteBuf encoded = ctx.alloc().buffer(4);encoded.writeInt((int) m.value());ctx.write(encoded, promise);}

   

    server:handler

public class TimeServerHandler2 extends ChannelInboundHandlerAdapter {/*** @see 当一个channel准备好的时候,发送一个32位的数字*/public void channelActive(final ChannelHandlerContext ctx) {// ByteBuf:没有了flip()。它只有2个功能:读、写// 读:// 写:当你写的时候,如果读取下标没有改变,则继续增长final ByteBuf time = ctx.alloc().buffer(4);time.writeInt((int) (System.currentTimeMillis() / 1000L + 2208988800L));final ChannelFuture f = ctx.writeAndFlush(new UnixTime());// 当写如完成的时候,执行f.addListener(ChannelFutureListener.CLOSE);}public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close();} }

 

client:handler

public class TimeClientHandler3 extends ChannelInboundHandlerAdapter {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {UnixTime m = (UnixTime) msg;System.out.println(m);ctx.close();}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close();} }


转载于:https://my.oschina.net/u/2246410/blog/652313

总结

以上是生活随笔为你收集整理的java的网络工具netty简介的全部内容,希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错,欢迎将生活随笔推荐给好友。