Netty的基本使用

@Test
public void testByteBuffer() throws Exception {
  RandomAccessFile file = new RandomAccessFile("data.txt", "rw");
  FileChannel channel = file.getChannel();
  ByteBuffer buffer = ByteBuffer.allocate(10);
  int readLen = 0;
  do {
    // 向 buffer 写入
    readLen = channel.read(buffer);
    log.info("读取到字节 {}",readLen);
    // 切换 buffer 读模式
    buffer.flip();
    while (buffer.hasRemaining()) {
      log.info("{}", (char) buffer.get());
    }
    // 切换 buffer 写模式
    buffer.clear();
  } while (readLen != -1);
}

// heap 内存
ByteBuffer buf = ByteBuffer.allocate(16);
// 直接内存
ByteBuffer directBuf = ByteBuffer.allocateDirect(16);

int readBytes = channel.read(buf); //返回从channel读取的长度

buf.put((byte)127);                //手动放入数据

int writeBytes = channel.write(buf);  // 返回写入到channle的自己数量

byte b = buf.get();                   // 手动读取数据，会移动position指针
byte b = buf.get(i);                  // 手动读取数据，不移动position指针

buf.rewind();                         // 重置position为0

/**
 * mark & reset
 * mark在读取时，做一个标记。即使position改变，只要调用reset就能回到mark的位置
 */
@Test
public void markAndRestTest() {
  ByteBuffer buffer = ByteBuffer.allocate(10);
  buffer.put("123456".getBytes());
  buffer.flip();

  byte b1 = buffer.get(); // '1'

  buffer.mark();

  byte b2 = buffer.get(); // '2'
  byte b3 = buffer.get(); // '3'

  buffer.reset();

  byte hasB2 = buffer.get(); //'2'
  assert hasB2 == b2;
}

// 字符串转ByteBuffer
ByteBuffer buffer1 = StandardCharsets.UTF_8.encode("你好");
debug(buffer1);

// ByteBuffer转字符串
CharBuffer buffer2 = StandardCharsets.UTF_8.decode(buffer1);
System.out.println(buffer2.toString());

// 分散读取，有一个文本文件 3parts.txt 内容 onetwothree
@Test
public void partsReadTest() {
  try (RandomAccessFile file = new RandomAccessFile("3parts.txt", "rw")) {
    FileChannel channel = file.getChannel();
    ByteBuffer a = ByteBuffer.allocate(3);
    ByteBuffer b = ByteBuffer.allocate(3);
    ByteBuffer c = ByteBuffer.allocate(5);
    channel.read(new ByteBuffer[]{a, b, c});
    a.flip();
    b.flip();
    c.flip();
    ByteBufferUtil.debugAll(a);	//one
    ByteBufferUtil.debugAll(b); //two
    ByteBufferUtil.debugAll(c); //three
  } catch (IOException e) {
    e.printStackTrace();
  }
}

// 将多个buffer的数据合并写入到一个channel
@Test
public void gatheringWritesTest() {
  try (RandomAccessFile file = new RandomAccessFile("2parts.txt", "rw")) {
    FileChannel channel = file.getChannel();
    ByteBuffer d = ByteBuffer.allocate(4);
    ByteBuffer e = ByteBuffer.allocate(4);

    d.put(new byte[]{'f', 'o', 'u', 'r'});
    e.put(new byte[]{'f', 'i', 'v', 'e'});

    d.flip();
    e.flip();

    channel.write(new ByteBuffer[]{d, e});
  } catch (IOException e) {
    e.printStackTrace();
  }
}

// 会从channel读取数据填充ByteBuffer，返回值表示读到了多少字节，-1表示到达了文件的末尾
int readBytes = channel.read(buffer);

// 正确的写入姿势，如果buffer较大，一次会写不完
while(buffer.hasRemaining()) {
    channel.write(buffer);
}

// 获取当前位置
long pos = channel.position();

/**
 * 设置当前位置
 * 设置当前位置时，如果设置为文件的末尾
 *  - 这时读取会返回 -1 
 *  - 这时写入，会追加内容，但要注意如果position超过了文件末尾，再写入时在新内容和原末尾之间会有空洞（00）
 */
channel.position(newPos);

channel.size(); //获取文件大小

