优点：
并发高
传输快
封装好
netty的底层是nio，所以第一阶段我们先了解nio

学习前的准备，NIO学习

传统的bio 阻塞i/o模型，一个线程只能处理一个客户端，客户端没任务时线程阻塞，资源利用率较低，引入线程同样如此
image.png
nio 非阻塞i/o模型，引入三大组件
image.png

三大组件

channel

  1. # 管道 数据输入输出的通道的通道
  2. # 常见的channel有
  3. FileChannel
  4. ServerSocketChannel
  5. SocketChannel
  1. //工作在阻塞模式下
  3. FileChannel channel = new FileInputStream("1.txt").getChannel();
  5. public static void main(String[] args) {
  6.         try {
  7.             FileChannel from =new FileInputStream("1.txt").getChannel();
  8.             FileChannel to = new FileOutputStream("2.txt").getChannel();
  9.             from.transferTo(0, from.size(), to);//一次只能传输2g的文件
  10.             System.out.println("ok");
  11.         } catch (IOException e) {
  12.             // TODO Auto-generated catch block
  13.             e.printStackTrace();
  14.         }
  15.     }

buffer

  1. # 缓冲区 用来协助数据输入与输出
  2. ByteBuffer 
  3. 数据结构
  4. private int position = 0;   //buffer当前读写位置
  5. private int limit;//buffer的读写限制
  6. private int capacity;//buffer的大小

image.png
image.png
image.png

  1. ByteBuffer buffer = ByteBuffer.allocate(12); //使用allocate创建并分配空间，分配堆内存，读写效率高
  2. ByteBuffer buffer = ByteBuffer.allocateDirect(12);//分配的是直接内存，读写效率低，且不受垃圾回收影响
  4. 常用api：
  6. buffer.put();//向缓冲区写入
  8. buffer.get();//从缓冲区获取
  10. buffer.flip();//切换为读模式
  12. buffer.clear();//切换为写模式
  14. buffer.compact();//切换为写模式,并将没读取的字节向前移动
  16. buffer.remaining();//获得buffer中未读的字节数
  18. buffer.hasRemaining();//判断buffer是否全部读取
  20. buffer.rewind();//从头开始读
  23. buffer.mark();//记录position的位置
  24. buffer.reset();//跳转到mark标记的位置
  27. String str = Charset.defaultCharset().decode(buffer).toString();//buffer转换为String
  28. buffer = StandardCharsets.UTF_8.encode(str);//字符串转换为buffer

selector

阻塞模式

  1. server
  3. package feizuse;
  5. import java.io.IOException;
  6. import java.net.InetSocketAddress;
  7. import java.nio.ByteBuffer;
  8. import java.nio.channels.ServerSocketChannel;
  9. import java.nio.channels.SocketChannel;
  10. import java.nio.charset.Charset;
  11. import java.util.ArrayList;
  12. import java.util.List;
  15. public class server {
  17.     public static void main(String[] args) throws IOException {
  18.         //创建服务器
  19.         ServerSocketChannel server = ServerSocketChannel.open();
  20.         //绑定监听端口
  21.         server.bind(new InetSocketAddress(8080));
  22.         //创建数组用来存放客户端channel
  23.         List<SocketChannel> channels = new ArrayList<SocketChannel>();
  24.         server.configureBlocking(false);//ServerSocketChannel改为非阻塞模式
  26.         while(true) {
  27.             System.out.println("accept -------before");
  28.             SocketChannel sc = server.accept(); //阻塞方法
  29.             System.out.println("accept -------after");
  30.             channels.add(sc);
  31.             if(sc != null) {
  32.                 sc.configureBlocking(false);//SocketChannel非阻塞模式
  33.             }
  34.             //读取channel里面的数据
  35.             for(SocketChannel s :channels) {
  36.                 ByteBuffer buffer = ByteBuffer.allocate(16);
  37.                 System.out.println("read -------before");
  38.                 s.read(buffer);//阻塞方法
  39.                 System.out.println("read -------after");
  40.                 buffer.flip();
  41.                 String str = Charset.defaultCharset().decode(buffer).toString();
  42.                 System.out.println(str+s.getLocalAddress());
  43.                 buffer.clear();
  44.             }
  45.         }
  48.     }
  50. }
  53. client
  55. package feizuse;
  57. import java.io.IOException;
  58. import java.net.InetSocketAddress;
  59. import java.nio.channels.SocketChannel;
  60. import java.nio.charset.Charset;
  61. import java.util.Scanner;
  63. public class client1 {
  65.     public static void main(String[] args) throws IOException {
  66.         SocketChannel sc = SocketChannel.open();
  68.         sc.connect(new InetSocketAddress(8080));
  70.         while(true) {
  71.             Scanner scanner =new Scanner(System.in);
  72.             String str = scanner.next();
  74.             sc.write(Charset.defaultCharset().encode(str));
  75.         }
  76.     }
  78. }

