一、netty的ByteBuf

创建ByteBuf

  1. import io.netty.buffer.*;
  3. //非池化堆内存分配
  4. ByteBuf buf = UnpooledByteBufAllocator.DEFAULT.heapBuffer(int initialCapacity, int maxCapacity);
  5. //池化堆内存分配
  6. ByteBuf buf = PooledByteBufAllocator.DEFAULT.heapBuffer(int initialCapacity, int maxCapacity);
  7. //默认分配(根据当前环境默认指定是池化还是非池化)
  8. ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(int initialCapacity, int maxCapacity);

使用ByteBuf

  1. import io.netty.buffer.*;
  3. @Test
  4. public void myBytebuf(){
  5.     ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(8, 12);
  6.     System.out.println("初始大小8字节，最大12字节");
  7.     print(buf);
  8.     System.out.println("-----填充4个字节后-----------");
  9.     buf.writeBytes(new byte[]{1,2,3,4});
  10.     print(buf);
  11.     System.out.println("-----填充4个字节后-----------");
  12.     buf.writeBytes(new byte[]{1,2,3,4});
  13.     print(buf);
  14.     System.out.println("-----填充4个字节后-----------");
  15.     buf.writeBytes(new byte[]{1,2,3,4});
  16.     print(buf);
  17.     System.out.println("-----填充4个字节后-----------");
  18.     buf.writeBytes(new byte[]{1,2,3,4});
  19.     print(buf);
  20. }

初始大小8字节，最大12字节buf.isReadable() :false buf.readerIndex() :0 buf.readableBytes() 0 buf.isWritable() :true buf.writerIndex() :0 buf.writableBytes() :8 buf.capacity() :8 buf.maxCapacity() :12 buf.isDirect() :true ——-填充4个字节后—————- buf.isReadable() :true buf.readerIndex() :0 buf.readableBytes() 4 buf.isWritable() :true buf.writerIndex() :4 buf.writableBytes() :4 buf.capacity() :8 buf.maxCapacity() :12 buf.isDirect() :true ——-填充4个字节后—————- buf.isReadable() :true buf.readerIndex() :0 buf.readableBytes() 8 buf.isWritable() :false buf.writerIndex() :8 buf.writableBytes() :0 buf.capacity() :8 buf.maxCapacity() :12 buf.isDirect() :true ——-填充4个字节后—————- buf.isReadable() :true buf.readerIndex() :0 buf.readableBytes() 12 buf.isWritable() :false buf.writerIndex() :12 buf.writableBytes() :0 buf.capacity() :12 buf.maxCapacity() :12 buf.isDirect() :true ——-填充4个字节后—————-

java.lang.IndexOutOfBoundsException: writerIndex(12) + minWritableBytes(4) exceeds maxCapacity(12): PooledUnsafeDirectByteBuf(ridx: 0, widx: 12, cap: 12/12)

  1. at io.netty.buffer.AbstractByteBuf.ensureWritable0(AbstractByteBuf.java:295)
  2. at io.netty.buffer.AbstractByteBuf.ensureWritable(AbstractByteBuf.java:282)
  3. at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1074)
  4. at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1082)
  5. at com.bjmashibing.system.io.netty.MyNetty.myBytebuf(MyNetty.java:65)
  6. at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  7. at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  8. at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  9. at java.lang.reflect.Method.invoke(Method.java:498)
  10. at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:47)
  11. at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
  12. at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:44)
  13. at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
  14. at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:271)
  15. at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:70)
  16. at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:50)
  17. at org.junit.runners.ParentRunner$3.run(ParentRunner.java:238)
  18. at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:63)
  19. at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:236)
  20. at org.junit.runners.ParentRunner.access$000(ParentRunner.java:53)
  21. at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:229)
  22. at org.junit.runners.ParentRunner.run(ParentRunner.java:309)
  23. at org.junit.runner.JUnitCore.run(JUnitCore.java:160)
  24. at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
  25. at com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:33)
  26. at com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:230)
  27. at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:58)

Process finished with exit code -1

二、netty之client 端