public void fileChannelTransferToTest() {
  try (
    FileChannel from = new FileInputStream("data.txt").getChannel();
    FileChannel to = new FileOutputStream("to.txt").getChannel();
  ) {
    // 效率高，底层会利用操作系统的零拷贝进行优化
    long size = from.size();
    // left 变量代表还剩余多少字节
    for (long left = size; left > 0; ) {
      System.out.println("position:" + (size - left) + " left:" + left);
      left -= from.transferTo((size - left), left, to);
    }
  } catch (IOException e) {
    e.printStackTrace();
  }
}

Path source = Paths.get("1.txt");                     // 相对路径使用user.dir环境变量来定位1.txt

Path source = Paths.get("d:\\1.txt");                 // 绝对路径代表了d:\1.txt

Path projects = Paths.get("d:\\data", "projects");    // 代表了d:\data\projects

Path path = Paths.get("d:\\data\\projects\\a\\..\\b"); 
path.normalize()                                      // 计算为d:\data\projects\b

Path path = Paths.get("data.txt");
System.out.println(Files.exists(path));

Path path = Paths.get("d1");
Files.createDirectory(path);

Path path = Paths.get("d1/d2/d3");
Files.createDirectories(path);

Path source = Paths.get("data.txt");
Path target = Paths.get("target.txt");
// 如果文件已存在，会抛异常FileAlreadyExistsException
// 可以使用Files.copy(source, target, StandardCopyOption.REPLACE_EXISTING);
Files.copy(source, target);

Path source = Paths.get("data.txt");
Path target = Paths.get("data.txt");

// StandardCopyOption.ATOMIC_MOVE 保证文件移动的原子性
Files.move(source, target, StandardCopyOption.ATOMIC_MOVE);

Path target = Paths.get("target.txt");
Files.delete(target); // 如果文件不存在，会抛异常 NoSuchFileException

Path target = Paths.get("d1");
Files.delete(target); // 如果目录还有内容，会抛异常 DirectoryNotEmptyException

@Test
public void loopDirTest() throws Exception {
  Path path = Paths.get("C:\\Program Files\\Java\\jdk1.8.0_181");
  Files.walkFileTree(path, new SimpleFileVisitor<Path>(){

    @Override
    public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
      log.info("preVisitDirectory dir={}", dir.getFileName());
      return super.preVisitDirectory(dir, attrs);
    }

    @Override
    public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
      log.info("visitFile file={}", file.getFileName());
      return super.visitFile(file, attrs);
    }

    @Override
    public FileVisitResult visitFileFailed(Path file, IOException exc) throws IOException {
      log.info("visitFileFailed file={} exc={}", file.getFileName(), exc.getMessage());
      return super.visitFileFailed(file, exc);
    }

    @Override
    public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
      log.info("postVisitDirectory dir={}", dir.getFileName());
      return super.postVisitDirectory(dir, exc);
    }
  });
}

@Test
public void blockIOTest() throws IOException {
  // 使用 nio 来理解阻塞模式, 单线程
  // 0. ByteBuffer
  ByteBuffer buffer = ByteBuffer.allocate(16);
  // 1. 创建了服务器
  ServerSocketChannel ssc = ServerSocketChannel.open();

  // 2. 绑定监听端口
  ssc.bind(new InetSocketAddress(8080));

  // 3. 连接集合
  List<SocketChannel> channels = new ArrayList<>();
  while (true) {
    // 4. accept 建立与客户端连接， SocketChannel 用来与客户端之间通信
    log.debug("connecting...");
    SocketChannel sc = ssc.accept(); // 阻塞方法，线程停止运行
    log.debug("connected... {}", sc);
    channels.add(sc);
    for (SocketChannel channel : channels) {
      // 5. 接收客户端发送的数据
      log.debug("before read... {}", channel);
      channel.read(buffer); // 阻塞方法，线程停止运行
      buffer.flip();
      ByteBufferUtil.debugRead(buffer);
      buffer.clear();
      log.debug("after read...{}", channel);
    }
  }
}