selector处理

  1. package feizuse;
  3. import java.io.IOException;
  4. import java.net.InetSocketAddress;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.SelectionKey;
  7. import java.nio.channels.Selector;
  8. import java.nio.channels.ServerSocketChannel;
  9. import java.nio.channels.SocketChannel;
  10. import java.util.Iterator;
  12. public class server2 {
  14.     public static void main(String[] args){
  16.         try {
  17.             //创建一个Selector
  18.             Selector selector =Selector.open();
  20.             //创建服务器
  21.             ServerSocketChannel server = ServerSocketChannel.open();
  22.             server.configureBlocking(false);
  23.             //绑定监听端口
  24.             server.bind(new InetSocketAddress(8080));
  25.             //建立Selector与ServerSocketChannel的连接
  26.             //SelectionKey可以知道是哪个客户端发生的事件
  27.             /**
  28.              * accept --连接时触发
  29.              * connect --连接建立后触发
  30.              * read  --可读事件
  31.              * write  --可写事件
  32.              */
  33.             SelectionKey sskey =  server.register(selector, 0);
  34.             //只关注连接事件
  35.             sskey.interestOps(SelectionKey.OP_ACCEPT);
  36.             while(true) {
  37.                 selector.select();//阻塞方法，没有事件就阻塞
  38.                 Iterator<SelectionKey> iter = selector.selectedKeys().iterator();//获取所有事件的合集
  39.                 //如果是连接事件
  40.                 while(iter.hasNext()) {
  41.                     SelectionKey key =iter.next();
  42.                     iter.remove();
  43.                     if(key.isAcceptable()) {
  44.                         ServerSocketChannel channel =(ServerSocketChannel) key.channel();
  45.                         SocketChannel schannel =  channel.accept();
  46.                         schannel.configureBlocking(false);
  47.                         SelectionKey skey =  schannel.register(selector, 0, null);
  48.                         skey.interestOps(SelectionKey.OP_READ);
  49.                     }else if(key.isReadable()) {
  50.                         try {
  51.                             /**
  52.                              * 客户端断开连接会添加一个read事件
  53.                              */
  54.                             SocketChannel channel=(SocketChannel) key.channel();
  55.                             ByteBuffer buffer =ByteBuffer.allocate(16);
  56.                             int iscancel =channel.read(buffer);//正常情况下返回传输的字节，正常推出返回-1
  57.                             if(iscancel==-1) {
  58.                                 key.cancel();
  59.                                 System.out.println("正常推出");
  60.                             }
  61.                         }catch(IOException e) {
  62.                             e.printStackTrace();
  63.                             //异常退出
  64.                             key.cancel();
  65.                         }
  67.                     }
  68.                 }
  69.             }
  70.         }catch(IOException e) {
  71.             e.printStackTrace();
  72.         }
  73.     }
  75. }

多线程优化

  1. package feizuse;
  3. import java.io.IOException;
  4. import java.net.InetSocketAddress;
  5. import java.nio.channels.SelectionKey;
  6. import java.nio.channels.Selector;
  7. import java.nio.channels.ServerSocketChannel;
  8. import java.nio.channels.SocketChannel;
  9. import java.util.Iterator;
  11. public class boss {
  13.     public static void main(String[] args) {
  14.         //当前是boos线程
  15.         Thread.currentThread().setName("boos");
  16.         try {
  17.             //建立服务端
  18.             ServerSocketChannel ssc = ServerSocketChannel.open();
  19.             //绑定接口
  20.             ssc.bind(new InetSocketAddress(8080));
  21.             ssc.configureBlocking(false);//开启非阻塞模式
  23.             Selector selector =Selector.open();
  24.             SelectionKey sskey =  ssc.register(selector,SelectionKey.OP_ACCEPT);//注册selector并关注连接事件
  25.             //创建指定数量的worker
  26.             worker w1 = new worker("w1");
  27.             while(true) {
  28.                 selector.select();//阻塞方法等待事件
  29.                 Iterator<SelectionKey> iter = selector.selectedKeys().iterator();//获取所有事件的合集
  30.                 while(iter.hasNext()) {
  31.                     SelectionKey key = iter.next();
  32.                     iter.remove();
  33.                     if(key.isAcceptable()) {
  34.                         //如果是连接事件
  35.                         ServerSocketChannel channel =(ServerSocketChannel) key.channel();
  36.                         SocketChannel schannel =  channel.accept();
  37.                         //给该channel注册key
  38.                         schannel.configureBlocking(false);
  39.                         //注册w1的selector并关注读事件
  40.                         w1.regesit(schannel);
  41.                     }
  42.                 }
  43.             }
  45.         } catch (IOException e) {
  46.             // TODO Auto-generated catch block
  47.             e.printStackTrace();
  48.         }
  49.     }
  50. }
  59. package feizuse;
  61. import java.io.IOException;
  62. import java.nio.ByteBuffer;
  63. import java.nio.channels.ClosedChannelException;
  64. import java.nio.channels.SelectionKey;
  65. import java.nio.channels.Selector;
  66. import java.nio.channels.SocketChannel;
  67. import java.nio.charset.Charset;
  68. import java.util.Iterator;
  69. import java.util.concurrent.ConcurrentLinkedQueue;
  71. public class worker implements Runnable{
  72.     //处理读事件
  73.     private Thread t1;//当前线程
  74.     private Selector selector;//selector    
  75.     private ConcurrentLinkedQueue<Runnable> queue =new ConcurrentLinkedQueue<Runnable>();
  77.     worker(String name){
  78.         this.t1 =new Thread(this,name);
  79.         try {
  80.             this.selector =Selector.open();
  81.         } catch (IOException e) {
  82.             // TODO Auto-generated catch block
  83.             e.printStackTrace();
  84.         }
  85.         t1.start();
  86.     }
  88.     void regesit(SocketChannel s) {
  89.         System.out.println(Thread.currentThread().getName()+"注册");
  90.         queue.add(()->{
  91.             try {
  92.                 s.register(selector, SelectionKey.OP_READ);
  93.             } catch (ClosedChannelException e) {
  94.                 // TODO Auto-generated catch block
  95.                 e.printStackTrace();
  96.             }
  97.         });
  98.         selector.wakeup();
  100.     }
  102.     @Override
  103.     public void run() {
  104.         while(true) {
  105.             try {
  106.                 System.out.println(Thread.currentThread().getName()+"run");
  107.                 System.out.println("阻塞");
  108.                 selector.select();
  109.                 Runnable task =  queue.poll();
  110.                 if(task!=null) {
  111.                     System.out.println(Thread.currentThread().getName()+"任务");
  112.                     System.out.println("执行任务");
  113.                     task.run();
  114.                 }
  115.                 Iterator<SelectionKey> iter =  selector.selectedKeys().iterator();
  116.                 while(iter.hasNext()) {
  117.                     SelectionKey key =iter.next();
  118.                     iter.remove();
  119.                     if(key.isReadable()) {
  120.                         System.out.println("有读任务");
  121.                         try {
  122.                             SocketChannel channel = (SocketChannel) key.channel();
  123.                             ByteBuffer buffer=ByteBuffer.allocate(16);
  124.                             int isclose = channel.read(buffer);
  125.                             if(isclose ==-1) {
  126.                                 key.cancel();
  127.                             }
  128.                             System.out.println(buffer.position());
  129.                         }catch(IOException e) {
  130.                             key.cancel();
  131.                         }
  133.                     }
  134.                 }
  135.             } catch (IOException e) {
  136.                 // TODO Auto-generated catch block
  137.                 e.printStackTrace();
  138.             }
  139.         }
  141.     }
  142. }