2.1 cleintModeV1

  1. import io.netty.buffer.ByteBuf;
  2. import io.netty.buffer.Unpooled;
  3. import io.netty.channel.ChannelFuture;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.ChannelInboundHandlerAdapter;
  6. import io.netty.channel.ChannelPipeline;
  7. import io.netty.channel.nio.NioEventLoopGroup;
  8. import io.netty.channel.socket.nio.NioSocketChannel;
  9. import io.netty.util.CharsetUtil;
  10. import org.junit.Test;
  12. import java.net.InetSocketAddress;
  14. public class MyNetty {
  16.     @Test
  17.     public void cleintMode() throws InterruptedException {
  18.         //创建时间监听器
  19.         NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup(1);
  20.         //创建客户端
  21.         NioSocketChannel nioSocketChannel = new NioSocketChannel();
  22.         nioEventLoopGroup.register(nioSocketChannel);
  23.         //连接服务端
  24.         nioSocketChannel.connect(new InetSocketAddress("192.168.235.133",9000));
  25.         //先向服务端发送“hello server”
  26.         ByteBuf hello = Unpooled.copiedBuffer("hello server".getBytes());
  27.         ChannelFuture channelFuture = nioSocketChannel.writeAndFlush(hello);
  28.         //注册处理器 监听服务端的响应
  29.         ChannelPipeline pipeline = nioSocketChannel.pipeline();
  30.         pipeline.addLast(new ChannelInboundHandlerAdapter(){
  31.             @Override
  32.             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  33.                 //msg -> ByteBuf
  34.                 ByteBuf byteBuf =  (ByteBuf)msg;
  35.                 CharSequence data = byteBuf.getCharSequence(0, byteBuf.readableBytes(), CharsetUtil.UTF_8);
  36.                 System.out.println("客户端读取到服务端"+ctx.pipeline().channel().remoteAddress().toString()+"的数据==>:"+data);
  37.             }
  38.         });
  39.         //设置客户端为同步
  40.         channelFuture.channel().closeFuture().sync();
  41.         //程序能执行到这里证明连接已经关闭了
  42.         System.out.println("连接关闭了");
  43.     }
  45. }

打开服务端，使用命令开启一个临时监听端口

  1. [root@192 ~]# nc -l 192.168.235.133 9000
  2. hello server
  3. hello client
  4. ^C
  5. [root@192 ~]#

运行程序，获得结果：

客户端读取到服务端/192.168.235.133:9000的数据==>: 客户端读取到服务端/192.168.235.133:9000的数据==>:hello client

连接关闭了

2.2 nettyClientModeV2

  1. import io.netty.bootstrap.Bootstrap;
  2. import io.netty.buffer.ByteBuf;
  3. import io.netty.buffer.Unpooled;
  4. import io.netty.channel.*;
  5. import io.netty.channel.nio.NioEventLoopGroup;
  6. import io.netty.channel.socket.nio.NioSocketChannel;
  7. import io.netty.util.CharsetUtil;
  8. import org.junit.Test;
  10. import java.net.InetSocketAddress;
  12. public class MyNetty {
  14.     @Test
  15.     public void nettyClientMode() throws InterruptedException {
  16.         //创建时间监听器
  17.         NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup(1);
  18.         //netty的一个引导工具
  19.         Bootstrap bootstrap = new Bootstrap();
  20.         //设置事件循环组
  21.         bootstrap.group(nioEventLoopGroup);
  22.         //设置通道为NioSocketChannel
  23.         bootstrap.channel(NioSocketChannel.class);
  24.         //设置事件处理器
  25.         bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
  26.             @Override
  27.             protected void initChannel(NioSocketChannel ch) throws Exception {
  28.                 ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
  29.                     @Override
  30.                     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  31.                         //msg -> ByteBuf
  32.                         ByteBuf byteBuf =  (ByteBuf)msg;
  33.                         CharSequence data = byteBuf.getCharSequence(0, byteBuf.readableBytes(), CharsetUtil.UTF_8);
  34.                         System.out.println("客户端读取到服务端"+ctx.pipeline().channel().remoteAddress().toString()+"的数据==>:"+data);
  35.                     }
  36.                 });
  37.             }
  38.         });
  39.         //设置要链接的服务端地址+端口
  40.         ChannelFuture channelFuture = bootstrap.connect(new InetSocketAddress("192.168.235.133", 9000));
  41.         //先向服务端发送“hello server”
  42.         ByteBuf hello = Unpooled.copiedBuffer("hello server".getBytes());
  43.         channelFuture.channel().writeAndFlush(hello);
  44.         //设置程序同步
  45.         channelFuture.channel().closeFuture().sync();
  46.         //程序能执行到这里证明连接已经关闭了
  47.         System.out.println("连接关闭了");
  48.     }
  50. }

