Java 类io.netty.util.CharsetUtil 实例源码

/**
 * 返回http信息
 * @param ctx
 * @param req
 * @param res
 */
private static void sendHttpResponse(
        ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
    // Generate an error page if response getStatus code is not OK (200).
    if (res.getStatus().code() != 200) {
        ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);
        res.content().writeBytes(buf);
        buf.release();
        HttpHeaders.setContentLength(res, res.content().readableBytes());
    }

    // Send the response and close the connection if necessary.
    ChannelFuture f = ctx.channel().writeAndFlush(res);
    if (!HttpHeaders.isKeepAlive(req) || res.getStatus().code() != 200) {
        f.addListener(ChannelFutureListener.CLOSE);
    }
}

/**
 * 获取请求参数的Map
 * @param request
 * @return
 */
public static Map<String, List<String>> getParameterMap(HttpRequest request){
    Map<String, List<String>> paramMap = new HashMap<String, List<String>>();

    HttpMethod method = request.method();
    if(HttpMethod.GET.equals(method)){
        String uri = request.uri();
        QueryStringDecoder queryDecoder = new QueryStringDecoder(uri, CharsetUtil.UTF_8);
        paramMap = queryDecoder.parameters();

    }else if(HttpMethod.POST.equals(method)){
        FullHttpRequest fullRequest = (FullHttpRequest) request;
        paramMap = getPostParamMap(fullRequest);
    }

    return paramMap;
}

private boolean writeResponse(HttpObject currentObj, ChannelHandlerContext ctx) {
    // Decide whether to close the connection or not.
    boolean keepAlive = HttpHeaders.isKeepAlive(request);

    // Build the response object.
    FullHttpResponse response = new DefaultFullHttpResponse(
        HTTP_1_1, currentObj.getDecoderResult().isSuccess() ? OK : BAD_REQUEST,
        Unpooled.copiedBuffer(buf.toString(), CharsetUtil.UTF_8));

    response.headers().set(CONTENT_TYPE, "application/json");

    if (keepAlive) {
        // Add 'Content-Length' header only for a keep-alive connection.
        response.headers().set(CONTENT_LENGTH, response.content().readableBytes());
        // Add keep alive header as per:
        // - http://www.w3.org/Protocols/HTTP/1.1/draft-ietf-http-v11-spec-01.html#Connection
        response.headers().set(CONNECTION, HttpHeaders.Values.KEEP_ALIVE);
    }

    // Write the response.
    ctx.write(response);

    return keepAlive;
}

public void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
   if (res.getStatus().code() != 200) {
      ByteBuf f = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);
      res.content().clear();
      res.content().writeBytes(f);
      f.release();
   }

   HttpHeaders.setContentLength(res, res.content().readableBytes());
   ChannelFuture f1;
   f1 = ctx.channel().writeAndFlush(res);

   if (!HttpHeaders.isKeepAlive(req) || res.getStatus().code() != 200) {
      f1.addListener(ChannelFutureListener.CLOSE);
   }
}

public static String hashToBase64(ByteBuf objectState) {
    ByteBuffer bf = objectState.internalNioBuffer(objectState.readerIndex(), objectState.readableBytes());
    long h1 = LongHashFunction.farmUo().hashBytes(bf);
    long h2 = LongHashFunction.xx().hashBytes(bf);

    ByteBuf buf = ByteBufAllocator.DEFAULT.buffer((2 * Long.SIZE) / Byte.SIZE);
    try {
        buf.writeLong(h1).writeLong(h2);
        ByteBuf b = Base64.encode(buf);
        try {
            String s = b.toString(CharsetUtil.UTF_8);
            return s.substring(0, s.length() - 2);
        } finally {
            b.release();
        }
    } finally {
        buf.release();
    }
}

/**
 * 请求分发与处理
 *
 * @param request http协议请求
 * @return 处理结果
 * @throws InvocationTargetException 调用异常
 * @throws IllegalAccessException    参数异常
 */