Netty入门

Netty是什么

netty是异步的，事件驱动的网络编程框架

image.png

服务器

  1. package com.netty;
  3. import io.netty.bootstrap.ServerBootstrap;
  4. import io.netty.channel.ChannelInitializer;
  5. import io.netty.channel.nio.NioEventLoopGroup;
  6. import io.netty.channel.socket.nio.NioServerSocketChannel;
  7. import io.netty.channel.socket.nio.NioSocketChannel;
  8. import io.netty.handler.codec.string.StringDecoder;
  10. public class server01 {
  13.     public static void main(String[] args) {
  14.         //服务器启动器
  15.         new ServerBootstrap()
  16.                 .group(new NioEventLoopGroup())
  17.                 .channel(NioServerSocketChannel.class)
  18.                 .childHandler(new ChannelInitializer<NioSocketChannel>() {
  19.                     @Override
  20.                     protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
  21.                         nioSocketChannel.pipeline().addLast(new StringDecoder());
  22.                     }
  23.                 }).bind(8080);
  24.     }
  25. }

客户端

  1. package com.netty;
  3. import io.netty.bootstrap.Bootstrap;
  4. import io.netty.channel.ChannelInitializer;
  5. import io.netty.channel.nio.NioEventLoopGroup;
  6. import io.netty.channel.socket.nio.NioSocketChannel;
  7. import io.netty.handler.codec.string.StringDecoder;
  9. import java.net.InetSocketAddress;
  11. public class client01 {
  12.     public static void main(String[] args) throws InterruptedException {
  13.         new Bootstrap()
  14.                 .group(new NioEventLoopGroup())
  15.                 .channel(NioSocketChannel.class)
  16.                 .handler(new ChannelInitializer<NioSocketChannel>() {
  17.                     @Override
  18.                     protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
  19.                         nioSocketChannel.pipeline().addLast(new StringDecoder());
  20.                     }
  21.                 }).connect(new InetSocketAddress(8080))//连接服务器
  22.                 .sync()
  23.                 .channel()
  24.                 .writeAndFlush("nihao");
  25.     }
  26. }

组件