三、netty之server端

3.1 serverModeV1

  1. import io.netty.bootstrap.Bootstrap;
  2. import io.netty.buffer.ByteBuf;
  3. import io.netty.buffer.Unpooled;
  4. import io.netty.channel.*;
  5. import io.netty.channel.nio.NioEventLoopGroup;
  6. import io.netty.channel.socket.SocketChannel;
  7. import io.netty.channel.socket.nio.NioServerSocketChannel;
  8. import io.netty.channel.socket.nio.NioSocketChannel;
  9. import io.netty.util.CharsetUtil;
  10. import org.junit.Test;
  12. import java.net.InetSocketAddress;
  14. public class MyNetty {
  16.     @Test
  17.     public void serverModeV1() throws InterruptedException {
  18.         //创建事件监听器
  19.         NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(1);
  20.         //NIO版的socket
  21.         NioServerSocketChannel nioServerSocketChannel = new NioServerSocketChannel();
  23.         //注册到事件循环组中
  24.         eventLoopGroup.register(nioServerSocketChannel);
  26.         nioServerSocketChannel.pipeline().addLast(new MyAcceptHander().setEventLoopGroup(eventLoopGroup).setChannelHandler(new ChannelInit()));
  28.         //设置监听端口
  29.         ChannelFuture bind = nioServerSocketChannel.bind(new InetSocketAddress(9000));
  31.         bind.sync().channel().closeFuture().sync();
  32.         System.out.println("server close....");
  33.     }
  36.     /**
  37.      * 自定义处理接收状态变化的处理器
  38.      */
  39.     private class MyAcceptHander extends ChannelInboundHandlerAdapter{
  41.         private EventLoopGroup eventLoopGroup;
  42.         private ChannelHandler channelHandler;
  44.         public MyAcceptHander setEventLoopGroup(EventLoopGroup eventLoopGroup) {
  45.             this.eventLoopGroup = eventLoopGroup;
  46.             return this;
  47.         }
  49.         public MyAcceptHander setChannelHandler(ChannelHandler channelHandler) {
  50.             this.channelHandler = channelHandler;
  51.             return this;
  52.         }
  54.         @Override
  55.         public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
  56.             System.out.println("触发了MyAcceptHander.channelRegistered()");
  57.         }
  59.         @Override
  60.         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  61.             System.out.println("触发了MyAcceptHander.channelRead()");
  62.             SocketChannel client = (SocketChannel)msg;
  63.             ChannelPipeline p = client.pipeline();
  64.             p.addLast(this.channelHandler);
  65. //            this.eventLoopGroup.register(client);
  66.         }
  67.     }
  69.     /**
  70.      * 为啥要有一个inithandler，可以没有，但是MyInHandler就得设计成单例
  71.      */
  72.     @ChannelHandler.Sharable
  73.     class ChannelInit extends ChannelInboundHandlerAdapter{
  75.         @Override
  76.         public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
  77.             Channel client = ctx.channel();
  78.             ChannelPipeline p = client.pipeline();
  79.             p.addLast(new MyInHandler());//2,client::pipeline[ChannelInit,MyInHandler]
  80.             ctx.pipeline().remove(this);
  81.         }
  82.     }
  85.     /**
  86.      * 就是用户自己实现的，你能说让用户放弃属性的操作吗
  87.      * @ChannelHandler.Sharable 不应该被强压给coder
  88.      */
  89.     class MyInHandler extends ChannelInboundHandlerAdapter {
  91.         @Override
  92.         public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
  93.             System.out.println("client  registed...");
  94.         }
  96.         @Override
  97.         public void channelActive(ChannelHandlerContext ctx) throws Exception {
  98.             System.out.println("client active...");
  99.         }
  101.         @Override
  102.         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  103.             ByteBuf buf = (ByteBuf) msg;
  104. //        CharSequence str = buf.readCharSequence(buf.readableBytes(), CharsetUtil.UTF_8);
  105.             CharSequence str = buf.getCharSequence(0,buf.readableBytes(), CharsetUtil.UTF_8);
  106.             System.out.println(str);
  107.             ctx.writeAndFlush(buf);
  108.         }
  109.     }
  111. }