public Object doDispatcher(FullHttpRequest request) throws InvocationTargetException, IllegalAccessException {
    Object[] args;
    String uri = request.uri();
    if (uri.endsWith("favicon.ico")) {
        return "";
    }

    AceServiceBean aceServiceBean = Context.getAceServiceBean(uri);
    AceHttpMethod aceHttpMethod = AceHttpMethod.getAceHttpMethod(request.method().toString());
    ByteBuf content = request.content();
    //如果要多次解析，请用 request.content().copy()
    QueryStringDecoder decoder = new QueryStringDecoder(uri);
    Map<String, List<String>> requestMap = decoder.parameters();
    Object result = aceServiceBean.exec(uri, aceHttpMethod, requestMap, content == null ? null : content.toString(CharsetUtil.UTF_8));
    String contentType = request.headers().get("Content-Type");
    if (result == null) {
        ApplicationInfo mock = new ApplicationInfo();
        mock.setName("ace");
        mock.setVersion("1.0");
        mock.setDesc(" mock  !!! ");
        result = mock;
    }
    return result;

}

private static void sendHttpResponse(ChannelHandlerContext ctx,
        HttpRequest req, FullHttpResponse res) {
    // Generate an error page if response getStatus code is not OK (200).
    if (res.getStatus().code() != 200) {
        ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(),
                CharsetUtil.UTF_8);
        res.content().writeBytes(buf);
        buf.release();
        setContentLength(res, res.content().readableBytes());
    }

    // Send the response and close the connection if necessary.
    ChannelFuture f = ctx.writeAndFlush(res);
    if (!isKeepAlive(req) || res.getStatus().code() != 200) {
        f.addListener(ChannelFutureListener.CLOSE);
    }
}

private void processProtocolInitFrame(ByteBuf buffer, List<Object> out) {
    if (buffer.readableBytes() >= 8) {
        CharSequence protocolName = buffer.readCharSequence(4, CharsetUtil.US_ASCII);
        buffer.skipBytes(1);
        byte majorVersion = buffer.readByte();
        byte minorVersion = buffer.readByte();
        byte revision = buffer.readByte();

        if (!AMQP_PROTOCOL_IDENTIFIER.equals(protocolName)) {
            out.add(new AmqpBadMessage(new IllegalArgumentException("Unknown protocol name " +
                                                                           protocolName.toString())));
            currentState = State.BAD_MESSAGE;
        }

        out.add(new ProtocolInitFrame(majorVersion, minorVersion, revision));
    }
}