@Test
public void nonBlockIOTest() throws IOException {
  // 使用 nio 来理解非阻塞模式, 单线程
  // 0. ByteBuffer
  ByteBuffer buffer = ByteBuffer.allocate(16);
  // 1. 创建了服务器
  ServerSocketChannel ssc = ServerSocketChannel.open();
  // 非阻塞模式
  ssc.configureBlocking(false);
  // 2. 绑定监听端口
  ssc.bind(new InetSocketAddress(8080));
  // 3. 连接集合
  List<SocketChannel> channels = new ArrayList<>();
  while (true) {
    // 4. accept 建立与客户端连接， SocketChannel 用来与客户端之间通信
    SocketChannel sc = ssc.accept(); // 非阻塞，线程还会继续运行，如果没有连接建立，但sc是null
    if (sc != null) {
      log.debug("connected... {}", sc);
      sc.configureBlocking(false); // 非阻塞模式
      channels.add(sc);
    }
    for (SocketChannel channel : channels) {
      // 5. 接收客户端发送的数据
      int read = channel.read(buffer);// 非阻塞，线程仍然会继续运行，如果没有读到数据，read 返回 0
      if (read > 0) {
        buffer.flip();
        ByteBufferUtil.debugRead(buffer);
        buffer.clear();
        log.debug("after read...{}", channel);
      }
    }
  }
}

Selector selector = Selector.open();

/**
 * channel 必须工作在非阻塞模式
 * 绑定的事件类型可以有
 *  - connect - 客户端连接成功时触发
 *  - accept  - 服务器端成功接受连接时触发
 *  - read    - 数据可读入时触发，有因为接收能力弱，数据暂不能读入的情况
 *  - write   - 数据可写出时触发，有因为发送能力弱，数据暂不能写出的情况
 */
channel.configureBlocking(false);
SelectionKey key = channel.register(selector, SelectionKey.OP_ACCEPT);

// 阻塞直到绑定事件发生
int count = selector.select();

// 阻塞直到绑定事件发生，或是超时（时间单位为 ms）
int count = selector.select(1000L);

// 不会阻塞，也就是不管有没有事件，立刻返回，自己根据返回值检查是否有事件
int count = selector.selectNow();

@Test
public void acceptorTest() {
  try (ServerSocketChannel channel = ServerSocketChannel.open()) {
    channel.bind(new InetSocketAddress(8080));
    System.out.println(channel);
    Selector selector = Selector.open();
    channel.configureBlocking(false);
    channel.register(selector, SelectionKey.OP_ACCEPT);
    while (true) {
      int count = selector.select();
      log.debug("select count: {}", count);
      // 获取所有事件
      Set<SelectionKey> keys = selector.selectedKeys();
      // 遍历所有事件，逐一处理
      Iterator<SelectionKey> iter = keys.iterator();
      while (iter.hasNext()) {
        SelectionKey key = iter.next();
        // 判断事件类型
        if (key.isAcceptable()) {
          ServerSocketChannel c = (ServerSocketChannel) key.channel();
          // 必须处理
          SocketChannel sc = c.accept();
          log.debug("{}", sc);
        }
        // 处理完毕，必须将事件移除
        iter.remove();
      }
    }
  } catch (IOException e) {
    e.printStackTrace();
  }
}

@Test
public void readEventTest() {
  try (ServerSocketChannel channel = ServerSocketChannel.open()) {
    channel.bind(new InetSocketAddress(8080));
    System.out.println(channel);
    Selector selector = Selector.open();
    channel.configureBlocking(false);
    channel.register(selector, SelectionKey.OP_ACCEPT);

    while (true) {
      int count = selector.select();
      log.debug("select count: {}", count);
      // 获取所有事件
      Set<SelectionKey> keys = selector.selectedKeys();

      // 遍历所有事件，逐一处理
      Iterator<SelectionKey> iter = keys.iterator();
      while (iter.hasNext()) {
        SelectionKey key = iter.next();
        // 判断事件类型
        if (key.isAcceptable()) {
          ServerSocketChannel c = (ServerSocketChannel) key.channel();
          // 必须处理
          SocketChannel sc = c.accept();
          sc.configureBlocking(false);
          sc.register(selector, SelectionKey.OP_READ);
          log.debug("连接已建立: {}", sc);
        } else if (key.isReadable()) {
          SocketChannel sc = (SocketChannel) key.channel();
          ByteBuffer buffer = ByteBuffer.allocate(128);
          int read = sc.read(buffer);
          if (read == -1) {
            // cancel会取消注册在selector上的channel，并从keys集合中删除key后续不会再监听事件
            key.cancel();
            sc.close();
          } else {
            buffer.flip();
            ByteBufferUtil.debugRead(buffer);
          }
        }
        // 处理完毕，必须将事件移除
        iter.remove();
      }
    }
  } catch (IOException e) {
    e.printStackTrace();
  }
}