3.2 nettyServerModeV2

  1. import io.netty.bootstrap.Bootstrap;
  2. import io.netty.bootstrap.ServerBootstrap;
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.buffer.Unpooled;
  5. import io.netty.channel.*;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.ServerSocketChannel;
  8. import io.netty.channel.socket.SocketChannel;
  9. import io.netty.channel.socket.nio.NioServerSocketChannel;
  10. import io.netty.channel.socket.nio.NioSocketChannel;
  11. import io.netty.util.CharsetUtil;
  12. import org.junit.Test;
  14. import java.net.InetSocketAddress;
  16. public class MyNetty {
  18.     @Test
  19.     public void nettyServerModeV2() throws InterruptedException {
  20.         //创建事件监听器
  21.         NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(1);
  22.         ServerBootstrap serverBootstrap = new ServerBootstrap();
  23.         serverBootstrap.group(eventLoopGroup,eventLoopGroup);
  24.         serverBootstrap.channel(NioServerSocketChannel.class);
  25.         serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {
  26.             @Override
  27.             protected void initChannel(NioSocketChannel ch) throws Exception {
  28.                     ch.pipeline().addLast(new MyInHandler());
  29.             }
  30.         });
  31.         ChannelFuture bind = serverBootstrap.bind(9000);
  32.         bind.channel().closeFuture().sync();
  33.         System.out.println("server close....");
  34.     }
  37.     /**
  38.      * 自定义处理接收状态变化的处理器
  39.      */
  40.     private class MyAcceptHander extends ChannelInboundHandlerAdapter{
  42.         private EventLoopGroup eventLoopGroup;
  43.         private ChannelHandler channelHandler;
  45.         public MyAcceptHander setEventLoopGroup(EventLoopGroup eventLoopGroup) {
  46.             this.eventLoopGroup = eventLoopGroup;
  47.             return this;
  48.         }
  50.         public MyAcceptHander setChannelHandler(ChannelHandler channelHandler) {
  51.             this.channelHandler = channelHandler;
  52.             return this;
  53.         }
  55.         @Override
  56.         public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
  57.             System.out.println("触发了MyAcceptHander.channelRegistered()");
  58.         }
  60.         @Override
  61.         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  62.             System.out.println("触发了MyAcceptHander.channelRead()");
  63.             SocketChannel client = (SocketChannel)msg;
  64.             ChannelPipeline p = client.pipeline();
  65.             p.addLast(this.channelHandler);
  66. //            this.eventLoopGroup.register(client);
  67.         }
  68.     }
  70.     /**
  71.      * 为啥要有一个inithandler，可以没有，但是MyInHandler就得设计成单例
  72.      */
  73.     @ChannelHandler.Sharable
  74.     class ChannelInit extends ChannelInboundHandlerAdapter{
  76.         @Override
  77.         public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
  78.             Channel client = ctx.channel();
  79.             ChannelPipeline p = client.pipeline();
  80.             p.addLast(new MyInHandler());//2,client::pipeline[ChannelInit,MyInHandler]
  81.             ctx.pipeline().remove(this);
  82.         }
  83.     }
  86.     /**
  87.      * 就是用户自己实现的，你能说让用户放弃属性的操作吗
  88.      * @ChannelHandler.Sharable 不应该被强压给coder
  89.      */
  90.     class MyInHandler extends ChannelInboundHandlerAdapter {
  92.         @Override
  93.         public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
  94.             System.out.println("client  registed...");
  95.         }
  97.         @Override
  98.         public void channelActive(ChannelHandlerContext ctx) throws Exception {
  99.             System.out.println("client active...");
  100.         }
  102.         @Override
  103.         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  104.             ByteBuf buf = (ByteBuf) msg;
  105. //        CharSequence str = buf.readCharSequence(buf.readableBytes(), CharsetUtil.UTF_8);
  106.             CharSequence str = buf.getCharSequence(0,buf.readableBytes(), CharsetUtil.UTF_8);
  107.             System.out.println(str);
  108.             ctx.writeAndFlush(buf);
  109.         }
  110.     }
  111. }