public void requestData(String data) {
    if (ctx != null) {
        try {
            ChannelFuture f = ctx.writeAndFlush(Unpooled.copiedBuffer(data, CharsetUtil.UTF_8)).sync();

            if (!f.isSuccess())
                try {
                    throw f.cause();
                } catch (Throwable throwable) {
                    throwable.printStackTrace();
                }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    ByteBuf in = (ByteBuf) msg;
    String sentData = in.toString(CharsetUtil.UTF_8);
    String returnee = sentData + "-::=::-" + "{}";

    RequestProcessor reprocessor = EchoServer.process(sentData);

    if (reprocessor != null)
        returnee = sentData + "-::=::-" + reprocessor.getProcessedData();

    ChannelFuture f = ctx.writeAndFlush(Unpooled.copiedBuffer(returnee, CharsetUtil.UTF_8)).sync();

    if (!f.isSuccess())
        try {
            throw f.cause();
        } catch (Throwable throwable) {
            throwable.printStackTrace();
        }
}

/**
 * End the response immediately
 *
 * @param data data to send
 * @param options more options, specify the second arg of a valid encoding option, e.g `gzip`, `deflate`
 */
public void end(String data, String[]... options) {

    header(CONTENT_LENGTH, Integer.toString(data.length()));

    if (httpResponse == null) {

        setHttpResponse(new DefaultFullHttpResponse(
                HTTP_1_1,
                getStatus(),
                Unpooled.copiedBuffer(data, CharsetUtil.UTF_8)
        ));
    }

    writeFlush(!keepAlive);
}

public byte[] httpPostJson(String url,String json) throws Exception
{
    HttpURLConnection conn=buildConn(url);
    try {
        conn.setRequestMethod("POST");
        conn.setDoOutput(true);
        conn.setRequestProperty("Content-Type","application/json");
        conn.getOutputStream().write(json.getBytes(CharsetUtil.UTF_8));
        conn.getOutputStream().flush();
        conn.getOutputStream().close();
        return InputStreamUtils.getBytes(conn.getInputStream());
    } finally {
        conn.getInputStream().close();
        conn.disconnect();
    }
}

@Override
public String readString(ByteBuf source) {
  int len = readUnsignedShort(source);
  String str = source.toString(source.readerIndex(), len, CharsetUtil.UTF_8);
  source.readerIndex(source.readerIndex() + len);
  return str;
}

@Override
public String readLongString(ByteBuf source) {
  int len = readInt(source);
  String str = source.toString(source.readerIndex(), len, CharsetUtil.UTF_8);
  source.readerIndex(source.readerIndex() + len);
  return str;
}

/**
 * 写入数据到客户端
 *
 * @param messageResult
 */
public void write(final MessageResult messageResult) {
    String json = messageResult.toJson();
    ByteBuf content = Unpooled.copiedBuffer(json, CharsetUtil.UTF_8);
    FullHttpResponse res = new DefaultFullHttpResponse(HTTP_1_1, OK, content);
    res.headers().set(CONTENT_TYPE, "text/html; charset=UTF-8");
    HttpHeaders.setContentLength(res, content.readableBytes());

    // Send the response
    ChannelFuture f = this.channel.writeAndFlush(res);
}

@Override
public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
    ChildData data = event.getData();
    if(data==null || data.getData()==null){
        return;
    }
    SlaveNode slaveNode = SlaveNode.parse(JSON.parseObject(data.getData(),JSONObject.class));
    if(slaveNode==null){
        LOGGER.error("get a null slaveNode with eventType={},path={},data={}",event.getType(),data.getPath(),data.getData());
    }else {
        switch (event.getType()) {
            case CHILD_ADDED:
                slaveNodeMap.put(slaveNode.getId(), slaveNode);
                LOGGER.info("CHILD_ADDED with path={},data={},current slaveNode size={}", data.getPath(), new String(data.getData(),CharsetUtil.UTF_8),slaveNodeMap.size());
                break;
            case CHILD_REMOVED:
                slaveNodeMap.remove(slaveNode.getId());
                LOGGER.info("CHILD_REMOVED with path={},data={},current slaveNode size={}", data.getPath(), new String(data.getData(),CharsetUtil.UTF_8),slaveNodeMap.size());
                break;
            case CHILD_UPDATED:
                slaveNodeMap.replace(slaveNode.getId(), slaveNode);
                LOGGER.info("CHILD_UPDATED with path={},data={},current slaveNode size={}", data.getPath(), new String(data.getData(),CharsetUtil.UTF_8),slaveNodeMap.size());
                break;
            default:
                break;
        }
    }
}

/**
 * 转换byte
 * @param content
 * @return
 */
public static byte[] getBytes(Object content){
    if(content==null){
        return EMPTY_CONTENT.getBytes(CharsetUtil.UTF_8);
    }
    String data = content.toString();
    data = (data==null || data.trim().length()==0)?EMPTY_CONTENT:data;
    return data.getBytes(CharsetUtil.UTF_8);
}

public void sendError(ChannelHandlerContext ctx, HttpResponseStatus status) {
   FullHttpResponse response = new DefaultFullHttpResponse(
         HttpVersion.HTTP_1_1, status, Unpooled.copiedBuffer("Failure: " + status + "\r\n", CharsetUtil.UTF_8));
   response.headers().set(HttpHeaders.Names.CONTENT_TYPE, "text/plain; charset=UTF-8");

   // Close the connection as soon as the error message is sent.
   ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}

/**
 * 在一个连接建立时被调用,确保了数据会尽可能快地写入服务器
 * 此时编码了字符串 "hello netty!!!" 的字节缓冲区
 * @param ctx
 * @throws Exception
 */
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    /**
     * 当被通知Channel是活跃的时候,发送一条消息
     */
    ctx.writeAndFlush(Unpooled.copiedBuffer("hello netty!!!", CharsetUtil.UTF_8));
}

@Override
public Map<String, String> decode(ByteBuf buf, State state) {
    String value = buf.toString(CharsetUtil.UTF_8);
    Map<String, String> result = new HashMap<String, String>();
    for (String entry : value.split("\r\n|\n")) {
        String[] parts = entry.split(":");
        if (parts.length == 2) {
            result.put(parts[0], parts[1]);
        }
    }
    return result;
}