EventLoop

  1. #EventLoop是处理事件的类，他包含selector和thread
  4. package com.netty;
  7. import io.netty.bootstrap.ServerBootstrap;
  8. import io.netty.buffer.ByteBuf;
  9. import io.netty.channel.*;
  10. import io.netty.channel.nio.NioEventLoopGroup;
  11. import io.netty.channel.socket.nio.NioServerSocketChannel;
  12. import io.netty.channel.socket.nio.NioSocketChannel;
  14. import java.nio.charset.Charset;
  17. public class server01 {
  19.     public static void main(String[] args) {
  20.         //服务器启动器,用来装配组件
  21.         //创建一个独立的eventgroup
  22.         EventLoopGroup group = new DefaultEventLoopGroup(1);
  23.         new ServerBootstrap()
  24.                 //装配eventloop组，
  25.                 .group(new NioEventLoopGroup(1),new NioEventLoopGroup(2))
  26.                 //channel具体实现
  27.                 .channel(NioServerSocketChannel.class)
  28.                 //eventloop处理逻辑
  29.                 .childHandler(new ChannelInitializer<NioSocketChannel>() {
  30.                     @Override
  31.                     protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
  32.                         nioSocketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter(){
  33.                             @Override
  34.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  35.                                 ByteBuf buf = (ByteBuf) msg;
  36.                                 System.out.println(buf.toString(Charset.defaultCharset()));
  37.                                 System.out.println(Thread.currentThread().getName());
  38.                                 //多个hander需要将消息向下传递
  39.                                 ctx.fireChannelRead(msg);
  40.                             }
  41.                         }).addLast(group,new ChannelInboundHandlerAdapter(){
  42.                             @Override
  43.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  44.                                 ByteBuf buf = (ByteBuf) msg;
  45.                                 System.out.println(Thread.currentThread().getName());
  46.                                 System.out.println(buf.toString(Charset.defaultCharset())+"2");
  47.                             }
  48.                         });
  49.                     }
  50.                 })
  51.                 .bind(8080);
  52.     }
  53. }

channel

  1. //关闭连接
  2. channel.close();
  3. //写入内存但不立即刷出
  4. channel.write();
  5. //将内存的数据刷出
  6. channel.flush();
  7. //写入内存并立即刷出
  8. channel.writeAndFlush("1231");
  12. //用来处理异步处理的结果的类
  13. ChannelFuture
  14. //connect是异步非阻塞方法，调用该方法的线程会继续向下执行代码，连接过程交给另外一个线程处理
  15. 如果不调用sync，连接的线程并没有执行完毕，也就是不存在channel，会出错，主线程需要等待连接建立成功后才能向下运行
  16. .connect(new InetSocketAddress("localhost", 8080));
  17. //连接服务器
  18. 处理的两种方式
  19. 1：sync（）是让主线程阻塞，等待nio线程执行完建立连接的过程返回连接对象ChannelFuture后在向下执行
  20. //sync（）是main线程处理，等待结果返回后自己向下继续处理
  21. channelFuture.sync();
  22. Channel channel = channelFuture.channel();
  23. channel.writeAndFlush();
  25. 2：另外一个处理方式，main调用connect后自己不等待结果，让nio线程自己处理
  26. nio建立连接之后，会继续调用operationComplete方法，并执行里面的代码
  27. channelFuture.addListener(new ChannelFutureListener() {
  28.     @Override
  29.     public void operationComplete(ChannelFuture channelFuture) throws Exception {
  30.         Channel channel = channelFuture.channel();
  31.         channel.writeAndFlush();
  32.     }
  33. });
  37. 处理断开连接的善后操作
  38. close是异步操作，所以和connect一样，我们有两种处理办法。
  39. channel.close();//异步操作
  41. //同步处理办法
  42. ChannelFuture close = channel.closeFuture();
  43. close.sync();//阻塞当前线程，等待连接关闭后主线程进行善后处理
  45. //异步处理办法，关闭连接的线程关闭连接后执行该方法
  46. close.addListener(new ChannelFutureListener() {
  47.             @Override
  48.             public void operationComplete(ChannelFuture channelFuture) throws Exception {
  49.                 System.out.println("善后处理");
  50.             }
  51.         });
  • 关闭eventloop

    group.shutdownGracefully();//停止eventloop