private static void split(ByteBuffer source) {
  source.flip();
  for (int i = 0; i < source.limit(); i++) {
    // 找到一条完整消息
    if (source.get(i) == '\n') {
      int length = i + 1 - source.position();
      // 把这条完整消息存入新的 ByteBuffer
      ByteBuffer target = ByteBuffer.allocate(length);
      // 从 source 读，向 target 写
      for (int j = 0; j < length; j++) {
        target.put(source.get());
      }
      ByteBufferUtil.debugAll(target);
    }
  }
  source.compact();
}


// 使用 \n 区分消息段
if (key.isReadable()) {
  try {
    SocketChannel channel = (SocketChannel) key.channel(); // 拿到触发事件的channel
    // 获取selectionKey上关联的附件，在accept时设置上的
    ByteBuffer buffer = (ByteBuffer) key.attachment();
    int read = channel.read(buffer); // 如果是正常断开，read的方法的返回值是-1
    if(read == -1) {
      key.cancel();
    } else {
      split(buffer);
      // 需要扩容
      if (buffer.position() == buffer.limit()) {
        ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2);
        buffer.flip();
        newBuffer.put(buffer);
        key.attach(newBuffer);
      }
    }
  } catch (IOException e) {
    e.printStackTrace();
    key.cancel();  // 因为客户端断开了,因此需要将 key 取消
  }
}

if (key.isAcceptable()) {
  SocketChannel sc = ssc.accept();
  sc.configureBlocking(false);
  SelectionKey sckey = sc.register(selector, SelectionKey.OP_READ);
  // 1. 向客户端发送内容
  StringBuilder sb = new StringBuilder();
  for (int i = 0; i < 3000000; i++) {
    sb.append("a");
  }
  ByteBuffer buffer = Charset.defaultCharset().encode(sb.toString());
  int write = sc.write(buffer);
  // 3. write 表示实际写了多少字节
  System.out.println("实际写入字节:" + write);
  // 4. 如果有剩余未读字节，才需要关注写事件
  if (buffer.hasRemaining()) {
    // read 1  write 4
    // 在原有关注事件的基础上，多关注 写事件
    sckey.interestOps(sckey.interestOps() + SelectionKey.OP_WRITE);
    // 把 buffer 作为附件加入 sckey
    sckey.attach(buffer);
  }
} else if (key.isWritable()) {
  ByteBuffer buffer = (ByteBuffer) key.attachment();
  SocketChannel sc = (SocketChannel) key.channel();
  int write = sc.write(buffer);
  System.out.println("实际写入字节:" + write);
  if (!buffer.hasRemaining()) { // 写完了
    key.interestOps(key.interestOps() - SelectionKey.OP_WRITE); // 如果不取消，会每次可写均会触发write事件
    key.attach(null);
  }
}

@Test
public void simpleServerTest() throws IOException {
  new ServerBootstrap()
    // 创建 NioEventLoopGroup，可以简单理解为 线程池 + Selector
    .group(new NioEventLoopGroup())
    // 选择服务Socket实现类，其中NioServerSocketChannel表示基于NIO的服务器端实现
    .channel(NioServerSocketChannel.class)
    //  childHandler添加的处理器都是给SocketChannel用的，而不是给ServerSocketChannel
    //  ChannelInitializer处理器（仅执行一次），它的作用是待客户端SocketChannel建立连接后，执行initChannel以便添加更多的处理器
    .childHandler(new ChannelInitializer<NioSocketChannel>() {

      protected void initChannel(NioSocketChannel ch) {
        // SocketChannel 的处理器，解码 ByteBuf => String
        ch.pipeline().addLast(new StringDecoder());
        
        // SocketChannel 的业务处理器，使用上一个处理器的处理结果
        ch.pipeline().addLast(new SimpleChannelInboundHandler<String>() {

          @Override
          protected void channelRead0(ChannelHandlerContext ctx, String msg) {
            System.out.println(msg);
          }
        });
      }
    })
    // 绑定的监听端口
    .bind(8080);
  System.in.read();
}