@Override
protected void encode(ChannelHandlerContext ctx, CommandData<?, ?> msg, ByteBuf out) throws Exception {
    try {
        out.writeByte(ARGS_PREFIX);
        int len = 1 + msg.getParams().length;
        if (msg.getCommand().getSubName() != null) {
            len++;
        }
        out.writeBytes(convert(len));
        out.writeBytes(CRLF);

        writeArgument(out, msg.getCommand().getName().getBytes(CharsetUtil.UTF_8));
        if (msg.getCommand().getSubName() != null) {
            writeArgument(out, msg.getCommand().getSubName().getBytes(CharsetUtil.UTF_8));
        }

        for (Object param : msg.getParams()) {
            ByteBuf buf = encode(param);
            writeArgument(out, buf);
            if (!(param instanceof ByteBuf)) {
                buf.release();
            }
        }

        if (log.isTraceEnabled()) {
            log.trace("channel: {} message: {}", ctx.channel(), out.toString(CharsetUtil.UTF_8));
        }
    } catch (Exception e) {
        msg.getPromise().tryFailure(e);
        throw e;
    }
}

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    /**
     * 客户端写一条消息给服务端
     */
    ctx.writeAndFlush(Unpooled.copiedBuffer("hi server", CharsetUtil.UTF_8));
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, Object msg)
        throws Exception {
    Channel ch = ctx.channel();
    if (!handshaker.isHandshakeComplete()) {
        handshaker.finishHandshake(ch, (FullHttpResponse) msg);
        System.out.println("WebSocket Client connected!");
        handshakeFuture.setSuccess();
        return;
    }
    if (msg instanceof FullHttpResponse) {
        FullHttpResponse response = (FullHttpResponse) msg;
        throw new IllegalStateException(
                "Unexpected FullHttpResponse (getStatus=" + response.getStatus() +
                    ", content=" + response.content().toString(CharsetUtil.UTF_8) + ')');
    } else if (msg instanceof WebSocketFrame) {
        WebSocketFrame frame = (WebSocketFrame) msg;
        if (msg instanceof TextWebSocketFrame) {
            TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
            System.out.println("WebSocket Client received message: " + textFrame.text());
        } else if (msg instanceof PongWebSocketFrame) {
            System.out.println("WebSocket Client received pong");
        } else if (msg instanceof CloseWebSocketFrame) {
            System.out.println("WebSocket Client received closing");
            ch.close();
        }
    }
}

public String decode(String source) throws Exception {
    String tokenSource = source;
    int zeroFlag = 0;
    if (tokenSource.startsWith("$")) {
        zeroFlag = Integer.parseInt(tokenSource.substring(1, 3));
        tokenSource = tokenSource.substring(3);
    }
    String token = tokenSource.replace('-', '+').replace('_', '/').replace('.', '=');
    byte[] base64Decode = SecurityUtil.base64Decode(token.getBytes(CharsetUtil.UTF_8));
    byte[] iv = new byte[0X10];
    byte[] content = new byte[base64Decode.length - iv.length];
    System.arraycopy(base64Decode, 0, iv, 0, iv.length);
    System.arraycopy(base64Decode, iv.length, content, 0, content.length);
    Cipher cipher = Cipher.getInstance(TRANSFORMATION);
    cipher.init(Cipher.DECRYPT_MODE, generateKey, new IvParameterSpec(iv));
    byte[] doFinal = cipher.doFinal(content);
    // 去补零
    int realSize = doFinal.length;
    if (zeroFlag > 0) {
        realSize = doFinal.length - zeroFlag;
        byte[] afterZero = new byte[realSize];
        System.arraycopy(doFinal, 0, afterZero, 0, afterZero.length);
        return new String(afterZero, CharsetUtil.UTF_8);
    } else {
        return new String(doFinal, CharsetUtil.UTF_8);
    }
}

public static byte[] download(String url, int port, String username, String password, String remotePath,
        String fileName) throws IOException {
    FTPClient ftp = new FTPClient();
    ftp.setConnectTimeout(5000);
    ftp.setAutodetectUTF8(true);
    ftp.setCharset(CharsetUtil.UTF_8);
    ftp.setControlEncoding(CharsetUtil.UTF_8.name());
    try {
        ftp.connect(url, port);
        ftp.login(username, password);// 登录
        if (!FTPReply.isPositiveCompletion(ftp.getReplyCode())) {
            ftp.disconnect();
            throw new IOException("login fail!");
        }
        ftp.changeWorkingDirectory(remotePath);
        ftp.setFileType(FTPClient.BINARY_FILE_TYPE);
        FTPFile[] fs = ftp.listFiles();
        for (FTPFile ff : fs) {
            if (ff.getName().equals(fileName)) {
                try (ByteArrayOutputStream is = new ByteArrayOutputStream();) {
                    ftp.retrieveFile(ff.getName(), is);
                    byte[] result = is.toByteArray();
                    return result;
                }
            }
        }

        ftp.logout();
    } finally {
        if (ftp.isConnected()) {
            ftp.disconnect();
        }
    }
    return null;
}

