Netty核心類及其作用是什么

這篇文章主要介紹了Netty核心類及其作用是什么的相關知識，內容詳細易懂，操作簡單快捷，具有一定借鑒價值，相信大家閱讀完這篇Netty核心類及其作用是什么文章都會有所收獲，下面我們一起來看看吧。

MessageToByteEncoder

MessageToByteEncoder是一個抽象編碼器，子類可重寫encode方法把對象編碼為ByteBuf輸出。

MessageToByteEncoder繼承自ChannelOutboundHandlerAdapter，encode在出站是被調用。

public class MyMessageEncoder extends MessageToByteEncoder<MessagePO> {
    @Override
    protected void encode(ChannelHandlerContext ctx, MessagePO msg, ByteBuf out) throws Exception {
        System.out.println("MyMessageEncoder.encode，被調用");
        String json = JSONObject.toJSONString(msg);
        out.writeInt(json.getBytes(StandardCharsets.UTF_8).length);
        out.writeBytes(json.getBytes(StandardCharsets.UTF_8));
    }
}

ByteToMessageDecoder

ByteToMessageDecoder是一種ChannelInboundHandler，可以稱為解碼器，負責將byte字節流(ByteBuf)轉換成一種Message，Message是應用可以自己定義的一種Java對象。

ByteToMessageDecoder：用于將字節轉為消息，需要檢測緩沖區是否有足夠的字節。

public class MyMessageDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        System.out.println("MyMessageDecoder.decode，被調用");
        while (in.readableBytes() >= 4){
            int num = in.readInt();
            System.out.println("解碼出一個整數："+num);
            out.add(num);
        }
    }
}

ReplayingDecoder

ReplayingDecoder：繼承自ByteToMessageDecoder，不需要檢測緩沖區是否有足夠的字節，但是ReplayingDecoder的速度略慢于ByteToMessageDecoder，而且并不是所有的ByteBuf都支持。

項目復雜度高用ReplayingDecoder，否則使用ByteToMessageDecoder。

public class MyMessageDecoder extends ReplayingDecoder<Void> {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        System.out.println("MyMessageDecoder.decode，被調用");
        int length = in.readInt();
        byte[] content = new byte[length];
        in.readBytes(content);
        String json = new String(content,StandardCharsets.UTF_8);
        MessagePO po = JSONObject.parseObject(json,MessagePO.class);
        out.add(po);
    }
}

MessageToMessageEncoder

用于從一種消息編碼為另外一種消息，例如從POJO到POJO，是一種ChannelOutboundHandler

MessageToMessageDecoder

從一種消息解碼為另一種消息，例如POJO到POJO，是一種ChannelInboundHandler

MessageToMessageCodec

整合了MessageToMessageEncoder 和 MessageToMessageDecoder

public class RequestMessageCodec extends MessageToMessageCodec<String, RequestData> {
    @Override
    protected void encode(ChannelHandlerContext ctx, RequestData msg, List<Object> out) throws Exception {
        System.out.println("RequestMessageCodec.encode 被調用 " + msg);
        String json = JSONObject.toJSONString(msg);
        out.add(json);
    }
    @Override
    protected void decode(ChannelHandlerContext ctx, String msg, List<Object> out) throws Exception {
        System.out.println("RequestMessageCodec.decode 被調用 " + msg);
        RequestData po = JSONObject.parseObject(msg, RequestData.class);
        out.add(po);
    }
}

ChannelInitializer

ChannelInitializer是一種特殊的ChannelInboundHandler，可以通過一種簡單的方式（調用initChannel方法）來初始化Channel。

通常在Bootstrap.handler(ChannelHandler) ， ServerBootstrap.handler(ChannelHandler) 和 ServerBootstrap.childHandler(ChannelHandler)中給Channel設置ChannelPipeline。

注意：當initChannel被執行完后，會將當前的handler從Pipeline中移除。

Bootstrap bootstrap = new Bootstrap().group(group)//設置線程組
    .channel(NioSocketChannel.class)//設置客戶端通道的實現類
    .handler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast(new NettyClientHandler());//加入自己的處理器
        }
    });