Future

  • jdk future ```java package com.netty;

import java.util.concurrent.*;

public class future01 {

  1. public static void main(String[] args) throws ExecutionException, InterruptedException {
  2.     /**
  3.      * jdk版本的future
  4.      */
  5.     //创建线程池
  6.     ExecutorService service = Executors.newFixedThreadPool(2);
  7.     Future<Integer> future = service.submit(new Callable<Integer>() {
  8.         //提交任务
  9.         @Override
  10.         public Integer call() throws Exception {
  11.             Thread.sleep(1000);
  12.             return 70;
  13.         }
  14.     });
  15.     //主线程接收结果
  16.     System.out.println(future.get());
  17. }

}

  2. - netty future
  3. ```java
  4.         /**
  5.          * netty 版本的future
  6.          */
  8.         NioEventLoopGroup group = new NioEventLoopGroup(2);
  9.         EventLoop next = group.next();
  10.         Future<Integer> future = next.submit(new Callable<Integer>() {
  12.             @Override
  13.             public Integer call() throws Exception {
  14.                 Thread.sleep(1000);
  15.                 return 70;
  16.             }
  17.         });
  18.         //netty有同步和异步两种
  19.         System.out.println(future.get());
  21.         //第二种异步
  22.         future.addListener(new GenericFutureListener<Future<? super Integer>>() {
  23.             @Override
  24.             public void operationComplete(Future<? super Integer> future) throws Exception {
  25.                 System.out.println(future.getNow());
  26.             }
  27.         });
  • promise ```java package com.netty;

import io.netty.channel.EventLoop; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.util.concurrent.DefaultPromise;

import java.util.concurrent.ExecutionException;

public class promise { public static void main(String[] args) throws ExecutionException, InterruptedException { NioEventLoopGroup group =new NioEventLoopGroup(2); EventLoop next = group.next(); DefaultPromise promise = new DefaultPromise<>(next);

  1.     new Thread(()->{
  2.         try {
  3.             Thread.sleep(1000);
  4.         } catch (InterruptedException e) {
  5.             e.printStackTrace();
  6.             //失败了添加异常
  7.             promise.setFailure(e);
  8.         }
  9.         promise.setSuccess(80);
  10.     }).start();
  11.     System.out.println(promise.get());
  12. }

}

  1. <a name="DaUl9"></a>
  2. #### popeline和handler
  3. ```java
  4. package com.netty;
  6. import io.netty.bootstrap.ServerBootstrap;
  7. import io.netty.channel.*;
  8. import io.netty.channel.nio.NioEventLoopGroup;
  9. import io.netty.channel.socket.SocketChannel;
  10. import io.netty.channel.socket.nio.NioServerSocketChannel;
  11. import io.netty.channel.socket.nio.NioSocketChannel;
  13. public class hander {
  15.     public static void main(String[] args) {
  16.         new ServerBootstrap()
  17.                 .group(new NioEventLoopGroup(1),new NioEventLoopGroup(2))
  18.                 .channel(NioServerSocketChannel.class)
  19.                 .childHandler(new ChannelInitializer<NioSocketChannel>() {
  21.                     @Override
  22.                     protected void initChannel(NioSocketChannel ch) throws Exception {
  23.                         ChannelPipeline pipeline = ch.pipeline();
  25.                         //向流水线添加处理器
  26.                         //head -> h1->h2->h3->h4->h5->tail
  28.                         pipeline.addLast(new ChannelInboundHandlerAdapter(){
  29.                             @Override
  30.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  31.                                 System.out.println("h1");
  32.                                 super.channelRead(ctx, msg);
  33.                             }
  34.                         });
  36.                         pipeline.addLast(new ChannelInboundHandlerAdapter(){
  38.                             @Override
  39.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  40.                                 System.out.println("h2");
  41.                                 ctx.fireChannelRead(msg);
  42.                             }
  43.                         });
  44.                         pipeline.addLast(new ChannelInboundHandlerAdapter(){
  46.                             @Override
  47.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  48.                                 System.out.println("h3");
  49.                                 //服务器发送数据才会触发出站处理器
  50.                                 //ch的writeAndFlush是从tail从后向前执行出站处理器
  51.                                 ch.writeAndFlush(ctx.alloc().buffer().writeBytes("nishi".getBytes()));
  52.                                 //ctx的writeAndFlush是从当前处理器向前寻找出站处理器
  53.                                 //ctx.writeAndFlush(ctx.alloc().buffer().writeBytes("132".getBytes()));
  54.                             }
  55.                         });
  56.                         pipeline.addLast(new ChannelOutboundHandlerAdapter(){
  58.                             @Override
  59.                             public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
  60.                                 System.out.println("h4");
  61.                                 super.write(ctx, msg, promise);
  62.                             }
  63.                         });
  65.                         pipeline.addLast(new ChannelOutboundHandlerAdapter(){
  67.                             @Override
  68.                             public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
  69.                                 System.out.println("h5");
  70.                                 super.write(ctx, msg, promise);
  71.                             }
  72.                         });
  73.                     }
  74.                 }).bind(8080);
  75.     }
  76. }

image.png这里使用的ch的writeandflush
image.png这里使用的ctx的writeandflush，因为h3之前没有出站处理器，所以没有打印结果

handler的调试工具类

  1. package com.netty;
  3. import io.netty.buffer.ByteBufAllocator;
  4. import io.netty.channel.embedded.EmbeddedChannel;
  6. public class testhandler {
  8.     public static void main(String[] args) {
  9.         /**
  10.          * 假设以上有n个入站处理器和n个出站处理器
  11.          */
  12.         EmbeddedChannel channel = new EmbeddedChannel();
  13.         //模拟入站操作
  14.         channel.writeInbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("1231".getBytes()));
  15.         //模拟出站操作
  16.         channel.writeOutbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("1231".getBytes()));
  17.     }
  18. }

Bytebuf

  1. package com.netty;
  4. import io.netty.buffer.ByteBuf;
  5. import io.netty.buffer.ByteBufAllocator;
  7. public class buf {
  9.     public static void main(String[] args) {
  10.         //默认大小是256，也可以自己设置，netty的bytebuf可以自动扩容
  11.         ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
  12.         buffer.writeBytes("1213".getBytes());
  14.         //支持两种内存分配，堆内存和直接内存（读写效率高）
  15.         ByteBuf buf1 = ByteBufAllocator.DEFAULT.heapBuffer();//堆内存，垃圾回收
  16.         ByteBuf buf2 = ByteBufAllocator.DEFAULT.directBuffer();//直接内存，建议手动释放
  18.         //池化
  19.         //4.1之后windows平台默认开启池化
  20.     }
  21. }