public static List<byte[]> download(String url, int port, String username, String password, String remotePath,
        String dirName, String filePattern) throws IOException {
    List<byte[]> result = new ArrayList<>();
    FTPClient ftp = new FTPClient();
    ftp.setConnectTimeout(5000);
    ftp.setAutodetectUTF8(true);
    ftp.setCharset(CharsetUtil.UTF_8);
    ftp.setControlEncoding(CharsetUtil.UTF_8.name());
    try {
        ftp.connect(url, port);
        ftp.login(username, password);// 登录
        if (!FTPReply.isPositiveCompletion(ftp.getReplyCode())) {
            ftp.disconnect();
            throw new IOException("login fail!");
        }
        ftp.changeWorkingDirectory(remotePath);
        ftp.changeWorkingDirectory(dirName);
        ftp.setFileType(FTPClient.BINARY_FILE_TYPE);
        FTPFile[] fs = ftp.listFiles();
        for (FTPFile ff : fs) {
            if (ff.getName().endsWith("." + filePattern)) {
                try (ByteArrayOutputStream is = new ByteArrayOutputStream();) {
                    ftp.retrieveFile(ff.getName(), is);
                    result.add(is.toByteArray());
                }
            }
        }

        ftp.logout();
    } finally {
        if (ftp.isConnected()) {
            ftp.disconnect();
        }
    }
    return result;
}

public static String md5Encode32(String source) throws NoSuchAlgorithmException {
    byte[] strTemp = source.getBytes(CharsetUtil.UTF_8);
    MessageDigest mdTemp = MessageDigest.getInstance(ALGORITHM_MD5);
    mdTemp.update(strTemp);
    byte[] md = mdTemp.digest();
    char[] hexEncode = hexEncode(md);
    return new String(hexEncode);
}

public static byte[] utf8Encode(CharSequence string) {
    try {
        ByteBuffer bytes = CharsetUtil.UTF_8.newEncoder().encode(CharBuffer.wrap(string));
        byte[] bytesCopy = new byte[bytes.limit()];
        System.arraycopy(bytes.array(), 0, bytesCopy, 0, bytes.limit());
        return bytesCopy;
    } catch (CharacterCodingException e) {
        throw new IllegalArgumentException("Encoding failed", e);
    }
}

private void sendError(ChannelHandlerContext ctx, HttpResponseStatus status) {
    FullHttpResponse response = new DefaultFullHttpResponse(
            HttpVersion.HTTP_1_1, status, Unpooled.copiedBuffer("Failure: " + status + "\r\n", CharsetUtil.UTF_8));
    response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8");

    // Close the connection as soon as the error message is sent.
    ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}

protected void initChannel(SocketChannel channel) throws Exception {
    log.info("New channel created");
    channel.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8));
    channel.pipeline().addLast(new MessageDecoder());
    handleNewNodeConnection(channel);

}

/**
 * 写入String类型
 * 
 * @param buffer
 * @param value
 */
protected final void putString(String value) {
    if (value == null) {
        putByteArray(null);
    } else if ("".equals(value)) {
        putByteArray(new byte[0]);
    } else {
        byte[] bytes = value.getBytes(CharsetUtil.UTF_8);
        putByteArray(bytes);
    }
}

/**
 * 读取String类型
 * 
 * @param buffer
 * @return
 */
protected final String getString() {
    byte[] bytes = getByteArray();
    if (bytes == null) {
        return null;
    } else if (bytes.length == 0) {
        return "";
    } else {
        return new String(bytes, CharsetUtil.UTF_8);
    }
}

@Override
public void run() {
    EventLoopGroup group = new NioEventLoopGroup();
    try {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                        pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
                        pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
                        pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
                        pipeline.addLast(new SimpleClientChannelHandler());
                    }

                });
        ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
        if (channelFuture.isSuccess()) {
            System.out.println(String.format("connect server(%s:%s) sucess", host, port));
        }
        channelFuture.channel().closeFuture().sync();
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        group.shutdownGracefully();
    }
}

