Netty实践
Netty是JBOSS針對網絡開發的一套應用框架，它也是在NIO的基礎上發展起來的。netty基于異步的事件驅動，具有高性能、高擴展性等特性，它提供了統一的底層協議接口，使得開發者從底層的網絡協議（比如 TCP/IP、UDP）中解脫出來。
TCP傳輸面向的是字節流，存在粘包半包問題。Netty提供了三種基本的解碼類（顯然只有讀數據時才會有該問題）來解決粘包拆包問題：LineBasedFrameDecoder、DelimiterBasedFrameDecoder、LengthFieldBasedFrameDecoder。其中，前二者所讀碼流到達指定長度或遇到分隔符時認為結束，若讀到的數據大于指定長度則拋TooLongFrameException并忽略之前讀到的碼流；最后一者每次讀固定長度碼流。此外，也針對特定協議提供了一些解決該問題的解碼類，如ProtobufVarint32FrameDecoder）
Netty封裝得很好，使得使用起來比較容易，按照“套路”填代碼即可。給幾個示例：
1、示例
Maven依賴：
1 <dependency> 2 <groupId>io.netty</groupId> 3 <artifactId>netty-all</artifactId> 4 <version>5.0.0.Alpha2</version> 5 </dependency> 6 7 <dependency><!-- 只有使用到Protobuf時才需要 --> 8 <groupId>com.google.protobuf</groupId> 9 <artifactId>protobuf-java</artifactId> 10 <version>2.5.0</version> 11 </dependency> View Code1.1、服務端/客戶端
Server:
1 public class SimpleChatServer { 2 3 private static int port = 8080; 4 5 public SimpleChatServer(int port) { 6 this.port = port; 7 } 8 9 public void run() throws Exception { 10 11 EventLoopGroup bossGroup = new NioEventLoopGroup(); 12 EventLoopGroup workerGroup = new NioEventLoopGroup(); 13 try { 14 ServerBootstrap b = new ServerBootstrap(); 15 b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 128) 16 .childHandler(new SimpleChatServerInitializer()).childOption(ChannelOption.SO_KEEPALIVE, true); 17 18 System.out.println("server 啟動了"); 19 20 // 綁定端口，開始接收進來的連接 21 // b.bind(11122);//可以綁定多個端口 22 ChannelFuture f = b.bind(port).sync(); 23 24 // 等待服務器 socket 關閉 。 25 // 在這個例子中，這不會發生，但你可以優雅地關閉你的服務器。 26 f.channel().closeFuture().sync(); 27 } finally { 28 workerGroup.shutdownGracefully(); 29 bossGroup.shutdownGracefully(); 30 31 System.out.println("server 關閉了"); 32 } 33 } 34 35 public static void main(String[] args) throws Exception { 36 new SimpleChatServer(port).run(); 37 } 38 } SimpleChatServer 1 public class SimpleChatServerInitializer extends ChannelInitializer<SocketChannel> { 2 private static final StringDecoder DECODER = new StringDecoder(); 3 private static final StringEncoder ENCODER = new StringEncoder(); 4 5 @Override 6 public void initChannel(SocketChannel ch) throws Exception {// Pipeline里的Handler是從底層開始向上添加的，故流動方向為后添加的輸出給先添加的、或先添加的讀入給后添加的 7 ChannelPipeline pipeline = ch.pipeline(); 8 9 // 添加ChannelHandler，順序是敏感的；名字任意，不沖突即可，也可以不指定名字 10 11 // Netty提供了三種解碼器解決粘包拆包問題：LineBasedFrameDecoder、DelimiterBasedFrameDecoder、LengthFieldBasedFrameDecoder。 12 // 其中，前二者所讀碼流到達指定長度或遇到分隔符時認為結束，若讀到的數據大于指定長度則拋TooLongFrameException并忽略之前讀到的碼流；最后一者每次讀固定長度碼流。 13 // 也可以繼承ByteToMessageDecoder自己處理 14 pipeline.addLast("FrameDecoder", new ProtobufVarint32FrameDecoder()); 15 pipeline.addLast("StringDecoder", DECODER); 16 17 // 解碼只會應用于讀數據時、編碼只會應用于寫數據時，因此解碼器與編碼器添加的先后順序在客戶端和服務端中可不同，但編碼器添加的順序須桶，解碼器亦然。 18 pipeline.addLast("StringEncoder", ENCODER); 19 20 pipeline.addLast("handler", new SimpleChatServerHandler()); 21 22 System.out.println("client " + ch.remoteAddress() + " 連接上"); 23 } 24 } SimpleChatServerInitializer 1 public class SimpleChatServerHandler extends SimpleChannelInboundHandler<String> { 2 3 public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); 4 5 // 一個客戶端連上再斷開時，六個事件的觸發順序：加入、(連接上(在SimpleChatServerInitializer中))、在線、異常、掉線、離開 6 @Override 7 public void handlerAdded(ChannelHandlerContext ctx) throws Exception {// 在ctx加入本Handler時觸發，一般在此做初始化工作，如創建buf 8 Channel incoming = ctx.channel(); 9 for (Channel channel : channels) { 10 channel.writeAndFlush("[SERVER] - " + incoming.remoteAddress() + " 加入\n"); 11 } 12 System.out.println("client " + incoming.remoteAddress() + " 加入"); 13 channels.add(ctx.channel()); 14 } 15 16 @Override 17 public void channelActive(ChannelHandlerContext ctx) throws Exception {// 當客戶端和服務端建立tcp成功之后，Netty的NIO線程會調用channelActive 18 Channel incoming = ctx.channel(); 19 System.out.println("client " + incoming.remoteAddress() + " 在線"); 20 } 21 22 @Override 23 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 24 Channel incoming = ctx.channel(); 25 System.err.println("client " + incoming.remoteAddress() + " 異常:" + cause.getMessage()); 26 // 當出現異常就關閉連接 27 // cause.printStackTrace(); 28 ctx.close(); 29 } 30 31 @Override 32 public void channelInactive(ChannelHandlerContext ctx) throws Exception { 33 Channel incoming = ctx.channel(); 34 System.out.println("client " + incoming.remoteAddress() + " 掉線"); 35 } 36 37 @Override 38 public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {// 從ctx移除本Handler時觸發 39 Channel incoming = ctx.channel(); 40 for (Channel channel : channels) { 41 channel.writeAndFlush("[SERVER] - " + incoming.remoteAddress() + " 離開\n"); 42 } 43 System.out.println("client " + incoming.remoteAddress() + " 離開"); 44 channels.remove(ctx.channel()); 45 } 46 47 // 優先級高于messageReceived方法，有了這個方法就會屏蔽messageReceived方法 48 // @Override 49 // public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 50 // System.out.println("channelRead"); 51 // Channel incoming = ctx.channel(); 52 // for (Channel channel : channels) { 53 // if (channel != incoming){ 54 // channel.writeAndFlush("[" + incoming.remoteAddress() + "]" + msg + "\n"); 55 // } else { 56 // channel.writeAndFlush("server: " + msg + "\n"); 57 // } 58 // } 59 // } 60 61 @Override 62 protected void messageReceived(ChannelHandlerContext ctx, String msg) throws Exception { 63 Channel incoming = ctx.channel(); 64 System.out.println("**" + incoming.remoteAddress() + " send: " + msg); 65 for (Channel channel : channels) { 66 if (channel != incoming) { 67 // System.out.println("[" + incoming.remoteAddress() + "] " + msg); 68 channel.writeAndFlush("[" + incoming.remoteAddress() + "] " + msg + "\n"); 69 } else { 70 // System.out.println("server: " + msg); 71 channel.writeAndFlush("server: " + msg + "\n"); 72 } 73 } 74 } 75 76 } SimpleChatServerHandlerClient:
1 public class SimpleChatClient { 2 3 private String host; 4 private int port; 5 6 public SimpleChatClient(String host, int port) { 7 this.host = host; 8 this.port = port; 9 } 10 11 public void run() throws Exception { 12 EventLoopGroup group = new NioEventLoopGroup(); 13 try { 14 Bootstrap bootstrap = new Bootstrap().group(group).channel(NioSocketChannel.class) 15 .handler(new SimpleChatClientInitializer()); 16 Channel channel = bootstrap.connect(host, port).sync().channel(); 17 18 Scanner sc = new Scanner(System.in); 19 System.out.println("please enter..."); 20 boolean exit = false; 21 // 輸入exit，退出系統 22 while (!exit) { 23 String str = sc.next(); 24 channel.writeAndFlush(str + "\r\n"); 25 if (str.equalsIgnoreCase("exit")) { 26 exit = true; 27 channel.close(); 28 } 29 } 30 sc.close(); 31 } catch (Exception e) { 32 e.printStackTrace(); 33 } finally { 34 group.shutdownGracefully(); 35 } 36 } 37 38 public static void main(String[] args) throws Exception { 39 new SimpleChatClient("localhost", 8080).run(); 40 } 41 } SimpleChatClient 1 public class SimpleChatClientInitializer extends ChannelInitializer<SocketChannel> { 2 private static final StringDecoder DECODER = new StringDecoder(); 3 private static final StringEncoder ENCODER = new StringEncoder(); 4 5 @Override 6 public void initChannel(SocketChannel ch) throws Exception {// Pipeline里的Handler是從底層開始向上添加的，故流動方向為后添加的輸出給先添加的、或先添加的讀入給后添加的 7 ChannelPipeline pipeline = ch.pipeline(); 8 9 // 添加ChannelHandler，順序是敏感的；名字任意，不沖突即可，也可以不指定名字 10 11 // Netty提供了三種解碼器解決粘包拆包問題：LineBasedFrameDecoder、DelimiterBasedFrameDecoder、LengthFieldBasedFrameDecoder。 12 // 其中，前二者所讀碼流到達指定長度或遇到分隔符時認為結束，若讀到的數據大于指定長度則拋TooLongFrameException并忽略之前讀到的碼流；最后一者每次讀固定長度碼流。 13 // 也可以繼承ByteToMessageDecoder自己處理 14 pipeline.addLast("FrameDecoder", new ProtobufVarint32FrameDecoder()); 15 pipeline.addLast("StringDecoder", DECODER); 16 17 // 解碼只會應用于讀數據時、編碼只會應用于寫數據時，因此解碼器與編碼器添加的先后順序在客戶端和服務端中可不同，但編碼器添加的順序須桶，解碼器亦然。 18 pipeline.addLast("StringEncoder", ENCODER); 19 20 pipeline.addLast("handler", new SimpleChatClientHandler()); 21 } 22 } SimpleChatClientInitializer 1 public class SimpleChatClientHandler extends SimpleChannelInboundHandler<String> { 2 3 // 優先級高于messageReceived方法，有了這個方法就會屏蔽messageReceived方法 4 // @Override 5 // public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 6 // System.out.println(msg.toString()); 7 // } 8 9 @Override 10 protected void messageReceived(ChannelHandlerContext ctx, String msg) throws Exception { 11 System.out.println(msg); 12 } 13 } SimpleChatClientHandler?
以下幾個是官方示例（翻譯）：
1.2、DiscardServer
服務端接收客戶端的消息，不返回任何信息：
1 class DiscardServerHandler extends ChannelHandlerAdapter { // (1) 2 3 // @Override 4 // public void channelRead(ChannelHandlerContext ctx, Object msg) { // (2) 5 // // 以靜默方式丟棄接收的數據 6 // ((ByteBuf) msg).release(); // (3)ByteBuf屬于引用計數的對象，必須通過release()方法顯式釋放。 7 // } 8 9 @Override 10 public void channelRead(ChannelHandlerContext ctx, Object msg) { // (2) 11 ByteBuf in = (ByteBuf) msg; 12 try { 13 System.out.println(in.toString(Charset.defaultCharset())); 14 while (in.isReadable()) { // (1) 15 System.out.print((char) in.readByte()); 16 System.out.flush(); 17 } // 這個低效的循環可以簡化為 System.out.println(in.toString(io.netty.util.CharsetUtil.US_ASCII)); 18 } finally { 19 ReferenceCountUtil.release(msg);// (2) // 或者寫為 in.release(); 20 } 21 // 或者直接打印 22 System.out.println("Yes, A new client in = " + ctx.name()); 23 } 24 25 @Override 26 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (4) 27 // 出現異常時關閉連接。 28 cause.printStackTrace(); 29 ctx.close(); 30 } 31 } 32 33 public class DiscardServer {// 可以用telnet連接并輸入進行測試，客戶端沒有任何輸出 34 private int port; 35 36 public DiscardServer(int port 37 38 ) { 39 this.port = port; 40 } 41 42 public void run() throws Exception { 43 EventLoopGroup bossGroup = new NioEventLoopGroup(); // (1)NioEventLoopGroup是處理I/O操作的多線程事件循環。第一個通常稱為“boss”，接受傳入連接。 第二個通常稱為“worker”，當“boss”接受連接并且向“worker”注冊接受連接，則“worker”處理所接受連接的流量。 44 // 使用多少個線程以及如何將它們映射到創建的通道取決于EventLoopGroup實現，甚至可以通過構造函數進行配置。 45 EventLoopGroup workerGroup = new NioEventLoopGroup(); 46 try { 47 ServerBootstrap b = new ServerBootstrap(); // (2) 48 b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class) // (3)該類用于實例化新的通道以接受傳入連接 49 .childHandler(new ChannelInitializer<SocketChannel>() { // (4)ChannelInitializer是一個特殊的處理程序，用于幫助用戶配置新的通道。 很可能要通過添加一些處理程序。隨著應用程序變得復雜，可能會向管道中添加更多處理程序 50 @Override 51 public void initChannel(SocketChannel ch) throws Exception { 52 ch.pipeline().addLast(new DiscardServerHandler());// 每次有新連接都會創建一個新的DiscardServerHandler處理之，，而不是只用一個來處理所有的 53 } 54 }).option(ChannelOption.SO_BACKLOG, 128) // (5)指定Channel實現的參數 55 .childOption(ChannelOption.SO_KEEPALIVE, true); // (6) 56 57 System.out.println("server 啟動了"); 58 59 // Bind and start to accept incoming connections. 60 // b.bind(11122);//可以綁定多個端口 61 ChannelFuture f = b.bind(port).sync(); // (7) 62 63 // Wait until the server socket is closed. 64 // In this example, this does not happen, but you can do that to gracefully 65 // shut down your server. 66 f.channel().closeFuture().sync(); 67 } finally { 68 workerGroup.shutdownGracefully(); 69 bossGroup.shutdownGracefully(); 70 71 System.out.println("server 關閉了"); 72 } 73 } 74 75 public static void main(String[] args) throws Exception { 76 int port; 77 if (args.length > 0) { 78 port = Integer.parseInt(args[0]); 79 } else { 80 port = 8080; 81 } 82 new DiscardServer(port).run();// 可以用telnet連接并輸入進行測試，客戶端沒有任何輸出 83 } 84 } DiscardServer1.3、TimeServer_StremBased
時間服務器：服務端收到客戶端連接就往客戶端發生時間，發完后就關閉連接；客戶端連接上服務端，收到消息后就關閉連接，讀的一方需要處理 粘包拆包 的問題
1 //http://www.yiibai.com/netty/netty-time-server.htmlclass 2 /** 3 * 時間服務器<br> 4 * 服務端收到客戶端連接就往客戶端發生時間，發完后就關閉連接；客戶端連接上服務端，收到消息后就關閉連接。<br> 5 * 只有讀者會有粘包拆包的問題，所以這里只有客戶端可能有該問題。 6 */ 7 class TimeServerHandler_StreamBased extends ChannelHandlerAdapter { 8 // 因為時間服務器將忽略任何接收到的數據，但是一旦建立連接就發送消息，所以我們不能使用channelRead()方法。而是覆蓋channelActive()方法。 9 @Override 10 public void channelActive(final ChannelHandlerContext ctx) { // (1) 11 final ByteBuf time = ctx.alloc().buffer(4); // (2) 12 13 // 在Java NIO中發送消息之前需調用java.nio.ByteBuffer.flip()，但Netty ByteBuf沒有這樣的方法，它只有兩個指針;一個用于讀取操作，另一個用于寫入操作。當您向ByteBuf寫入內容時，寫入索引會增加，而讀取器索引不會更改。讀取器索引和寫入器索引分別表示消息的開始和結束位置。 14 time.writeInt((int) (System.currentTimeMillis() / 1000L + 2208988800L)); 15 16 // ChannelHandlerContext.write()(和writeAndFlush())方法返回一個ChannelFuture。 17 // ChannelFuture表示尚未發生的I/O操作。這意味著，任何請求的操作可能尚未執行，因為所有操作在Netty中是異步的。因此，需要在ChannelFuture完成后調用close()方法。注意，close()也可能不會立即關閉連接，并返回一個ChannelFuture。 18 final ChannelFuture f = ctx.writeAndFlush(time); // (3) 19 20 // 當寫請求完成時，我們如何得到通知？添加監聽器 21 f.addListener(new ChannelFutureListener() { 22 @Override 23 public void operationComplete(ChannelFuture future) { 24 assert f == future; 25 ctx.close();// 發完就關閉連接 26 } 27 }); // (4)// 可以使用預定義的監聽器來簡化代碼：f.addListener(ChannelFutureListener.CLOSE); 28 } 29 30 @Override 31 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 32 cause.printStackTrace(); 33 ctx.close(); 34 } 35 } 36 37 public class TimeServer_StreamBased { 38 private int port; 39 40 public TimeServer_StreamBased(int port) { 41 this.port = port; 42 } 43 44 public void run() throws Exception { 45 EventLoopGroup bossGroup = new NioEventLoopGroup(); // (1)NioEventLoopGroup是處理I/O操作的多線程事件循環。第一個通常稱為“boss”，接受傳入連接。 第二個通常稱為“worker”，當“boss”接受連接并且向“worker”注冊接受連接，則“worker”處理所接受連接的流量。 46 // 使用多少個線程以及如何將它們映射到創建的通道取決于EventLoopGroup實現，甚至可以通過構造函數進行配置。 47 EventLoopGroup workerGroup = new NioEventLoopGroup(); 48 try { 49 ServerBootstrap b = new ServerBootstrap(); // (2) 50 b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class) // (3)該類用于實例化新的通道以接受傳入連接 51 .childHandler(new ChannelInitializer<SocketChannel>() { // (4)ChannelInitializer是一個特殊的處理程序，用于幫助用戶配置新的通道。 很可能要通過添加一些處理程序。隨著應用程序變得復雜，可能會向管道中添加更多處理程序 52 @Override 53 public void initChannel(SocketChannel ch) throws Exception { 54 ch.pipeline().addLast(new TimeServerHandler_StreamBased());// 每次有新連接就創建一個新的該handler來處理，而不是只用一個來處理所有的 55 } 56 }).option(ChannelOption.SO_BACKLOG, 128) // (5)指定Channel實現的參數 57 .childOption(ChannelOption.SO_KEEPALIVE, true); // (6) 58 59 System.out.println("server 啟動了"); 60 61 // Bind and start to accept incoming connections. 62 // b.bind(11122);//可以綁定多個端口 63 ChannelFuture f = b.bind(port).sync(); // (7) 64 65 // Wait until the server socket is closed. 66 // In this example, this does not happen, but you can do that to gracefully 67 // shut down your server. 68 f.channel().closeFuture().sync(); 69 } finally { 70 workerGroup.shutdownGracefully(); 71 bossGroup.shutdownGracefully(); 72 73 System.out.println("server 關閉了"); 74 } 75 } 76 77 public static void main(String[] args) throws Exception { 78 int port = 8080; 79 80 new TimeServer_StreamBased(port).run();// 可以用telnet連接并輸入進行測試，客戶端輸出二進制的32位整數，所以為亂碼 81 } 82 } 83 84 class TimeClient_StreamBased { 85 public static void main(String[] args) throws Exception { 86 String host = "localhost"; 87 int port = 8080; 88 EventLoopGroup workerGroup = new NioEventLoopGroup(); 89 90 try { 91 Bootstrap b = new Bootstrap(); // (1)Bootstrap與ServerBootstrap類似，只是它用于非服務器通道，例如客戶端或無連接通道。 92 b.group(workerGroup); // (2)如果只指定一個EventLoopGroup，它將同時用作boss組和worker組。boss組和worker組不是用于客戶端。 93 b.channel(NioSocketChannel.class); // (3)不使用NioServerSocketChannel，而是使用NioSocketChannel來創建客戶端通道。 94 b.option(ChannelOption.SO_KEEPALIVE, true); // (4)這里不像我們使用的ServerBootstrap，所以不使用childOption()，因為客戶端SocketChannel沒有父類。 95 b.handler(new ChannelInitializer<SocketChannel>() { 96 @Override 97 public void initChannel(SocketChannel ch) throws Exception { 98 // 以下兩種方式都可以解決讀取粘包拆包的問題 99 ch.pipeline().addLast(new TimeClientHandler_1_TimeDecoder(), new TimeClientHandler_1_withproblem()); 100 // ch.pipeline().addLast(new TimeClientHandler_2()); 101 } 102 }); 103 104 // Start the client. 105 ChannelFuture f = b.connect(host, port).sync(); // (5)應該調用connect()方法，而不是bind()方法 106 107 // Wait until the connection is closed. 108 f.channel().closeFuture().sync(); 109 } finally { 110 workerGroup.shutdownGracefully(); 111 } 112 } 113 } 114 115 /** 116 * 此Handler單獨使用的話存在粘包拆包的問題 117 */ 118 class TimeClientHandler_1_withproblem extends ChannelHandlerAdapter { 119 @Override 120 public void channelRead(ChannelHandlerContext ctx, Object msg) {// 粘包拆包問題，在TCP/IP的基于流的傳輸中，接收的數據被存儲到套接字接收緩沖器中。不幸的是，基于流的傳輸的緩沖器不是分組的隊列，而是字節的隊列。這意味著，即使將兩個消息作為兩個獨立的數據包發送，操作系統也不會將它們視為兩個消息，而只是一組字節。 121 // 因此，不能保證讀的是輸入的完整。不過，由于返回的是32位int，數據量小，所以這里很少可能發生此問題。 122 ByteBuf m = (ByteBuf) msg; // (1) 123 try { 124 long currentTimeMillis = (m.readUnsignedInt() - 2208988800L) * 1000L;// 每次讀取到m的字節數不一定是4，所以存在粘包拆包的問題 125 Date currentTime = new Date(currentTimeMillis); 126 System.out.println("Default Date Format:" + currentTime.toString()); 127 128 SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); 129 String dateString = formatter.format(currentTime); 130 // 轉換一下成中國人的時間格式 131 System.out.println("Date Format:" + dateString); 132 133 ctx.close(); 134 } finally { 135 m.release(); 136 } 137 } 138 139 @Override 140 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 141 cause.printStackTrace(); 142 ctx.close(); 143 } 144 } 145 146 /** 147 * 與TimeClientHandler_1_withproblem搭配使用，解決粘包拆包問題，且易擴展 <br> 148 * <br> 149 * 在客戶端的ClientHandler之前加上一層，使得字節夠了才往ClientHandler傳，以解決粘包拆包的問題 150 */ 151 class TimeClientHandler_1_TimeDecoder extends ByteToMessageDecoder { // (1)ByteToMessageDecoder是ChannelHandlerAdapter的一個實現，它使得處理碎片問題變得容易 152 @Override 153 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) { // (2) 154 if (in.readableBytes() < 4) { 155 return; // (3) 156 } 157 out.add(in.readBytes(4)); // (4) 158 } 159 } 160 161 /** 162 * 此的Handler能解決讀取粘包拆包問題，但可擴展性差<br> 163 * <br> 164 * 在客戶端的ClientHandler里解決粘包拆包問題：字節夠了才讀 165 */ 166 class TimeClientHandler_2 extends ChannelHandlerAdapter { 167 private ByteBuf buf; 168 169 @Override 170 public void handlerAdded(ChannelHandlerContext ctx) { 171 buf = ctx.alloc().buffer(4); // (1)初始化 172 } 173 174 @Override 175 public void handlerRemoved(ChannelHandlerContext ctx) { 176 buf.release(); // (1)初始化的釋放 177 buf = null; 178 } 179 180 @Override 181 public void channelRead(ChannelHandlerContext ctx, Object msg) { 182 ByteBuf m = (ByteBuf) msg; 183 buf.writeBytes(m); // (2) 184 m.release(); 185 186 if (buf.readableBytes() >= 4) { // (3) 187 long currentTimeMillis = (buf.readUnsignedInt() - 2208988800L) * 1000L; 188 System.out.println(new Date(currentTimeMillis)); 189 ctx.close(); 190 } 191 } 192 193 @Override 194 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 195 cause.printStackTrace(); 196 ctx.close(); 197 } 198 } TimeServer_StreamBased1.4、TimeServer_POJOBased
時間服務器：同上，但通過POJO來通信，同樣，讀的一方需要處理 粘包拆包 的問題
1 /** 2 * 時間服務器<br> 3 * 服務端收到客戶端連接就往客戶端發送時間，發完后就關閉連接；客戶端連接上服務端，收到消息后就關閉連接。<br> 4 * 由于只有服務端往客戶端發數據，所以服務端只要encode、客戶端只要decode 5 */ 6 7 class UnixTime { 8 private final long value; 9 10 public UnixTime() { 11 this(System.currentTimeMillis() / 1000L + 2208988800L); 12 } 13 14 public UnixTime(long value) { 15 this.value = value; 16 } 17 18 public long value() { 19 return value; 20 } 21 22 @Override 23 public String toString() { 24 Date date = new Date((value() - 2208988800L) * 1000L); 25 SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); 26 String dateString = formatter.format(date); 27 return dateString; 28 } 29 } 30 31 class TimeEncoder_POJOBased extends MessageToByteEncoder<UnixTime> { 32 @Override 33 protected void encode(ChannelHandlerContext ctx, UnixTime msg, ByteBuf out) { 34 out.writeInt((int) msg.value()); 35 } 36 } 37 38 class TimeServerHandler_POJOBased extends ChannelHandlerAdapter { 39 @Override 40 public void channelActive(ChannelHandlerContext ctx) {// 當客戶端和服務端建立tcp成功之后，Netty的NIO線程會調用channelActive 41 ChannelFuture f = ctx.writeAndFlush(new UnixTime()); 42 f.addListener(ChannelFutureListener.CLOSE); 43 } 44 45 @Override 46 public void handlerAdded(ChannelHandlerContext ctx) throws Exception {// 在ctx加入本Handler時觸發，一般在此做初始化工作，如創建buf 47 Channel incoming = ctx.channel(); 48 System.out.println("client " + incoming.remoteAddress() + " 加入"); 49 } 50 51 @Override 52 public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {// 從ctx移除本Handler時觸發 53 Channel incoming = ctx.channel(); 54 System.out.println("client " + incoming.remoteAddress() + " 離開"); 55 } 56 57 @Override 58 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 59 cause.printStackTrace(); 60 ctx.close(); 61 } 62 } 63 64 public class TimeServer_POJOBased { 65 private int port; 66 67 public TimeServer_POJOBased(int port) { 68 this.port = port; 69 } 70 71 public void run() throws Exception { 72 // NioEventLoopGroup類是個線程組，包含一組NIO線程，用于網絡事件的處理（實際上它就是Reactor線程組）。 創建的2個線程組，1個用于服務端接收客戶端的連接，另一個用于SocketChannel的 網絡讀寫 73 EventLoopGroup bossGroup = new NioEventLoopGroup(); // (1) 74 EventLoopGroup workerGroup = new NioEventLoopGroup(); 75 try { 76 // ServerBootstrap類是啟動NIO服務器的輔助啟動類 77 ServerBootstrap b = new ServerBootstrap(); // (2) 78 b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class) // (3) 79 .option(ChannelOption.SO_BACKLOG, 128).childHandler(new ChannelInitializer<SocketChannel>() { // (4) 80 @Override 81 public void initChannel(SocketChannel ch) throws Exception { 82 // Pipeline里的Handler是從底層開始向上疊加的，即后者輸出給前者、或前者讀出的給后者 83 // 由于只有服務端往客戶端發數據，所以服務端只要encode、客戶端只要decode 84 ch.pipeline().addLast(new TimeEncoder_POJOBased(), new TimeServerHandler_POJOBased()); 85 } 86 })// (5) 87 .childOption(ChannelOption.SO_KEEPALIVE, true); // (6) 88 89 System.out.println("server 啟動了"); 90 91 // 綁定端口，開始接收進來的連接 92 // b.bind(11122);//可以綁定多個端口 93 ChannelFuture f = b.bind(port).sync(); // (7) 94 95 // 等待服務器 socket 關閉 。 96 // 在這個例子中，這不會發生，但你可以優雅地關閉你的服務器。 97 f.channel().closeFuture().sync(); 98 } finally { 99 workerGroup.shutdownGracefully(); 100 bossGroup.shutdownGracefully(); 101 System.out.println("server 關閉了"); 102 } 103 } 104 105 public static void main(String[] args) throws Exception { 106 int port; 107 if (args.length > 0) { 108 port = Integer.parseInt(args[0]); 109 } else { 110 port = 8080; 111 } 112 new TimeServer_POJOBased(port).run(); 113 } 114 } 115 116 class TimeClient_POJOBased { 117 118 public static void main(String[] args) throws Exception { 119 120 String host = "127.0.0.1"; 121 int port = 8080; 122 EventLoopGroup workerGroup = new NioEventLoopGroup(); 123 124 try { 125 Bootstrap b = new Bootstrap(); // (1) 126 b.group(workerGroup); // (2) 127 b.channel(NioSocketChannel.class); // (3) 128 b.option(ChannelOption.SO_KEEPALIVE, true); // (4) 129 b.handler(new ChannelInitializer<SocketChannel>() { 130 @Override 131 public void initChannel(SocketChannel ch) throws Exception { 132 // Pipeline里的Handler是從底層開始向上添加的，故流動方向為后添加的輸出給先添加的、或先添加的讀入給后添加的 133 // 由于只有服務端往客戶端發數據，所以服務端只要encode、客戶端只要decode 134 ch.pipeline().addLast(new TimeDecoder_POJOBased(), new TimeClientHandler_POJOBased()); 135 } 136 }); 137 138 // 啟動客戶端 139 ChannelFuture f = b.connect(host, port).sync(); // (5) 140 141 // 等待連接關閉 142 f.channel().closeFuture().sync(); 143 } finally { 144 workerGroup.shutdownGracefully(); 145 } 146 } 147 } 148 149 class TimeDecoder_POJOBased extends ByteToMessageDecoder { 150 @Override 151 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { 152 // TODO Auto-generated method stub 153 if (in.readableBytes() < 4) {// 只有讀者存在粘包半包問題，這里不少于4個字節時才處理，以避免該問題 154 return; 155 } 156 out.add(new UnixTime(in.readUnsignedInt())); 157 } 158 } 159 160 class TimeClientHandler_POJOBased extends ChannelHandlerAdapter { 161 // @Override 162 public void channelRead(ChannelHandlerContext ctx, Object msg) { 163 UnixTime m = (UnixTime) msg; 164 System.out.println(m); 165 ctx.close(); 166 } 167 168 @Override 169 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 170 cause.printStackTrace(); 171 ctx.close(); 172 } 173 } TimeServer_POJOBased1.5、結合Protobuf
（就是把自動生成的類當成一個JavaBean來用，但是該JavaBean比普通的Bean更強大，如包含了與Protobuf格式間進行序列化/反序列化的方法）
Netty提供了與Protobuf相關的 解決粘包半包問題的編解碼器(ByteBuf與ByteBuf間) 以及 ByteBuf與自定義Protobuf類間的編解碼器 ：
- ProtobufVarint32FrameDecoder()
- ProtobufVarint32LengthFieldPrepender()
- ProtobufDecoder(Custom Protobuf Class)
- ProtobufEncoder()
示例（需要添加Protobuf依賴，）：
1 public final class PersonProbuf { 2 private PersonProbuf() { 3 } 4 5 public static void registerAllExtensions(com.google.protobuf.ExtensionRegistry registry) { 6 } 7 8 public interface PersonOrBuilder extends com.google.protobuf.MessageOrBuilder { 9 10 // optional int64 id = 1; 11 /** 12 * <code>optional int64 id = 1;</code> 13 * 14 * <pre> 15 *可選的字段，為64位整數.. 16 * </pre> 17 */ 18 boolean hasId(); 19 20 /** 21 * <code>optional int64 id = 1;</code> 22 * 23 * <pre> 24 *可選的字段，為64位整數.. 25 * </pre> 26 */ 27 long getId(); 28 29 // optional string name = 2; 30 /** 31 * <code>optional string name = 2;</code> 32 */ 33 boolean hasName(); 34 35 /** 36 * <code>optional string name = 2;</code> 37 */ 38 java.lang.String getName(); 39 40 /** 41 * <code>optional string name = 2;</code> 42 */ 43 com.google.protobuf.ByteString getNameBytes(); 44 45 // optional string sex = 3; 46 /** 47 * <code>optional string sex = 3;</code> 48 */ 49 boolean hasSex(); 50 51 /** 52 * <code>optional string sex = 3;</code> 53 */ 54 java.lang.String getSex(); 55 56 /** 57 * <code>optional string sex = 3;</code> 58 */ 59 com.google.protobuf.ByteString getSexBytes(); 60 61 // optional string tel = 4; 62 /** 63 * <code>optional string tel = 4;</code> 64 */ 65 boolean hasTel(); 66 67 /** 68 * <code>optional string tel = 4;</code> 69 */ 70 java.lang.String getTel(); 71 72 /** 73 * <code>optional string tel = 4;</code> 74 */ 75 com.google.protobuf.ByteString getTelBytes(); 76 } 77 78 /** 79 * Protobuf type {@code Person} 80 */ 81 public static final class Person extends com.google.protobuf.GeneratedMessage implements PersonOrBuilder { 82 // Use Person.newBuilder() to construct. 83 private Person(com.google.protobuf.GeneratedMessage.Builder<?> builder) { 84 super(builder); 85 this.unknownFields = builder.getUnknownFields(); 86 } 87 88 private Person(boolean noInit) { 89 this.unknownFields = com.google.protobuf.UnknownFieldSet.getDefaultInstance(); 90 } 91 92 private static final Person defaultInstance; 93 94 public static Person getDefaultInstance() { 95 return defaultInstance; 96 } 97 98 public Person getDefaultInstanceForType() { 99 return defaultInstance; 100 } 101 102 private final com.google.protobuf.UnknownFieldSet unknownFields; 103 104 @java.lang.Override 105 public final com.google.protobuf.UnknownFieldSet getUnknownFields() { 106 return this.unknownFields; 107 } 108 109 private Person(com.google.protobuf.CodedInputStream input, 110 com.google.protobuf.ExtensionRegistryLite extensionRegistry) 111 throws com.google.protobuf.InvalidProtocolBufferException { 112 initFields(); 113 int mutable_bitField0_ = 0; 114 com.google.protobuf.UnknownFieldSet.Builder unknownFields = com.google.protobuf.UnknownFieldSet 115 .newBuilder(); 116 try { 117 boolean done = false; 118 while (!done) { 119 int tag = input.readTag(); 120 switch (tag) { 121 case 0: 122 done = true; 123 break; 124 default: { 125 if (!parseUnknownField(input, unknownFields, extensionRegistry, tag)) { 126 done = true; 127 } 128 break; 129 } 130 case 8: { 131 bitField0_ |= 0x00000001; 132 id_ = input.readInt64(); 133 break; 134 } 135 case 18: { 136 bitField0_ |= 0x00000002; 137 name_ = input.readBytes(); 138 break; 139 } 140 case 26: { 141 bitField0_ |= 0x00000004; 142 sex_ = input.readBytes(); 143 break; 144 } 145 case 34: { 146 bitField0_ |= 0x00000008; 147 tel_ = input.readBytes(); 148 break; 149 } 150 } 151 } 152 } catch (com.google.protobuf.InvalidProtocolBufferException e) { 153 throw e.setUnfinishedMessage(this); 154 } catch (java.io.IOException e) { 155 throw new com.google.protobuf.InvalidProtocolBufferException(e.getMessage()).setUnfinishedMessage(this); 156 } finally { 157 this.unknownFields = unknownFields.build(); 158 makeExtensionsImmutable(); 159 } 160 } 161 162 public static final com.google.protobuf.Descriptors.Descriptor getDescriptor() { 163 return PersonProbuf.internal_static_Person_descriptor; 164 } 165 166 protected com.google.protobuf.GeneratedMessage.FieldAccessorTable internalGetFieldAccessorTable() { 167 return PersonProbuf.internal_static_Person_fieldAccessorTable 168 .ensureFieldAccessorsInitialized(PersonProbuf.Person.class, PersonProbuf.Person.Builder.class); 169 } 170 171 public static com.google.protobuf.Parser<Person> PARSER = new com.google.protobuf.AbstractParser<Person>() { 172 public Person parsePartialFrom(com.google.protobuf.CodedInputStream input, 173 com.google.protobuf.ExtensionRegistryLite extensionRegistry) 174 throws com.google.protobuf.InvalidProtocolBufferException { 175 return new Person(input, extensionRegistry); 176 } 177 }; 178 179 @java.lang.Override 180 public com.google.protobuf.Parser<Person> getParserForType() { 181 return PARSER; 182 } 183 184 private int bitField0_; 185 // optional int64 id = 1; 186 public static final int ID_FIELD_NUMBER = 1; 187 private long id_; 188 189 /** 190 * <code>optional int64 id = 1;</code> 191 * 192 * <pre> 193 *可選的字段，為64位整數.. 194 * </pre> 195 */ 196 public boolean hasId() { 197 return ((bitField0_ & 0x00000001) == 0x00000001); 198 } 199 200 /** 201 * <code>optional int64 id = 1;</code> 202 * 203 * <pre> 204 *可選的字段，為64位整數.. 205 * </pre> 206 */ 207 public long getId() { 208 return id_; 209 } 210 211 // optional string name = 2; 212 public static final int NAME_FIELD_NUMBER = 2; 213 private java.lang.Object name_; 214 215 /** 216 * <code>optional string name = 2;</code> 217 */ 218 public boolean hasName() { 219 return ((bitField0_ & 0x00000002) == 0x00000002); 220 } 221 222 /** 223 * <code>optional string name = 2;</code> 224 */ 225 public java.lang.String getName() { 226 java.lang.Object ref = name_; 227 if (ref instanceof java.lang.String) { 228 return (java.lang.String) ref; 229 } else { 230 com.google.protobuf.ByteString bs = (com.google.protobuf.ByteString) ref; 231 java.lang.String s = bs.toStringUtf8(); 232 if (bs.isValidUtf8()) { 233 name_ = s; 234 } 235 return s; 236 } 237 } 238 239 /** 240 * <code>optional string name = 2;</code> 241 */ 242 public com.google.protobuf.ByteString getNameBytes() { 243 java.lang.Object ref = name_; 244 if (ref instanceof java.lang.String) { 245 com.google.protobuf.ByteString b = com.google.protobuf.ByteString.copyFromUtf8((java.lang.String) ref); 246 name_ = b; 247 return b; 248 } else { 249 return (com.google.protobuf.ByteString) ref; 250 } 251 } 252 253 // optional string sex = 3; 254 public static final int SEX_FIELD_NUMBER = 3; 255 private java.lang.Object sex_; 256 257 /** 258 * <code>optional string sex = 3;</code> 259 */ 260 public boolean hasSex() { 261 return ((bitField0_ & 0x00000004) == 0x00000004); 262 } 263 264 /** 265 * <code>optional string sex = 3;</code> 266 */ 267 public java.lang.String getSex() { 268 java.lang.Object ref = sex_; 269 if (ref instanceof java.lang.String) { 270 return (java.lang.String) ref; 271 } else { 272 com.google.protobuf.ByteString bs = (com.google.protobuf.ByteString) ref; 273 java.lang.String s = bs.toStringUtf8(); 274 if (bs.isValidUtf8()) { 275 sex_ = s; 276 } 277 return s; 278 } 279 } 280 281 /** 282 * <code>optional string sex = 3;</code> 283 */ 284 public com.google.protobuf.ByteString getSexBytes() { 285 java.lang.Object ref = sex_; 286 if (ref instanceof java.lang.String) { 287 com.google.protobuf.ByteString b = com.google.protobuf.ByteString.copyFromUtf8((java.lang.String) ref); 288 sex_ = b; 289 return b; 290 } else { 291 return (com.google.protobuf.ByteString) ref; 292 } 293 } 294 295 // optional string tel = 4; 296 public static final int TEL_FIELD_NUMBER = 4; 297 private java.lang.Object tel_; 298 299 /** 300 * <code>optional string tel = 4;</code> 301 */ 302 public boolean hasTel() { 303 return ((bitField0_ & 0x00000008) == 0x00000008); 304 } 305 306 /** 307 * <code>optional string tel = 4;</code> 308 */ 309 public java.lang.String getTel() { 310 java.lang.Object ref = tel_; 311 if (ref instanceof java.lang.String) { 312 return (java.lang.String) ref; 313 } else { 314 com.google.protobuf.ByteString bs = (com.google.protobuf.ByteString) ref; 315 java.lang.String s = bs.toStringUtf8(); 316 if (bs.isValidUtf8()) { 317 tel_ = s; 318 } 319 return s; 320 } 321 } 322 323 /** 324 * <code>optional string tel = 4;</code> 325 */ 326 public com.google.protobuf.ByteString getTelBytes() { 327 java.lang.Object ref = tel_; 328 if (ref instanceof java.lang.String) { 329 com.google.protobuf.ByteString b = com.google.protobuf.ByteString.copyFromUtf8((java.lang.String) ref); 330 tel_ = b; 331 return b; 332 } else { 333 return (com.google.protobuf.ByteString) ref; 334 } 335 } 336 337 private void initFields() { 338 id_ = 0L; 339 name_ = ""; 340 sex_ = ""; 341 tel_ = ""; 342 } 343 344 private byte memoizedIsInitialized = -1; 345 346 public final boolean isInitialized() { 347 byte isInitialized = memoizedIsInitialized; 348 if (isInitialized != -1) 349 return isInitialized == 1; 350 351 memoizedIsInitialized = 1; 352 return true; 353 } 354 355 public void writeTo(com.google.protobuf.CodedOutputStream output) throws java.io.IOException { 356 getSerializedSize(); 357 if (((bitField0_ & 0x00000001) == 0x00000001)) { 358 output.writeInt64(1, id_); 359 } 360 if (((bitField0_ & 0x00000002) == 0x00000002)) { 361 output.writeBytes(2, getNameBytes()); 362 } 363 if (((bitField0_ & 0x00000004) == 0x00000004)) { 364 output.writeBytes(3, getSexBytes()); 365 } 366 if (((bitField0_ & 0x00000008) == 0x00000008)) { 367 output.writeBytes(4, getTelBytes()); 368 } 369 getUnknownFields().writeTo(output); 370 } 371 372 private int memoizedSerializedSize = -1; 373 374 public int getSerializedSize() { 375 int size = memoizedSerializedSize; 376 if (size != -1) 377 return size; 378 379 size = 0; 380 if (((bitField0_ & 0x00000001) == 0x00000001)) { 381 size += com.google.protobuf.CodedOutputStream.computeInt64Size(1, id_); 382 } 383 if (((bitField0_ & 0x00000002) == 0x00000002)) { 384 size += com.google.protobuf.CodedOutputStream.computeBytesSize(2, getNameBytes()); 385 } 386 if (((bitField0_ & 0x00000004) == 0x00000004)) { 387 size += com.google.protobuf.CodedOutputStream.computeBytesSize(3, getSexBytes()); 388 } 389 if (((bitField0_ & 0x00000008) == 0x00000008)) { 390 size += com.google.protobuf.CodedOutputStream.computeBytesSize(4, getTelBytes()); 391 } 392 size += getUnknownFields().getSerializedSize(); 393 memoizedSerializedSize = size; 394 return size; 395 } 396 397 private static final long serialVersionUID = 0L; 398 399 @java.lang.Override 400 protected java.lang.Object writeReplace() throws java.io.ObjectStreamException { 401 return super.writeReplace(); 402 } 403 404 public static PersonProbuf.Person parseFrom(com.google.protobuf.ByteString data) 405 throws com.google.protobuf.InvalidProtocolBufferException { 406 return PARSER.parseFrom(data); 407 } 408 409 public static PersonProbuf.Person parseFrom(com.google.protobuf.ByteString data, 410 com.google.protobuf.ExtensionRegistryLite extensionRegistry) 411 throws com.google.protobuf.InvalidProtocolBufferException { 412 return PARSER.parseFrom(data, extensionRegistry); 413 } 414 415 public static PersonProbuf.Person parseFrom(byte[] data) 416 throws com.google.protobuf.InvalidProtocolBufferException { 417 return PARSER.parseFrom(data); 418 } 419 420 public static PersonProbuf.Person parseFrom(byte[] data, 421 com.google.protobuf.ExtensionRegistryLite extensionRegistry) 422 throws com.google.protobuf.InvalidProtocolBufferException { 423 return PARSER.parseFrom(data, extensionRegistry); 424 } 425 426 public static PersonProbuf.Person parseFrom(java.io.InputStream input) throws java.io.IOException { 427 return PARSER.parseFrom(input); 428 } 429 430 public static PersonProbuf.Person parseFrom(java.io.InputStream input, 431 com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException { 432 return PARSER.parseFrom(input, extensionRegistry); 433 } 434 435 public static PersonProbuf.Person parseDelimitedFrom(java.io.InputStream input) throws java.io.IOException { 436 return PARSER.parseDelimitedFrom(input); 437 } 438 439 public static PersonProbuf.Person parseDelimitedFrom(java.io.InputStream input, 440 com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException { 441 return PARSER.parseDelimitedFrom(input, extensionRegistry); 442 } 443 444 public static PersonProbuf.Person parseFrom(com.google.protobuf.CodedInputStream input) 445 throws java.io.IOException { 446 return PARSER.parseFrom(input); 447 } 448 449 public static PersonProbuf.Person parseFrom(com.google.protobuf.CodedInputStream input, 450 com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException { 451 return PARSER.parseFrom(input, extensionRegistry); 452 } 453 454 public static Builder newBuilder() { 455 return Builder.create(); 456 } 457 458 public Builder newBuilderForType() { 459 return newBuilder(); 460 } 461 462 public static Builder newBuilder(PersonProbuf.Person prototype) { 463 return newBuilder().mergeFrom(prototype); 464 } 465 466 public Builder toBuilder() { 467 return newBuilder(this); 468 } 469 470 @java.lang.Override 471 protected Builder newBuilderForType(com.google.protobuf.GeneratedMessage.BuilderParent parent) { 472 Builder builder = new Builder(parent); 473 return builder; 474 } 475 476 /** 477 * Protobuf type {@code Person} 478 */ 479 public static final class Builder extends com.google.protobuf.GeneratedMessage.Builder<Builder> 480 implements PersonProbuf.PersonOrBuilder { 481 public static final com.google.protobuf.Descriptors.Descriptor getDescriptor() { 482 return PersonProbuf.internal_static_Person_descriptor; 483 } 484 485 protected com.google.protobuf.GeneratedMessage.FieldAccessorTable internalGetFieldAccessorTable() { 486 return PersonProbuf.internal_static_Person_fieldAccessorTable 487 .ensureFieldAccessorsInitialized(PersonProbuf.Person.class, PersonProbuf.Person.Builder.class); 488 } 489 490 // Construct using PersonProbuf.Person.newBuilder() 491 private Builder() { 492 maybeForceBuilderInitialization(); 493 } 494 495 private Builder(com.google.protobuf.GeneratedMessage.BuilderParent parent) { 496 super(parent); 497 maybeForceBuilderInitialization(); 498 } 499 500 private void maybeForceBuilderInitialization() { 501 if (com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders) { 502 } 503 } 504 505 private static Builder create() { 506 return new Builder(); 507 } 508 509 public Builder clear() { 510 super.clear(); 511 id_ = 0L; 512 bitField0_ = (bitField0_ & ~0x00000001); 513 name_ = ""; 514 bitField0_ = (bitField0_ & ~0x00000002); 515 sex_ = ""; 516 bitField0_ = (bitField0_ & ~0x00000004); 517 tel_ = ""; 518 bitField0_ = (bitField0_ & ~0x00000008); 519 return this; 520 } 521 522 public Builder clone() { 523 return create().mergeFrom(buildPartial()); 524 } 525 526 public com.google.protobuf.Descriptors.Descriptor getDescriptorForType() { 527 return PersonProbuf.internal_static_Person_descriptor; 528 } 529 530 public PersonProbuf.Person getDefaultInstanceForType() { 531 return PersonProbuf.Person.getDefaultInstance(); 532 } 533 534 public PersonProbuf.Person build() { 535 PersonProbuf.Person result = buildPartial(); 536 if (!result.isInitialized()) { 537 throw newUninitializedMessageException(result); 538 } 539 return result; 540 } 541 542 public PersonProbuf.Person buildPartial() { 543 PersonProbuf.Person result = new PersonProbuf.Person(this); 544 int from_bitField0_ = bitField0_; 545 int to_bitField0_ = 0; 546 if (((from_bitField0_ & 0x00000001) == 0x00000001)) { 547 to_bitField0_ |= 0x00000001; 548 } 549 result.id_ = id_; 550 if (((from_bitField0_ & 0x00000002) == 0x00000002)) { 551 to_bitField0_ |= 0x00000002; 552 } 553 result.name_ = name_; 554 if (((from_bitField0_ & 0x00000004) == 0x00000004)) { 555 to_bitField0_ |= 0x00000004; 556 } 557 result.sex_ = sex_; 558 if (((from_bitField0_ & 0x00000008) == 0x00000008)) { 559 to_bitField0_ |= 0x00000008; 560 } 561 result.tel_ = tel_; 562 result.bitField0_ = to_bitField0_; 563 onBuilt(); 564 return result; 565 } 566 567 public Builder mergeFrom(com.google.protobuf.Message other) { 568 if (other instanceof PersonProbuf.Person) { 569 return mergeFrom((PersonProbuf.Person) other); 570 } else { 571 super.mergeFrom(other); 572 return this; 573 } 574 } 575 576 public Builder mergeFrom(PersonProbuf.Person other) { 577 if (other == PersonProbuf.Person.getDefaultInstance()) 578 return this; 579 if (other.hasId()) { 580 setId(other.getId()); 581 } 582 if (other.hasName()) { 583 bitField0_ |= 0x00000002; 584 name_ = other.name_; 585 onChanged(); 586 } 587 if (other.hasSex()) { 588 bitField0_ |= 0x00000004; 589 sex_ = other.sex_; 590 onChanged(); 591 } 592 if (other.hasTel()) { 593 bitField0_ |= 0x00000008; 594 tel_ = other.tel_; 595 onChanged(); 596 } 597 this.mergeUnknownFields(other.getUnknownFields()); 598 return this; 599 } 600 601 public final boolean isInitialized() { 602 return true; 603 } 604 605 public Builder mergeFrom(com.google.protobuf.CodedInputStream input, 606 com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException { 607 PersonProbuf.Person parsedMessage = null; 608 try { 609 parsedMessage = PARSER.parsePartialFrom(input, extensionRegistry); 610 } catch (com.google.protobuf.InvalidProtocolBufferException e) { 611 parsedMessage = (PersonProbuf.Person) e.getUnfinishedMessage(); 612 throw e; 613 } finally { 614 if (parsedMessage != null) { 615 mergeFrom(parsedMessage); 616 } 617 } 618 return this; 619 } 620 621 private int bitField0_; 622 623 // optional int64 id = 1; 624 private long id_; 625 626 /** 627 * <code>optional int64 id = 1;</code> 628 * 629 * <pre> 630 *可選的字段，為64位整數.. 631 * </pre> 632 */ 633 public boolean hasId() { 634 return ((bitField0_ & 0x00000001) == 0x00000001); 635 } 636 637 /** 638 * <code>optional int64 id = 1;</code> 639 * 640 * <pre> 641 *可選的字段，為64位整數.. 642 * </pre> 643 */ 644 public long getId() { 645 return id_; 646 } 647 648 /** 649 * <code>optional int64 id = 1;</code> 650 * 651 * <pre> 652 *可選的字段，為64位整數.. 653 * </pre> 654 */ 655 public Builder setId(long value) { 656 bitField0_ |= 0x00000001; 657 id_ = value; 658 onChanged(); 659 return this; 660 } 661 662 /** 663 * <code>optional int64 id = 1;</code> 664 * 665 * <pre> 666 *可選的字段，為64位整數.. 667 * </pre> 668 */ 669 public Builder clearId() { 670 bitField0_ = (bitField0_ & ~0x00000001); 671 id_ = 0L; 672 onChanged(); 673 return this; 674 } 675 676 // optional string name = 2; 677 private java.lang.Object name_ = ""; 678 679 /** 680 * <code>optional string name = 2;</code> 681 */ 682 public boolean hasName() { 683 return ((bitField0_ & 0x00000002) == 0x00000002); 684 } 685 686 /** 687 * <code>optional string name = 2;</code> 688 */ 689 public java.lang.String getName() { 690 java.lang.Object ref = name_; 691 if (!(ref instanceof java.lang.String)) { 692 java.lang.String s = ((com.google.protobuf.ByteString) ref).toStringUtf8(); 693 name_ = s; 694 return s; 695 } else { 696 return (java.lang.String) ref; 697 } 698 } 699 700 /** 701 * <code>optional string name = 2;</code> 702 */ 703 public com.google.protobuf.ByteString getNameBytes() { 704 java.lang.Object ref = name_; 705 if (ref instanceof String) { 706 com.google.protobuf.ByteString b = com.google.protobuf.ByteString 707 .copyFromUtf8((java.lang.String) ref); 708 name_ = b; 709 return b; 710 } else { 711 return (com.google.protobuf.ByteString) ref; 712 } 713 } 714 715 /** 716 * <code>optional string name = 2;</code> 717 */ 718 public Builder setName(java.lang.String value) { 719 if (value == null) { 720 throw new NullPointerException(); 721 } 722 bitField0_ |= 0x00000002; 723 name_ = value; 724 onChanged(); 725 return this; 726 } 727 728 /** 729 * <code>optional string name = 2;</code> 730 */ 731 public Builder clearName() { 732 bitField0_ = (bitField0_ & ~0x00000002); 733 name_ = getDefaultInstance().getName(); 734 onChanged(); 735 return this; 736 } 737 738 /** 739 * <code>optional string name = 2;</code> 740 */ 741 public Builder setNameBytes(com.google.protobuf.ByteString value) { 742 if (value == null) { 743 throw new NullPointerException(); 744 } 745 bitField0_ |= 0x00000002; 746 name_ = value; 747 onChanged(); 748 return this; 749 } 750 751 // optional string sex = 3; 752 private java.lang.Object sex_ = ""; 753 754 /** 755 * <code>optional string sex = 3;</code> 756 */ 757 public boolean hasSex() { 758 return ((bitField0_ & 0x00000004) == 0x00000004); 759 } 760 761 /** 762 * <code>optional string sex = 3;</code> 763 */ 764 public java.lang.String getSex() { 765 java.lang.Object ref = sex_; 766 if (!(ref instanceof java.lang.String)) { 767 java.lang.String s = ((com.google.protobuf.ByteString) ref).toStringUtf8(); 768 sex_ = s; 769 return s; 770 } else { 771 return (java.lang.String) ref; 772 } 773 } 774 775 /** 776 * <code>optional string sex = 3;</code> 777 */ 778 public com.google.protobuf.ByteString getSexBytes() { 779 java.lang.Object ref = sex_; 780 if (ref instanceof String) { 781 com.google.protobuf.ByteString b = com.google.protobuf.ByteString 782 .copyFromUtf8((java.lang.String) ref); 783 sex_ = b; 784 return b; 785 } else { 786 return (com.google.protobuf.ByteString) ref; 787 } 788 } 789 790 /** 791 * <code>optional string sex = 3;</code> 792 */ 793 public Builder setSex(java.lang.String value) { 794 if (value == null) { 795 throw new NullPointerException(); 796 } 797 bitField0_ |= 0x00000004; 798 sex_ = value; 799 onChanged(); 800 return this; 801 } 802 803 /** 804 * <code>optional string sex = 3;</code> 805 */ 806 public Builder clearSex() { 807 bitField0_ = (bitField0_ & ~0x00000004); 808 sex_ = getDefaultInstance().getSex(); 809 onChanged(); 810 return this; 811 } 812 813 /** 814 * <code>optional string sex = 3;</code> 815 */ 816 public Builder setSexBytes(com.google.protobuf.ByteString value) { 817 if (value == null) { 818 throw new NullPointerException(); 819 } 820 bitField0_ |= 0x00000004; 821 sex_ = value; 822 onChanged(); 823 return this; 824 } 825 826 // optional string tel = 4; 827 private java.lang.Object tel_ = ""; 828 829 /** 830 * <code>optional string tel = 4;</code> 831 */ 832 public boolean hasTel() { 833 return ((bitField0_ & 0x00000008) == 0x00000008); 834 } 835 836 /** 837 * <code>optional string tel = 4;</code> 838 */ 839 public java.lang.String getTel() { 840 java.lang.Object ref = tel_; 841 if (!(ref instanceof java.lang.String)) { 842 java.lang.String s = ((com.google.protobuf.ByteString) ref).toStringUtf8(); 843 tel_ = s; 844 return s; 845 } else { 846 return (java.lang.String) ref; 847 } 848 } 849 850 /** 851 * <code>optional string tel = 4;</code> 852 */ 853 public com.google.protobuf.ByteString getTelBytes() { 854 java.lang.Object ref = tel_; 855 if (ref instanceof String) { 856 com.google.protobuf.ByteString b = com.google.protobuf.ByteString 857 .copyFromUtf8((java.lang.String) ref); 858 tel_ = b; 859 return b; 860 } else { 861 return (com.google.protobuf.ByteString) ref; 862 } 863 } 864 865 /** 866 * <code>optional string tel = 4;</code> 867 */ 868 public Builder setTel(java.lang.String value) { 869 if (value == null) { 870 throw new NullPointerException(); 871 } 872 bitField0_ |= 0x00000008; 873 tel_ = value; 874 onChanged(); 875 return this; 876 } 877 878 /** 879 * <code>optional string tel = 4;</code> 880 */ 881 public Builder clearTel() { 882 bitField0_ = (bitField0_ & ~0x00000008); 883 tel_ = getDefaultInstance().getTel(); 884 onChanged(); 885 return this; 886 } 887 888 /** 889 * <code>optional string tel = 4;</code> 890 */ 891 public Builder setTelBytes(com.google.protobuf.ByteString value) { 892 if (value == null) { 893 throw new NullPointerException(); 894 } 895 bitField0_ |= 0x00000008; 896 tel_ = value; 897 onChanged(); 898 return this; 899 } 900 901 // @@protoc_insertion_point(builder_scope:Person) 902 } 903 904 static { 905 defaultInstance = new Person(true); 906 defaultInstance.initFields(); 907 } 908 909 // @@protoc_insertion_point(class_scope:Person) 910 } 911 912 private static com.google.protobuf.Descriptors.Descriptor internal_static_Person_descriptor; 913 private static com.google.protobuf.GeneratedMessage.FieldAccessorTable internal_static_Person_fieldAccessorTable; 914 915 public static com.google.protobuf.Descriptors.FileDescriptor getDescriptor() { 916 return descriptor; 917 } 918 919 private static com.google.protobuf.Descriptors.FileDescriptor descriptor; 920 static { 921 java.lang.String[] descriptorData = { 922 "\n\006person\"<\n\006Person\022\n\n\002id\030\001 \001(\003\022\014\n\004name\030\002" 923 + " \001(\t\022\013\n\003sex\030\003 \001(\t\022\013\n\003tel\030\004 \001(\tB\016B\014Person" 924 + "Probuf" }; 925 com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner = new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() { 926 public com.google.protobuf.ExtensionRegistry assignDescriptors( 927 com.google.protobuf.Descriptors.FileDescriptor root) { 928 descriptor = root; 929 internal_static_Person_descriptor = getDescriptor().getMessageTypes().get(0); 930 internal_static_Person_fieldAccessorTable = new com.google.protobuf.GeneratedMessage.FieldAccessorTable( 931 internal_static_Person_descriptor, new java.lang.String[] { "Id", "Name", "Sex", "Tel", }); 932 return null; 933 } 934 }; 935 com.google.protobuf.Descriptors.FileDescriptor.internalBuildGeneratedFileFrom(descriptorData, 936 new com.google.protobuf.Descriptors.FileDescriptor[] {}, assigner); 937 } 938 939 // @@protoc_insertion_point(outer_class_scope) 940 } PersonProbuf（自動生成） 1 public class ReqClient { 2 3 public void connect(String host, int port) throws Exception { 4 // 配置服務端的NIO線程組 5 EventLoopGroup group = new NioEventLoopGroup(); 6 7 try { 8 // Bootstrap 類，是啟動NIO服務器的輔助啟動類 9 Bootstrap b = new Bootstrap(); 10 b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true) 11 .handler(new ChannelInitializer<SocketChannel>() { 12 @Override 13 public void initChannel(SocketChannel ch) throws Exception { 14 // 解碼器在讀時才用，編碼器在寫時才用 15 16 // 解碼類 17 ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());// frame解碼得到ByteBuf 18 ch.pipeline().addLast(new ProtobufDecoder(PersonProbuf.Person.getDefaultInstance()));// ByteBuf解碼得到自定義的Protobuf類 19 // 編碼類 20 ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender()); 21 ch.pipeline().addLast(new ProtobufEncoder()); 22 23 ch.pipeline().addLast(new ReqClientHandler()); 24 25 } 26 }); 27 28 // 發起異步連接操作 29 ChannelFuture f = b.connect(host, port).sync(); 30 System.out.println("客戶端啟動."); 31 32 // 等待客服端鏈路關閉 33 f.channel().closeFuture().sync(); 34 } finally { 35 group.shutdownGracefully(); 36 } 37 } 38 39 public static void main(String[] args) throws Exception { 40 int port = 8080; 41 if (args != null && args.length > 0) { 42 try { 43 port = Integer.valueOf(args[0]); 44 } catch (NumberFormatException ex) { 45 } 46 } 47 new ReqClient().connect("127.0.0.1", port); 48 } 49 } ReqClient 1 public class ReqClientHandler extends ChannelHandlerAdapter {// 由于添加了編碼器和解碼器，這里輸出和讀入的都已是自定義的Protobuf類型 2 3 @Override 4 public void channelActive(ChannelHandlerContext ctx) { 5 for (int i = 0; i < 2; i++) { 6 ctx.write(request(i)); 7 } 8 ctx.flush(); 9 } 10 11 private PersonProbuf.Person request(int id) { 12 PersonProbuf.Person.Builder builder = PersonProbuf.Person.newBuilder(); 13 builder.setId(id); 14 builder.setName("orange"); 15 builder.setSex("man"); 16 builder.setTel("999"); 17 18 return builder.build(); 19 } 20 21 @Override 22 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 23 System.out.println("receive server response:[" + msg + "]"); 24 } 25 26 @Override 27 public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { 28 ctx.flush(); 29 } 30 31 @Override 32 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 33 cause.printStackTrace(); 34 ctx.close(); 35 } 36 } ReqClientHandler 1 public class ReqServer { 2 3 public void bind(int port) throws Exception { 4 EventLoopGroup bossGroup = new NioEventLoopGroup(); 5 EventLoopGroup WorkerGroup = new NioEventLoopGroup(); 6 7 try { 8 ServerBootstrap b = new ServerBootstrap(); 9 b.group(bossGroup, WorkerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 1024) 10 .childHandler(new ChannelInitializer<SocketChannel>() { 11 @Override 12 public void initChannel(SocketChannel ch) { 13 // protobufDecoder僅僅負責編碼，并不支持讀半包，所以在之前，一定要有讀半包的處理器。 14 // 有三種方式可以選擇： 15 // 使用netty提供ProtobufVarint32FrameDecoder 16 // 繼承netty提供的通用半包處理器 LengthFieldBasedFrameDecoder 17 // 繼承ByteToMessageDecoder類，自己處理半包 18 19 // 解碼器在讀時才用，編碼器在寫時才用 20 21 // 解碼類 22 ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());// frame解碼得到ByteBuf 23 ch.pipeline().addLast(new ProtobufDecoder(PersonProbuf.Person.getDefaultInstance()));// ByteBuf解碼得到自定義的Protobuf類 24 // 編碼類 25 ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender()); 26 ch.pipeline().addLast(new ProtobufEncoder()); 27 28 ch.pipeline().addLast(new ReqServerHandler()); 29 } 30 }); 31 32 // 綁定端口,同步等待成功 33 ChannelFuture f = b.bind(port).sync(); 34 System.out.println("服務端啟動."); 35 36 // 等待服務端監聽端口關閉 37 f.channel().closeFuture().sync(); 38 } finally { 39 // 釋放線程池資源 40 bossGroup.shutdownGracefully(); 41 WorkerGroup.shutdownGracefully(); 42 } 43 } 44 45 public static void main(String[] args) throws Exception { 46 int port = 8080; 47 if (args != null && args.length > 0) { 48 try { 49 port = Integer.valueOf(args[0]); 50 } catch (NumberFormatException ex) { 51 } 52 } 53 new ReqServer().bind(port); 54 } 55 56 } ReqServer 1 public class ReqServerHandler extends ChannelHandlerAdapter {//由于添加了編碼器和解碼器，這里輸出和讀入的都已是自定義的Protobuf類型 2 @Override 3 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 4 PersonProbuf.Person people = (PersonProbuf.Person) msg; 5 if ("Orange".equalsIgnoreCase(people.getName())) { 6 // if("Orange".equals(people.getName())){ 7 System.out.println("accept client people:[" + people.toString() + "]"); 8 ctx.writeAndFlush(response(people.getId())); 9 } 10 } 11 12 private PersonProbuf.Person response(long peopleID) { 13 PersonProbuf.Person.Builder builder = PersonProbuf.Person.newBuilder(); 14 builder.setId(peopleID); 15 builder.setName("karl"); 16 builder.setSex("boy"); 17 builder.setTel("110"); 18 return builder.build(); 19 } 20 21 @Override 22 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 23 cause.printStackTrace(); 24 ctx.close(); 25 } 26 } ReqServerHandler其中用到了自動生成的 PersonProbuf.java，生成方法如下：
1、從 這里 下載壓縮包并解壓。（生成各種語言的代碼時需要用到相關庫）.
2、從 這里 下載protoc.exe放在上述包的src目錄下（用于根據.proto文件生產對應的java類），不妨把src目錄加入到環境變量。
3、定義消息格式，person.proto（消息格式的定義規則可參看 這里），如下：
1 //option java_package = "Serialization_ProtoBuf.ProtoBuf"; //寫包名的話，會自動創建包名指定的文件夾并把生成的類放在其下，此外，類里屬性或方法的調用會加上全限定名，移動到其他地方時難改，所以最好別加此項。 2 option java_outer_classname = "PersonProbuf"; 3 4 message Person { 5 optional int64 id=1; //可選的字段，為64位整數.. 6 optional string name=2; 7 optional string sex=3; 8 optional string tel=4; 9 } person.proto4、執行 protoc.exe --java_out=java類輸出目錄 proto文件路徑?，就自動生成了對應的類文件Personprobuf.java，文件名由person.proto里的?java_outer_classname?指定（很長。。。）
?
其他：Protobuf生成的類的基本操作（以Personprobuf.java為例）
1 class TestProtobuf { 2 3 public static void main(String[] args) { 4 PersonProbuf.Person.Builder builder = PersonProbuf.Person.newBuilder(); 5 6 builder.setId(1); 7 builder.setName("Karl"); 8 builder.setSex("boy"); 9 builder.setTel("110"); 10 PersonProbuf.Person person = builder.build(); 11 12 System.out.println(person.toString()); 13 14 System.out.println(person.toByteString()); 15 System.out.println(); 16 17 byte[] buf = person.toByteArray(); 18 for (byte b : buf) { 19 System.out.print(b); 20 } 21 System.out.println("\n"); 22 23 try { 24 PersonProbuf.Person person2 = PersonProbuf.Person.parseFrom(buf); 25 System.out.println(person2.getName() + ", " + person2.getTel()); 26 } catch (InvalidProtocolBufferException e) { 27 e.printStackTrace(); 28 } 29 } 30 } 31 32 33 //結果 34 id: 1 35 name: "Karl" 36 sex: "boy" 37 tel: "110" 38 39 <ByteString@31cefde0 size=18> 40 41 81184759711410826398111121343494948 42 43 Karl, 110 TestProtobuf?
2、參考資料
1、http://mangocool.com/1446174360500.html 1.1
2、http://netty.io/wiki/user-guide-for-4.x.html 官方示例
3、http://www.yiibai.com/netty/netty-discard-server.htm 官方示例翻譯
4、https://github.com/orange1438/Netty_Course/tree/master/src/main/java/Serialization_ProtoBuf 結合Protobuf
5、https://blog.csdn.net/dc_726/article/details/47912337 Netty總結
?
轉載于:https://www.cnblogs.com/z-sm/p/6749843.html
《新程序員》：云原生和全面數字化實踐50位技術專家共同創作，文字、視頻、音頻交互閱讀總結
- 上一篇： Windows下通过VNC远程访问Lin
- 下一篇： WebSocket学习与使用