@Test
public void simpleClientTest() throws IOException, InterruptedException {
  ChannelFuture client = new Bootstrap()
    // 创建 NioEventLoopGroup，可以简单理解为 线程池 + Selector
    .group(new NioEventLoopGroup())
    // 选择服务Socket实现类，其中NioSocketChannel表示基于NIO的客户器端实现
    .channel(NioSocketChannel.class)
    //  ChannelInitializer处理器（仅执行一次），它的作用是待客户端SocketChannel建立连接后，执行initChannel以便添加更多的处理器
    .handler(new ChannelInitializer<NioSocketChannel>() {

      protected void initChannel(NioSocketChannel ch) {
        // SocketChannel 的处理器，编码 ByteBuf => String
        ch.pipeline().addLast(new StringEncoder());
      }
      // 连接服务器
    }).connect(new InetSocketAddress(8080));
  // Future是异步的，要等待连接上服务器
  client.sync();
  // 发送数据
  client.channel().writeAndFlush("Hello world");
  System.in.read();
}

// 简单的示例
DefaultEventLoopGroup group = new DefaultEventLoopGroup(2);
System.out.println(group.next());  //io.netty.channel.DefaultEventLoop@60f82f98
System.out.println(group.next());  //io.netty.channel.DefaultEventLoop@35f983a6
System.out.println(group.next());  //io.netty.channel.DefaultEventLoop@60f82f98

// 也可以是有for循环
DefaultEventLoopGroup group = new DefaultEventLoopGroup(2);
for (EventExecutor eventLoop : group) {
    System.out.println(eventLoop);
}

@Test
public void simpleClientTest() throws IOException, InterruptedException {
  NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
  ChannelFuture client = new Bootstrap()
    .group(eventLoopGroup)
    .channel(NioSocketChannel.class)
    .handler(new ChannelInitializer<NioSocketChannel>() {
      protected void initChannel(NioSocketChannel ch) {
        ch.pipeline().addLast(new StringEncoder());
      }
    }).connect(new InetSocketAddress(8080));
  client.sync();
  client.channel().writeAndFlush("Hello world");
  // 优雅关闭
  eventLoopGroup.shutdownGracefully().sync();
}

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
  final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
  // 下一个 handler 的事件循环是否与当前的事件循环是同一个线程
  EventExecutor executor = next.executor();

  // 是，直接调用
  if (executor.inEventLoop()) {
    next.invokeChannelRead(m);
  } 
  // 不是，将要执行的代码作为任务提交给下一个事件循环处理（换人）
  else {
    executor.execute(new Runnable() {
      @Override
      public void run() {
        next.invokeChannelRead(m);
      }
    });
  }
}

public void simpleClientTest() throws IOException, InterruptedException {
  NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
  ChannelFuture client = new Bootstrap()
    .group(eventLoopGroup)
    .channel(NioSocketChannel.class)
    .handler(new ChannelInitializer<NioSocketChannel>() {

      protected void initChannel(NioSocketChannel ch) {
        ch.pipeline().addLast(new StringEncoder());
      }
    }).connect(new InetSocketAddress(8080));
  // connect方法是异步的，意味着不等连接建立，方法执行就返回了。
  // 因此channelFuture对象中不能【立刻】获得到正确的Channel对象
  client.sync();
  // 返回的是ChannelFuture对象，它的作用是利用channel()方法来获取Channel对象
  client.channel().writeAndFlush("Hello world");
  ChannelFuture channelFuture = client.channel().closeFuture();
  // 可执行channel关闭后的扫尾操作
  channelFuture.addListener(new GenericFutureListener<Future<? super Void>>() {

    @Override
    public void operationComplete(Future<? super Void> future) throws Exception {
      System.out.println("关闭后扫尾操作");
    }
  });
  eventLoopGroup.shutdownGracefully().sync();
}