组成 image.png bytebuf有两个指针分别用于读和写，而且自动动态扩容。 扩容规则： 未超过512，按16的整数倍扩容 超过512，按照 2 的n次方扩容，512是2^9

写入

  1. 可以写入所有类型
  2. writeIntLE()  //大端写入，一般采用
  3. writeInt()  //小端写入

读取

  1. buffer.readByte();
  2. buffer.markReaderIndex();//标记都指针位置
  3. buffer.resetReaderIndex();//回到标记位置

内存回收
零拷贝

  1. //物理内存一致，但是有独立的指针
  3. package com.netty;
  6. import io.netty.buffer.ByteBuf;
  7. import io.netty.buffer.ByteBufAllocator;
  9. import java.nio.charset.Charset;
  10. import java.nio.charset.StandardCharsets;
  12. public class buf {
  14.     public static void main(String[] args) {
  15.         //默认大小是256，也可以自己设置，netty的bytebuf可以自动扩容
  16.         ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
  18.         buffer.writeBytes("123456789\n123132145465\n".getBytes(StandardCharsets.UTF_8));
  20.         ByteBuf buf = buffer.slice(0,9);
  21.         System.out.println(buf.toString(Charset.defaultCharset()));
  22.     }
  23. }
  24. //内存释放使用的是计数
  25. buf.release();//计数减1
  26. buf.retain();//计数加1
  28. buf.duplicate();//零拷贝，拷贝全部
  29. buf.copy();//数据复制，拷贝的和原来的不是一块内存
  31. 小的bytebuf整合为大的，零拷贝
  32. CompositeByteBuf byteBufs = ByteBufAllocator.DEFAULT.compositeBuffer();
  33. byteBufs.addComponent(true,buf1,buf2);
  35. 在切片的过程中没有发生数据的赋值，两个buf使用的同一块内存的数据
  36. 切片后会对新产生的bytebuf的容量加以限制，不能新增，只能修改

工具类unpooled

  1. 提供了非池化bytebuf的创建、组合复制
  2. Unpooled.wrappedBuffer(buffer,buf);

总结

bytebuf的优势：

  1. 使用池化技术，可以重复利用池中的bytebuf，节省内存
  2. 读写独立
  3. 可以自动扩容
  4. 可以零拷贝，减少内存赋值

Netty进阶

粘包和半包

粘包： 一次发送多个数据包 产生原因： 接收方的缓存设置过大 滑动窗口缓存了多个报文 半包： 没有接收到一个完整的数据包，数据不完整 产生原因： MSS限制 接收方缓存设置太小，一次存放不下全部报文 滑动窗口无法发送全部

粘包半包解决方案

  1. 短连接

解决粘包问题，不能解决半包问题

  1. 定长解码器

能够正确解决粘包半包问题，但是浪费内存
image.png

  1. //FixedLengthFrameDecoder
  2. //server
  3. public class ChildHandler extends ChannelInitializer<SocketChannel> {
  4.     @Override
  5.     protected void initChannel(SocketChannel socketChannel) throws Exception {
  6.         Logger logger = LoggerFactory.getLogger(ChildHandler.class);
  7.         ChannelPipeline pipeline = socketChannel.pipeline();
  8.         pipeline.addLast(new FixedLengthFrameDecoder(20));        //从channel中每次读取20字节
  9.         pipeline.addLast(new StringDecoder());
  10.         pipeline.addLast(new EchoServerHandler());//打印服务器handler
  11.         pipeline.addLast(new StringEncoder());
  13.     }
  14. }
  1. 行解码器

    A decoder that splits the received ByteBufs on line endings.Both “\n” and “\r\n” are handled. The byte stream is expected to be in UTF-8 character encoding or ASCII. The current implementation uses direct byte to char cast and then compares that char to a few low range ASCII characters like ‘\n’ or ‘\r’. UTF-8 is not using low range [0..0x7F] byte values for multibyte codepoint representations therefore fully supported by this implementation. For a more general delimiter-based decoder, see DelimiterBasedFrameDecoder.

  1. //LineBasedFrameDecoder
  2. public class ChildHandler extends ChannelInitializer<SocketChannel> {
  3.     @Override
  4.     protected void initChannel(SocketChannel socketChannel) throws Exception {
  5.         Logger logger = LoggerFactory.getLogger(ChildHandler.class);
  6.         ChannelPipeline pipeline = socketChannel.pipeline();
  7.         pipeline.addLast(new LineBasedFrameDecoder(1024));    //行解码器将读到数据按照\n划分，如果在最大长度（1024）仍未读到换行符，则抛出异常。
  8.         pipeline.addLast(new StringDecoder());
  9.         //pipeline.addLast(new TimeServerHandler());
  10.         pipeline.addLast(new EchoServerHandler());
  11.         pipeline.addLast(new StringEncoder());
  13.     }
  14. }
  1. LTC解码器

    lengthFieldOffset = 0 //长度字段偏移量 lengthFieldLength = 2 //长度字段长度 lengthAdjustment = 0 //长度字段之后还要隔几个字节才是内容 initialBytesToStrip = 0 //从头剥离几个字节 image.png image.png

  1. LengthFieldBasedFrameDecoder

  1. 分隔符解码器

    1. public class ChildHandler extends ChannelInitializer<SocketChannel> {
    2.  @Override
    3.  protected void initChannel(SocketChannel socketChannel) throws Exception {
    4.      Logger logger = LoggerFactory.getLogger(ChildHandler.class);
    5.      ChannelPipeline pipeline = socketChannel.pipeline();
    6.      ByteBuf byteBuf = Unpooled.copiedBuffer("$_".getBytes());
    7.      pipeline.addLast(new DelimiterBasedFrameDecoder(1024,byteBuf));//分隔符解码器，按照指定分隔符划分。并且划分之后数据不含分隔符。
    8.      pipeline.addLast(new StringDecoder());
    9.      //pipeline.addLast(new TimeServerHandler());
    10.      pipeline.addLast(new EchoServerHandler());
    11.      pipeline.addLast(new StringEncoder());
    13.  }
    14. }

    序列化

    Java序列化的性能较低，并且无法跨语言，序列化效率低并且文件过大。 如今业界主流的编解码框架有 谷歌的 protobuf，脸书的thrift以及使用并不广泛的JBoss Marshalling。

