netty簡(jiǎn)單的服務(wù)端和客戶端創(chuàng)建

1、 Netty前言：高性能的三大主題：1) 傳輸：用什么樣的通道將數(shù)據(jù)發(fā)送給對(duì)方，BIO、NIO或者AIO，IO模型在很大程度上決定了框架的性能。2) 協(xié)議：采用什么樣的通信協(xié)議，HTTP或者內(nèi)部私有協(xié)議。協(xié)議的選擇不同，性能模型也不同。相比于公有協(xié)議，內(nèi)部私有協(xié)議的性能通常可以被設(shè)計(jì)的更優(yōu)。3) 線程：數(shù)據(jù)報(bào)如何讀取？讀取之后的編解碼在哪個(gè)線程進(jìn)行，編解碼后的消息如何派發(fā)，Reactor線程模型的不同，對(duì)性能的影響也非常大。Netty 高性能的表現(xiàn)：1. 異步非阻塞通信2. 零拷貝1) Netty的接收和發(fā)送ByteBuffer采用DIRECT BUFFERS，使用堆外直接內(nèi)存進(jìn)行Socket

2、讀寫(xiě)，不需要進(jìn)行字節(jié)緩沖區(qū)的二次拷貝2) Netty提供了組合Buffer對(duì)象，可以聚合多個(gè)ByteBuffer對(duì)象，用戶可以像操作一個(gè)Buffer那樣方便的對(duì)組合Buffer進(jìn)行操作，避免系統(tǒng)統(tǒng)通過(guò)內(nèi)存拷貝的方式將幾個(gè)小Buffer合并成一個(gè)大的Buffer。3) Netty的文件傳輸采用了transferTo方法，它可以直接將文件緩沖區(qū)的數(shù)據(jù)發(fā)送到目標(biāo)Channel，避免了傳統(tǒng)通過(guò)循環(huán)write方式導(dǎo)致的內(nèi)存拷貝問(wèn)題。3. 內(nèi)存池4. 高效的Reactor線程模型5. 無(wú)鎖化的串行設(shè)計(jì)理念6. 高效的并發(fā)編程7. 高性能的序列化框架8. 靈活的TCP參數(shù)配置能力Netty 服務(wù)端創(chuàng)建（詳細(xì)

3、解析）：1. 創(chuàng)建TimeService：代碼如下：package ty.demo;import ty.bootstrap.ServerBootstrap;import ty.channel.ChannelFuture;import ty.channel.ChannelInitializer;import ty.channel.ChannelOption;import ty.channel.ChannelPipeline;import ty.channel.EventLoopGroup;import ty.channel.nio.NioEventLoopGroup;import ty.chan

4、nel.socket.SocketChannel;import ty.channel.socket.nio.NioServerSocketChannel;import ty.handler.codec.DelimiterBasedFrameDecoder;import ty.handler.codec.Delimiters;import ty.handler.codec.string.StringDecoder;import ty.handler.codec.string.StringEncoder;/* * Time Netty服務(wù)端 * * author zhouxm * */public