// ==================================== 同步处理任务成功 ====================================
public void syncSuccessTest() throws ExecutionException, InterruptedException {
  DefaultEventLoop eventExecutors = new DefaultEventLoop();
  DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);

  eventExecutors.execute(()->{
    try {
      Thread.sleep(1000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    log.info("set success, {}",10);
    promise.setSuccess(10);
  });

  log.info("start...");
  log.info("getNow = {}",promise.getNow()); // 还没有结果
  log.info("get = {}",promise.get());
}

// ==================================== 异步处理任务成功 ====================================
public void asyncSuccessTest() throws ExecutionException, InterruptedException {
  DefaultEventLoop eventExecutors = new DefaultEventLoop();
  DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);

  // 设置回调，异步接收结果
  promise.addListener(future -> {
    log.info("getNow = {}", future.getNow());
  });

  // 等待 1000 后设置成功结果
  eventExecutors.execute(() -> {
    try {
      Thread.sleep(1000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    log.info("set success, {}", 10);
    promise.setSuccess(10);
  });

  log.info("start...");
  promise.await();
}

private static void log(ByteBuf buffer) {
  int length = buffer.readableBytes();
  int rows = length / 16 + (length % 15 == 0 ? 0 : 1) + 4;
  StringBuilder buf = new StringBuilder(rows * 80 * 2)
    .append("read index:").append(buffer.readerIndex())
    .append(" write index:").append(buffer.writerIndex())
    .append(" capacity:").append(buffer.capacity())
    .append(" class:").append(buffer.getClass())
    .append(NEWLINE);
  ByteBufUtil.appendPrettyHexDump(buf, buffer);
  System.out.println(buf.toString());
}

@Test
public void byteBufTest() {
  // 默认创建的是池化的直接内存
  ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(10);
  // read index:0 write index:0 capacity:10 class:class io.netty.buffer.PooledUnsafeDirectByteBuf
  log(buffer);
}

ByteBuf buffer = ByteBufAllocator.DEFAULT.heapBuffer(10);

ByteBuf buffer = ByteBufAllocator.DEFAULT.directBuffer(10);

@Test
public void byteBufReadTest() {
  ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(10);
  buffer.writeBytes("ABCDEFG".getBytes());

  // 读取，影响读指针
  System.out.println((char)buffer.readByte()); // A
  System.out.println((char)buffer.readByte()); // B
  System.out.println((char)buffer.readByte()); // C
  System.out.println((char)buffer.readByte()); // D

  // mark标记读指针
  buffer.markReaderIndex();
  System.out.println((char)buffer.readByte()); // E
  System.out.println((char)buffer.readByte()); // F
  // 读指针复位
  buffer.resetReaderIndex();
  System.out.println((char)buffer.readByte()); // E

  // get方法不影响指针
  System.out.println((char)buffer.getByte(0)); // A
}

@Test
public void byteBufUnpooledTest() {
  ByteBuf buffer1 = Unpooled.buffer(10);
  buffer1.release();
  ByteBuf buffer2 = Unpooled.buffer(10);
  buffer2.release();

  // false
  System.out.println(buffer1 == buffer2);

  ByteBuf buffer3 = ByteBufAllocator.DEFAULT.buffer(10);
  buffer3.release();
  ByteBuf buffer4 = ByteBufAllocator.DEFAULT.buffer(10);
  buffer4.release();

  // true
  System.out.println(buffer3 == buffer4);
}

ch.pipeline().addLast(new FixedLengthFrameDecoder(8));

ch.pipeline().addLast(new LineBasedFrameDecoder(1024));

// 最大长度，长度偏移，长度占用字节，长度调整，剥离字节数
ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024, 0, 1, 0, 1));

/**
  * readerIdleTime 读空闲时间
  * writerIdleTime 写空闲时间
  * allIdleTime 所有空闲时间
  * unit 时间单位
  */
pipeline.addLast(new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS));

//======================================= 集成Marshalling =======================================
public class MarshallingCodeFactory {

  public static MarshallingDecoder bulidMarshallingDecoder(){
    //首先通过Marshalling工具类获取Marshalling实例对象，参数serial标识创建的是java序列化工厂对象
    MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
    //创建了MarshallingConfiguration对象，配置版本号为5
    MarshallingConfiguration configuration = new MarshallingConfiguration();
    configuration.setVersion(5);
    //根据marshallerFactory好Configuration创建provider
    UnmarshallerProvider provider = new DefaultUnmarshallerProvider(marshallerFactory, configuration);
    //构建Netty的MarshallingDecoder对象，俩个参数分别为provider和单个消息序列化后的最大长度
    MarshallingDecoder decoder = new MarshallingDecoder(provider, 1024*1024);
    return decoder;
  }

  public static MarshallingEncoder bulidMarshallingEncoder(){
    MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
    MarshallingConfiguration configuration = new MarshallingConfiguration();
    configuration.setVersion(5);
    MarshallerProvider provider = new DefaultMarshallerProvider(marshallerFactory, configuration);
    MarshallingEncoder encoder = new MarshallingEncoder(provider);
    return encoder;
  }
}

/**
 * 客户端和服务器都加上Marshalling编解码器
 */
protected void initChannel(SocketChannel ch) throws Exception {
  //注意顺序
  ch.pipeline().addLast(MarshallingCodeFactory.bulidMarshallingDecoder());
  ch.pipeline().addLast(MarshallingCodeFactory.bulidMarshallingEncoder());
  //...
}