netty Java序列化

  1. //server
  2. public class ChildHandler extends ChannelInitializer<SocketChannel> {
  3.     @Override
  4.     protected void initChannel(SocketChannel socketChannel) throws Exception {
  5.         Logger logger = LoggerFactory.getLogger(ChildHandler.class);
  6.         ChannelPipeline pipeline = socketChannel.pipeline();
  7.         pipeline.addLast(new ObjectDecoder(1024*1024, ClassResolvers.weakCachingResolver(
  8.                 this.getClass().getClassLoader()
  9.         )));
  10.         pipeline.addLast(new ObjectEncoder());
  11.         pipeline.addLast(new GoodsServerHandler());
  13.     }
  14. }
  15. //client
  16. public class NettyClient {
  17.     public static void main(String[] args) throws InterruptedException {
  18.         ChannelFuture future = new Bootstrap()
  19.                 .group(new NioEventLoopGroup())
  20.                 .channel(NioSocketChannel.class)
  21.                 .handler(new ChannelInitializer<NioSocketChannel>() {
  22.                     @Override
  23.                     protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
  25.                         ChannelPipeline pipeline = nioSocketChannel.pipeline();
  26.                         pipeline.addLast(new ObjectDecoder(1024*1024, ClassResolvers.cacheDisabled(this.getClass().getClassLoader())));
  27.                         pipeline.addLast(new ObjectEncoder());
  28.                         pipeline.addLast(new ChannelInboundHandlerAdapter() {
  29.                             @Override
  30.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  31.                                 System.out.println(msg);
  32.                             }
  33.                             @Override
  34.                             public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  35.                                 cause.printStackTrace();
  36.                             }
  37.                         });
  38.                     }
  39.                 }).connect(new InetSocketAddress(8080))//连接服务器
  40.                 .sync();
  42.         Channel channel = future.channel();
  43.         for (int i = 0; i < 10; i++) {
  44.             UserOrder userOrder = new UserOrder(150322154+i,"zhangsan"+i,"手机");
  45.             channel.writeAndFlush(userOrder);
  46.         }
  48.     }
  49. }

Protobuf 基本使用

使用protobuf，我们不能使用自己创建的类，而是需要编写扩展名为.proto的配置文件。然后使用工具生成类。

  1. syntax ="proto3";        //什么版本就写什么
  2. package netty;        //默认包名
  3. option java_package="项目存放类的全包名如com.netty02.bean";        //如果不一致，会报错，还要自己改包名很麻烦。
  4. option java_outer_classname = "生成的类的类名";        //如果我们要使用的bean类名叫xxx，这里就写xxxProto，因为生成的类是 xxxProto，该类里面有个内部类叫 xxx
  6. //message定义类属性
  7. message xxx{
  8. int32 reqId =0;
  9. string name =2;
  10. string address =3;
  11. }
  13. //最后需要注意的是v2和v3有改动，具体请自行查阅。