ServerBootstrap bootstrap = new ServerBootstrap().group(bossGroup, workerGroup)
    .channel(NioServerSocketChannel.class)//使用NioServerSocketChannel作為服務器的通道實現
    .option(ChannelOption.SO_BACKLOG, 128)//設置線程隊列等待連接的個數
    .childOption(ChannelOption.SO_KEEPALIVE, true)//設置保持活動連接狀態
//      .handler(null)//該Handler對應bossGroup
    .childHandler(new ChannelInitializer<SocketChannel>() {//給workerGroup的EventLoop對應的管道設置處理器
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast(new NettyServerHandler());
        }
    });

SimpleChannelInboundHandler

SimpleChannelInboundHandler繼承自ChannelInboundHandlerAdapter，可以通過泛型來規定消息類型。

處理入站的數據我們只需要實現channelRead0方法。

SimpleChannelInboundHandler在接收到數據后會自動release掉數據占用的Bytebuffer資源，ChannelInboundHandlerAdapter不會自動釋放。

public class MyClientHandler extends SimpleChannelInboundHandler<MessagePO> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MessagePO msg) throws Exception {
        System.out.println("收到服務端消息：" + msg);
    }
}

DefaultEventLoopGroup

在向pipline中添加ChannelHandler時，可以提供一個新的線程組，Handler業務會在該線程中執行。

當加ChannelHandler需要執行多線程并發業務時，DefaultEventLoopGroup可以派上大用處。

如果沒有設置DefaultEventLoopGroup，默認使用的是EventLoopGroup workerGroup = new NioEventLoopGroup();

DefaultEventLoopGroup businessGroup = new DefaultEventLoopGroup(100);
...
addLast(businessGroup, new MyNettyServerHandler())

/**
 * 讀取客戶端發送過來的消息
 */
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    ByteBuf byteBuf = (ByteBuf) msg;
    System.out.println("收到客戶信息：" + byteBuf.toString(CharsetUtil.UTF_8));
    System.out.println("客戶端地址：" + ctx.channel().remoteAddress());
    System.out.println("處理線程：" + Thread.currentThread().getName());
    ctx.executor().parent().execute(()->{
        try {
            System.out.println("parent().execute Thread = " + Thread.currentThread().getName());
            TimeUnit.SECONDS.sleep(2L);
            System.out.println("parent任務執行完成1");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
    ctx.executor().parent().execute(()->{
        try {
            System.out.println("parent().execute Thread = " + Thread.currentThread().getName());
            TimeUnit.SECONDS.sleep(2L);
            System.out.println("parent任務執行完成2");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
    ctx.executor().parent().execute(()->{
        try {
            System.out.println("parent().execute Thread = " + Thread.currentThread().getName());
            TimeUnit.SECONDS.sleep(2L);
            System.out.println("parent任務執行完成3");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
}

以上代碼執行日志如下：

收到客戶信息：Hello 服務端
客戶端地址：/127.0.0.1:60345
處理線程：defaultEventLoopGroup-4-1
parent().execute Thread = defaultEventLoopGroup-4-2
parent().execute Thread = defaultEventLoopGroup-4-3
程序繼續~~ defaultEventLoopGroup-4-1
parent().execute Thread = defaultEventLoopGroup-4-4
parent任務執行完成1
parent任務執行完成3
parent任務執行完成2

EventLoop定時任務

可以在Handler中通過方法ctx.channel().eventLoop().schedule()添加定時任務

ctx.channel().eventLoop().schedule(()->{
    try {
        System.out.println("Thread.currentThread().getName() = " + Thread.currentThread().getName());
        TimeUnit.SECONDS.sleep(2L);
        System.out.println("定時任務執行完成");
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
},10L,TimeUnit.SECONDS);

關于“Netty核心類及其作用是什么”這篇文章的內容就介紹到這里，感謝各位的閱讀！相信大家對“Netty核心類及其作用是什么”知識都有一定的了解，大家如果還想學習更多知識，歡迎關注億速云行業資訊頻道。