io.netty.buffer.Unpooled#unreleasableBuffer ( )源码实例Demo

@Setup(Level.Trial)
public void setup() {
    byte[] bytes = new byte[256];
    content = Unpooled.buffer(bytes.length);
    content.writeBytes(bytes);
    ByteBuf testContent = Unpooled.unreleasableBuffer(content.asReadOnly());

    List<RedisMessage> rList = new ArrayList<RedisMessage>(arraySize);
    for (int i = 0; i < arraySize; ++i) {
        rList.add(new FullBulkStringRedisMessage(testContent));
    }
    redisArray = new ArrayRedisMessage(rList);
    encoder = new RedisEncoder();
    context = new EmbeddedChannelWriteReleaseHandlerContext(pooledAllocator ? PooledByteBufAllocator.DEFAULT :
            UnpooledByteBufAllocator.DEFAULT, encoder) {
        @Override
        protected void handleException(Throwable t) {
            handleUnexpectedException(t);
        }
    };
}
 

@Setup
public void setup() {
    // Use buffer sizes that will also allow to write UTF-8 without grow the buffer
    buffer = Unpooled.directBuffer(512);
    wrapped = Unpooled.unreleasableBuffer(Unpooled.directBuffer(512));
    asciiSequence = new StringBuilder(128);
    for (int i = 0; i < 128; i++) {
        asciiSequence.append('a');
    }
    ascii = asciiSequence.toString();

    // Generate some mixed UTF-8 String for benchmark
    utf8Sequence = new StringBuilder(128);
    char[] chars = "Some UTF-8 like äÄ∏ŒŒ".toCharArray();
    for (int i = 0; i < 128; i++) {
        utf8Sequence.append(chars[i % chars.length]);
    }
    utf8 = utf8Sequence.toString();
    asciiSequence = utf8Sequence;

    asciiBuffer = Unpooled.copiedBuffer(ascii, CharsetUtil.US_ASCII);
    utf8Buffer = Unpooled.copiedBuffer(utf8, CharsetUtil.UTF_8);
}
 

public void server(int port) throws Exception {
	final ByteBuf buf = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Hi!\r\n", Charset.forName("UTF-8")));
	EventLoopGroup group = new OioEventLoopGroup();
	try {
		ServerBootstrap b = new ServerBootstrap();
		b.group(group).channel(OioServerSocketChannel.class).localAddress(new InetSocketAddress(port))
				.childHandler(new ChannelInitializer<SocketChannel>() {
					@Override
					public void initChannel(SocketChannel ch) throws Exception {
						ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
							@Override
							public void channelActive(ChannelHandlerContext ctx) throws Exception {
								ctx.writeAndFlush(buf.duplicate()).addListener(ChannelFutureListener.CLOSE);
							}
						});
					}
				});
		ChannelFuture f = b.bind().sync();
		f.channel().closeFuture().sync();
	} finally {
		group.shutdownGracefully().sync();
	}
}
 

@Setup
public void setup() {
    // Use buffer sizes that will also allow to write UTF-8 without grow the buffer
    buffer = Unpooled.directBuffer(512);
    wrapped = Unpooled.unreleasableBuffer(Unpooled.directBuffer(512));
    asciiSequence = new StringBuilder(128);
    for (int i = 0; i < 128; i++) {
        asciiSequence.append('a');
    }
    ascii = asciiSequence.toString();

    // Generate some mixed UTF-8 String for benchmark
    utf8Sequence = new StringBuilder(128);
    char[] chars = "Some UTF-8 like äÄ∏ŒŒ".toCharArray();
    for (int i = 0; i < 128; i++) {
        utf8Sequence.append(chars[i % chars.length]);
    }
    utf8 = utf8Sequence.toString();
    asciiSequence = utf8Sequence;
}
 

public void server(int port) throws Exception {
    final ByteBuf buf =
            Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Hi!\r\n",
                    StandardCharsets.UTF_8));
    //为非阻塞模式使用NioEventLoopGroup
    NioEventLoopGroup group = new NioEventLoopGroup();
    try {
        //创建ServerBootstrap
        ServerBootstrap b = new ServerBootstrap();
        b.group(group).channel(NioServerSocketChannel.class)
                .localAddress(new InetSocketAddress(port))
                //指定 ChannelInitializer，对于每个已接受的连接都调用它
                .childHandler(new ChannelInitializer<SocketChannel>() {
                                  @Override
                                  public void initChannel(SocketChannel ch)
                                          throws Exception {
                                      ch.pipeline().addLast(
                                              //添加 ChannelInboundHandlerAdapter以接收和处理事件
                                              new ChannelInboundHandlerAdapter() {
                                                  @Override
                                                  public void channelActive(
                                                          //将消息写到客户端，并添加ChannelFutureListener，
                                                          //以便消息一被写完就关闭连接
                                                          ChannelHandlerContext ctx) throws Exception {
                                                      ctx.writeAndFlush(buf.duplicate())
                                                              .addListener(
                                                                      ChannelFutureListener.CLOSE);
                                                  }
                                              });
                                  }
                              }
                );
        //绑定服务器以接受连接
        ChannelFuture f = b.bind().sync();
        f.channel().closeFuture().sync();
    } finally {
        //释放所有的资源
        group.shutdownGracefully().sync();
    }
}
 