然后就是使用工具生成类了。
在bin目录下有可执行文件，我这里在当前目录下创建新文件，会车之后，当前目录下就会出现结构和java_package一样的目录，在里面就能找到java文件，我们复制进项目即可。
image.png

  1. //server
  3. public class ChildHandler extends ChannelInitializer<SocketChannel> {
  4.     @Override
  5.     protected void initChannel(SocketChannel socketChannel) throws Exception {
  6.         Logger logger = LoggerFactory.getLogger(ChildHandler.class);
  7.         ChannelPipeline pipeline = socketChannel.pipeline();
  8.         //解码
  9.         pipeline.addLast(new ProtobufVarint32FrameDecoder());
  10.         pipeline.addLast(new ProtobufDecoder(SubBeanProto.SubBean.getDefaultInstance()));    //这里面就是我们要解码的类的类型。
  13.         pipeline.addLast(new GoodsServerHandler());        //自定义处理器
  16.         pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
  17.         pipeline.addLast(new ProtobufEncoder());        //编码
  19.     }
  20. }
  21. //client
  22. public static void main(String[] args) throws InterruptedException {
  23.         ChannelFuture future = new Bootstrap()
  24.                 .group(new NioEventLoopGroup())
  25.                 .channel(NioSocketChannel.class)
  26.                 .handler(new ChannelInitializer<NioSocketChannel>() {
  27.                     @Override
  28.                     protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
  29.                         ChannelPipeline pipeline = nioSocketChannel.pipeline();
  31.                         pipeline.addLast(new ProtobufVarint32FrameDecoder());
  32.                         pipeline.addLast(new ProtobufDecoder(SubRespProto.SubResp.getDefaultInstance()));
  33.                         pipeline.addLast(new ChannelInboundHandlerAdapter() {
  34.                             @Override
  35.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  36.                                 SubRespProto.SubResp resp = (SubRespProto.SubResp) msg;
  37.                                 System.out.println(resp.toString());
  38.                             }
  39.                             @Override
  40.                             public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  41.                                 cause.printStackTrace();
  42.                             }
  43.                         });
  44.                         pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
  45.                         pipeline.addLast(new ProtobufEncoder());
  46.                     }
  47.                 }).connect(new InetSocketAddress(8080))//连接服务器
  48.                 .sync();
  50.         Channel channel = future.channel();
  51.         for (int i = 0; i < 10; i++) {
  52.             //使用protobuf
  53.             //这里构建对象使用builder，获取builder方法如下
  54.             SubBeanProto.SubBean.Builder builder =SubBeanProto.SubBean.newBuilder();
  55.             builder.setSubReqId(i);
  56.             builder.setName("zhangsan"+i);
  57.             builder.setProductName("手机");
  58.             channel.writeAndFlush(builder.build());
  59.         }
  61.     }

上面有两个netty支持protobuf的编解码器，ProtobufVarint32FrameDecoder和ProtobufVarint32LengthFieldPrepender，这两个编解码器请看源码。
源码分析博客

protobuf只能处理ascall码，汉字会根据系统设置编码，使用toString方法，不能将汉字解码，建议单独正确解码后打印。

结果
image.png

协议

Redis通信协议

  1. 例：
  2.     set key value
  3.    1.语句长度 *3
  4.    2.命令长度（set） $3
  5.    3.命令 set
  6.    4.长度（key） $4
  7.    5.值   name
  8.    6.长度  （value） $8
  9.    7.值      zhangsan

Http通信协议

  1. package com.netty;
  3. import io.netty.bootstrap.ServerBootstrap;
  4. import io.netty.channel.*;
  5. import io.netty.channel.nio.NioEventLoopGroup;
  6. import io.netty.channel.socket.nio.NioServerSocketChannel;
  7. import io.netty.channel.socket.nio.NioSocketChannel;
  8. import io.netty.handler.codec.http.DefaultFullHttpResponse;
  9. import io.netty.handler.codec.http.HttpRequest;
  10. import io.netty.handler.codec.http.HttpResponseStatus;
  11. import io.netty.handler.codec.http.HttpServerCodec;
  13. import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH;
  16. public class httpserve {
  17.     public static void main(String[] args) throws InterruptedException {
  18.         ServerBootstrap bootstrap = new ServerBootstrap()
  19.                 .group(new NioEventLoopGroup())
  20.                 .channel(NioServerSocketChannel.class)
  21.                 .childHandler(new ChannelInitializer<NioSocketChannel>() {
  23.                     @Override
  24.                     protected void initChannel(NioSocketChannel ch) throws Exception {
  25.                         ChannelPipeline pipeline = ch.pipeline();
  26.                         pipeline.addLast(new HttpServerCodec());
  27.                         pipeline.addLast(new SimpleChannelInboundHandler<HttpRequest>(){
  29.                             @Override
  30.                             protected void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) throws Exception {
  31.                                 System.out.println(msg);
  32.                                 DefaultFullHttpResponse res =new DefaultFullHttpResponse(msg.getProtocolVersion(), HttpResponseStatus.OK);
  33.                                 byte[] bytes = "nihao".getBytes();
  34.                                 res.headers().setInt(CONTENT_LENGTH,bytes.length);
  35.                                 res.content().writeBytes(bytes);
  36.                                 ctx.writeAndFlush(res);
  37.                             }
  39.                         });
  40.                     }
  41.                 });
  42.         ChannelFuture bind = bootstrap.bind(8080);
  43.         bind.sync();
  44.     }
  45. }

自定义协议要素

  1. 魔数：

用来在第一时间判断是否是无效数据包

  1. 版本号：

可支持协议的升级

  1. 序列化算法

消息正文到底采用哪种序列化反序列化方式

  1. 指令类型

是登录注册还是其他业务

  1. 请求序号

为了双工通信，提供异步功能

  1. 正文长度
  2. 消息正文

编码解码

优化

连接超时

  1. .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,5000)