private void sendError(ChannelHandlerContext ctx, HttpResponseStatus status) {
    FullHttpResponse response = new DefaultFullHttpResponse(
            HTTP_1_1, status, Unpooled.copiedBuffer("Failure: " + status + "\r\n", CharsetUtil.UTF_8));
    response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8");

    // Close the connection as soon as the error message is sent.
    ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}

private void sendListing(ChannelHandlerContext ctx, File dir) throws IOException {
    FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
    response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html; charset=UTF-8");
    ByteBuf buffer = Unpooled.copiedBuffer(Pages.getDirectory(dir), CharsetUtil.UTF_8);
    response.content().writeBytes(buffer);
    buffer.release();

    // Close the connection as soon as the error message is sent.
    ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}

@Override
public void handleTimeout(ChannelHandlerContext ctx) {
    HttpRequest req = ctx.attr(HttpSessionStore.key).get();
    String sessionId = HttpSessionStore.getClientSessionId(req);
    HttpJsonpChannelEntity httpChannelEntity = (HttpJsonpChannelEntity) MemoryMetaPool
            .getChannelEntryByClientId(sessionId);
    httpChannelEntity.setCtx(null);
    // empty json
    ByteBuf content = Unpooled.copiedBuffer("{}", CharsetUtil.UTF_8);
    ctx.writeAndFlush(content).addListener(ChannelFutureListener.CLOSE);
}

private ChannelInitializer channelInitializer() {
    return new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
            ChannelPipeline pipeline = socketChannel.pipeline();
            pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
            pipeline.addLast("stringDecoder", new StringDecoder(CharsetUtil.UTF_8));
            pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
            pipeline.addLast("stringEncoder", new StringEncoder(CharsetUtil.UTF_8));
            pipeline.addLast("idleStateHandler", new IdleStateHandler(300, 0, 0));
            pipeline.addLast("handler", nodeChannelInBoundHandler());
        }
    };
}

@PostConstruct
public void start() throws Exception {


    ServerBootstrap bootstrap = new ServerBootstrap();
    bootstrap.group(bossGroup, workGroup)
            .channelFactory(NioServerSocketChannel::new)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {

                    ChannelPipeline pipeline = socketChannel.pipeline();
                    pipeline.addLast("logHandler", new LoggingHandler());
                    pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                    pipeline.addLast("stringDecoder", new StringDecoder(CharsetUtil.UTF_8));
                    pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
                    pipeline.addLast("stringEncoder", new StringEncoder(CharsetUtil.UTF_8));
                    pipeline.addLast("idleStateHandler", new IdleStateHandler(300, 0, 0));

                    pipeline.addLast("handler", deviceChannelInboundHandler);
                }
            })

            .option(ChannelOption.SO_BACKLOG, 1000000)  //连接队列深度
            .option(ChannelOption.TCP_NODELAY, true)   //设置 no_delay
            .option(ChannelOption.SO_SNDBUF, 2048).option(ChannelOption.SO_RCVBUF, 1024)
            .childOption(ChannelOption.TCP_NODELAY, true)
            .childOption(ChannelOption.SO_REUSEADDR, true)
            .childOption(ChannelOption.SO_SNDBUF, 2048).childOption(ChannelOption.SO_RCVBUF, 1024)
            .childOption(ChannelOption.SO_LINGER, 0);

    bootstrap.bind(goPushNodeServerConfig.getDevicePort()).sync();
    log.info("device server start successful! listening port: {}", goPushNodeServerConfig.getDevicePort());
}

private void sendError(ChannelHandlerContext ctx, HttpResponseStatus status) {
    ByteBuf content = Unpooled.copiedBuffer("Failure: " + status + "\r\n",
            CharsetUtil.UTF_8);
    FullHttpResponse response = new DefaultFullHttpResponse(
            HttpVersion.HTTP_1_1, status, content);
    response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8");

    // Close the connection as soon as the error message is sent.
    ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}

public static <T> T readBytes(byte[] bytes, Class<T> valueType ){
    try {
        return gson.fromJson(new String(bytes, CharsetUtil.UTF_8), valueType);
    } catch (Exception e){
        e.printStackTrace();
    }
    return null;
}