5、 class TimeService public void bind(int port) throws Exception / 配置NIO服務(wù)端線程組EventLoopGroup bossGroup = new NioEventLoopGroup();/ 用于服務(wù)端接收客戶端的鏈接EventLoopGroup workGroup = new NioEventLoopGroup();/ 用于SocketChannel網(wǎng)絡(luò)讀寫(xiě)try / 創(chuàng)建NIO服務(wù)端輔助啟動(dòng)類ServerBootstrap b = new ServerBootstrap();b.group(bossGroup, workGr

6、oup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 1024).childHandler(new ChildChannelHander();/ 綁定監(jiān)聽(tīng)端口，同步等待成功ChannelFuture f = b.bind(port).sync();/ 等待監(jiān)聽(tīng)端口關(guān)閉f.channel().closeFuture().sync();/ b.bind(port).sync().channel().closeFuture().sync();/也可以這樣將上面兩步驟合并，使代碼更簡(jiǎn)潔 finally /

7、 退出，釋放線程池資源bossGroup.shutdownGracefully();workGroup.shutdownGracefully();/ 用戶處理網(wǎng)絡(luò)IO事件,類似于Reactor模式中的handle類 例：消息編解碼 日志打印private class ChildChannelHander extends ChannelInitializer<SocketChannel> Overrideprotected void initChannel(SocketChannel arg0) throws Exception / TODO Auto-generated metho

8、d stubChannelPipeline pipeline = arg0.pipeline();/ 以("n")為結(jié)尾分割的 解碼器pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192,Delimiters.lineDelimiter();/ 字符串解碼 和 編碼pipeline.addLast("decoder", new StringDecoder();pipeline.addLast("encoder", new StringEncod

9、er();/ 自己的邏輯Handlerpipeline.addLast("handler", new TimeServerHander();public static void main(String args)int port = 8080;try new TimeService().bind(port); catch (Exception e) / TODO Auto-generated catch blocke.printStackTrace();2. TimeServerHander：代碼如下：package ty.demo;import ty.buffer.Byt

10、eBuf;import ty.buffer.Unpooled;import ty.channel.ChannelHandlerAdapter;import ty.channel.ChannelHandlerContext;/* * 邏輯handle類 * * author zhouxm * */public class TimeServerHander extends ChannelHandlerAdapter /* * (non-Javadoc) * * see * ty.channel.ChannelHandlerAdapter#channelActive(ty.channel * .Ch

11、annelHandlerContext) */Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception / TODO Auto-generated method stubsuper.channelActive(ctx);/* * (non-Javadoc) * * see ty.channel.ChannelHandlerAdapter#channelRead(ty.channel. * ChannelHandlerContext, java.lang.Object) */Overridepubl

12、ic void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception / TODO Auto-generated method stubByteBuf buf = (ByteBuf) msg;/ 將msg轉(zhuǎn)換成Netty ByteBuf 類似于jdk中的ByteBufferbyte req = new bytebuf.readableBytes();/ 根據(jù)字節(jié)數(shù)創(chuàng)建byte數(shù)組buf.readBytes(req);/ 將緩沖區(qū)中的字節(jié)數(shù)組復(fù)制到 byte數(shù)組中String body = new String(re

13、q, "utf-8");/ 創(chuàng)建構(gòu)造函數(shù)接收消息System.out.print("client send message :" + body);/ 打印String resString = "response msg to client "ByteBuf b = Unpooled.copiedBuffer(resString.getBytes();ctx.write(b);/ 發(fā)送消息給客戶端/* * (non-Javadoc) * * see * ty.channel.ChannelHandlerAdapter#channelRe

14、adComplete(ty.channel * .ChannelHandlerContext) */Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception / TODO Auto-generated method stubctx.flush();/ 將隊(duì)列中的消息寫(xiě)入到SocketChannel發(fā)送給客戶端/* * (non-Javadoc) * * see * ty.channel.ChannelHandlerAdapter#exceptionCaught(ty.channel *

15、 .ChannelHandlerContext, java.lang.Throwable) */Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)throws Exception / TODO Auto-generated method stubctx.close();/ 關(guān)閉，釋放資源創(chuàng)建Netty客戶端：一創(chuàng)建TimeClient:package ty.demo;import ty.bootstrap.Bootstrap;import ty.channel.ChannelFuture

16、;import ty.channel.ChannelInitializer;import ty.channel.ChannelOption;import ty.channel.ChannelPipeline;import ty.channel.EventLoopGroup;import ty.channel.nio.NioEventLoopGroup;import ty.channel.socket.SocketChannel;import ty.channel.socket.nio.NioSocketChannel;import ty.handler.codec.DelimiterBased

17、FrameDecoder;import ty.handler.codec.Delimiters;import ty.handler.codec.string.StringDecoder;import ty.handler.codec.string.StringEncoder;public class TimeClient public void connect(String host, int port) throws Exception / 配置客戶端NIO線程組EventLoopGroup group = new NioEventLoopGroup();try / 創(chuàng)建NIO客戶端輔助啟動(dòng)

18、類Bootstrap b = new Bootstrap();b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true).handler(new ClientTimeHander();/發(fā)起異步鏈接操作ChannelFuture f = b.connect(host, port).sync();/等待客戶端鏈路關(guān)閉f.channel().closeFuture().sync(); finally group.shutdownGracefully();/ 關(guān)閉 釋放資源private

19、 class ClientTimeHander extends ChannelInitializer<SocketChannel> Overrideprotected void initChannel(SocketChannel arg0) throws Exception / TODO Auto-generated method stubChannelPipeline pipeline = arg0.pipeline();/ 以("n")為結(jié)尾分割的 解碼器pipeline.addLast("framer", new DelimiterBa

20、sedFrameDecoder(8192,Delimiters.lineDelimiter();/ 字符串解碼 和 編碼pipeline.addLast("decoder", new StringDecoder();pipeline.addLast("encoder", new StringEncoder();/ 自己的邏輯Handlerpipeline.addLast("handler", new TimeClientHander();public static void main(String args) / TODO Auto-

21、generated method stubString host = "localhost"int port = 8080;try new TimeClient().connect(host, port); catch (Exception e) / TODO Auto-generated catch blocke.printStackTrace();二創(chuàng)建TimeClientHander:package ty.demo;import ty.buffer.ByteBuf;import ty.buffer.Unpooled;import ty.channel.ChannelH

22、andlerAdapter;import ty.channel.ChannelHandlerContext;public class TimeClientHander extends ChannelHandlerAdapter private final ByteBuf firstMsg;public TimeClientHander()byte req = "send msg to server date".getBytes();firstMsg = Unpooled.buffer(req.length);firstMsg.writeBytes(req);/* (non-Javadoc) * see ty.channel.ChannelHandlerAdapter#channelActive(ty.channel.ChannelHandlerContext) */Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception / TODO Auto-generated method stubctx.writeAndFlush(firstMsg);/* (non-Javadoc) * see ty.channel.Chan