@SuppressWarnings({ "unchecked", "rawtypes", "null" })
GatewayHandler(WebSocketClientHandshaker handshaker) {
   this.authorizedMessages = new SerializedSubject(PublishSubject.create());
   this.registeredMessages = new SerializedSubject(PublishSubject.create());
   this.platformMessages = new SerializedSubject(PublishSubject.create());
   this.protocolMessages = new SerializedSubject(PublishSubject.create());

   this.handshaker = handshaker;
   this.websocketFrameBuf = Unpooled.unreleasableBuffer(Unpooled.buffer(GatewayConnection.WEBSOCKETS_MAX_FRAME_LENGTH));

   this.lastHubMsg = System.nanoTime();
   this.lastPlatformMsg = System.nanoTime();
}
 

@Setup(Level.Trial)
public void setup() {
    byte[] bytes = new byte[256];
    content = Unpooled.buffer(bytes.length);
    content.writeBytes(bytes);
    ByteBuf testContent = Unpooled.unreleasableBuffer(content.asReadOnly());
    HttpHeaders headersWithChunked = new DefaultHttpHeaders(false);
    headersWithChunked.add(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
    HttpHeaders headersWithContentLength = new DefaultHttpHeaders(false);
    headersWithContentLength.add(HttpHeaderNames.CONTENT_LENGTH, testContent.readableBytes());

    fullRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, "/index", testContent,
            headersWithContentLength, EmptyHttpHeaders.INSTANCE);
    contentLengthRequest = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, "/index",
            headersWithContentLength);
    chunkedRequest = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, "/index", headersWithChunked);
    lastContent = new DefaultLastHttpContent(testContent, false);

    encoder = new HttpRequestEncoder();
    context = new EmbeddedChannelWriteReleaseHandlerContext(pooledAllocator ? PooledByteBufAllocator.DEFAULT :
            UnpooledByteBufAllocator.DEFAULT, encoder) {
        @Override
        protected void handleException(Throwable t) {
            handleUnexpectedException(t);
        }
    };
}
 

public ChannelBufferWrapper(final ByteBuf buffer, boolean releasable, boolean pooled) {
   if (!releasable) {
      this.buffer = Unpooled.unreleasableBuffer(buffer);
   } else {
      this.buffer = buffer;
   }
   this.releasable = releasable;
   this.isPooled = pooled;

}
 

public void server(int port)
        throws Exception {
    final ByteBuf buf =
            Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Hi!\r\n", StandardCharsets.UTF_8));
    EventLoopGroup group = new OioEventLoopGroup();
    try {
        //创建 ServerBootstrap
        ServerBootstrap b = new ServerBootstrap();
        b.group(group)
                //使用 OioEventLoopGroup以允许阻塞模式（旧的I/O）
                .channel(OioServerSocketChannel.class)
                .localAddress(new InetSocketAddress(port))
                //指定 ChannelInitializer，对于每个已接受的连接都调用它
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch)
                            throws Exception {
                        ch.pipeline().addLast(
                                //添加一个 ChannelInboundHandlerAdapter以拦截和处理事件
                                new ChannelInboundHandlerAdapter() {
                                    @Override
                                    public void channelActive(
                                            ChannelHandlerContext ctx)
                                            throws Exception {
                                        ctx.writeAndFlush(buf.duplicate())
                                                .addListener(
                                                        //将消息写到客户端，并添加 ChannelFutureListener，
                                                        //以便消息一被写完就关闭连接
                                                        ChannelFutureListener.CLOSE);
                                    }
                                });
                    }
                });
        //绑定服务器以接受连接
        ChannelFuture f = b.bind().sync();
        f.channel().closeFuture().sync();
    } finally {
        //释放所有的资源
        group.shutdownGracefully().sync();
    }
}
 

BufferImpl(int initialSizeHint) {
    buffer = Unpooled.unreleasableBuffer(Unpooled.buffer(initialSizeHint, Integer.MAX_VALUE));
}
 

@Override
public NetBuffer createBuffer(int initialSizeHint) {
    ByteBuf buffer = Unpooled.unreleasableBuffer(Unpooled.buffer(initialSizeHint, Integer.MAX_VALUE));
    return new NettyBuffer(buffer);
}