netty系列之:使用POJO替代buf

時間 2021-08-14

標籤 java git 程序員 github 瀏覽器 ide oop 編碼 netty code 欄目 Netty 简体版

原文原文鏈接

簡介

在以前的文章中咱們提到了，對於NioSocketChannel來講，它不接收最基本的string消息，只接收ByteBuf和FileRegion。可是ByteBuf是以二進制的形式進行處理的，對於程序員來講太不直觀了，處理起來也比較麻煩，有沒有可能直接處理java簡單對象呢？本文將會探討一下這個問題。java

decode和encode

好比咱們須要直接向channel中寫入一個字符串，在以前的文章中，咱們知道這是不能夠的，會報下面的錯誤：git

DefaultChannelPromise@57f5c075(failure: java.lang.UnsupportedOperationException: unsupported message type: String (expected: ByteBuf, FileRegion))

也就說ChannelPromise只接受ByteBuf和FileRegion，那麼怎麼作呢？程序員

既然ChannelPromise只接受ByteBuf和FileRegion，那麼咱們就須要把String對象轉換成ByteBuf便可。github

也就是說在寫入String以前把String轉換成ByteBuf，當要讀取數據的時候，再把ByteBuf轉換成String。瀏覽器

咱們知道ChannelPipeline中能夠添加多個handler，而且控制這些handler的順序。ide

那麼咱們的思路就出來了，在ChannelPipeline中添加一個encode，用於數據寫入的是對數據進行編碼成ByteBuf，而後再添加一個decode,用於在數據寫出的時候對數據進行解碼成對應的對象。oop

encode,decode是否是很熟悉？對了，這就是對象的序列化。編碼

對象序列化

netty中對象序列化是要把傳輸的對象和ByteBuf直接互相轉換，固然咱們能夠本身實現這個轉換對象。可是netty已經爲咱們提供了方便的兩個轉換類：ObjectEncoder和ObjectDecoder。netty

先看ObjectEncoder，他的做用就是將對象轉換成爲ByteBuf。code

這個類很簡單，咱們對其分析一下：

public class ObjectEncoder extends MessageToByteEncoder<Serializable> {
    private static final byte[] LENGTH_PLACEHOLDER = new byte[4];

    @Override
    protected void encode(ChannelHandlerContext ctx, Serializable msg, ByteBuf out) throws Exception {
        int startIdx = out.writerIndex();

        ByteBufOutputStream bout = new ByteBufOutputStream(out);
        ObjectOutputStream oout = null;
        try {
            bout.write(LENGTH_PLACEHOLDER);
            oout = new CompactObjectOutputStream(bout);
            oout.writeObject(msg);
            oout.flush();
        } finally {
            if (oout != null) {
                oout.close();
            } else {
                bout.close();
            }
        }

        int endIdx = out.writerIndex();

        out.setInt(startIdx, endIdx - startIdx - 4);
    }
}

ObjectEncoder繼承了MessageToByteEncoder，而MessageToByteEncoder又繼承了ChannelOutboundHandlerAdapter。爲何是OutBound呢？這是由於咱們是要對寫入的對象進行轉換，因此是outbound。

首先使用ByteBufOutputStream對out ByteBuf進行封裝，在bout中，首先寫入了一個LENGTH_PLACEHOLDER字段，用來表示stream中中Byte的長度。而後用一個CompactObjectOutputStream對bout進行封裝，最後就能夠用CompactObjectOutputStream寫入對象了。

對應的，netty還有一個ObjectDecoder對象，用於將ByteBuf轉換成對應的對象，ObjectDecoder繼承自LengthFieldBasedFrameDecoder，實際上他是一個ByteToMessageDecoder，也是一個ChannelInboundHandlerAdapter，用來對數據讀取進行處理。

咱們看下ObjectDecoder中最重要的decode方法：

protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        ByteBuf frame = (ByteBuf) super.decode(ctx, in);
        if (frame == null) {
            return null;
        }

        ObjectInputStream ois = new CompactObjectInputStream(new ByteBufInputStream(frame, true), classResolver);
        try {
            return ois.readObject();
        } finally {
            ois.close();
        }
    }

上面的代碼能夠看到，將輸入的ByteBuf轉換爲ByteBufInputStream，最後轉換成爲CompactObjectInputStream，就能夠直接讀取對象了。

使用編碼和解碼器

有了上面兩個編碼解碼器，直接須要將其添加到client和server端的ChannelPipeline中就能夠了。

對於server端，其核心代碼以下：

//定義bossGroup和workerGroup
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline p = ch.pipeline();
                    p.addLast(
                            // 添加encoder和decoder
                            new ObjectEncoder(),
                            new ObjectDecoder(ClassResolvers.cacheDisabled(null)),
                            new PojoServerHandler());
                }
             });

            // 綁定端口，並準備接受鏈接
            b.bind(PORT).sync().channel().closeFuture().sync();

一樣的，對於client端，咱們其核心代碼以下：

EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
             .channel(NioSocketChannel.class)
             .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline p = ch.pipeline();
                    p.addLast(
                            // 添加encoder和decoder
                            new ObjectEncoder(),
                            new ObjectDecoder(ClassResolvers.cacheDisabled(null)),
                            new PojoClientHandler());
                }
             });

            // 創建鏈接
            b.connect(HOST, PORT).sync().channel().closeFuture().sync();

能夠看到上面的邏輯就是將ObjectEncoder和ObjectDecoder添加到ChannelPipeline中便可。

最後，就能夠在客戶端和瀏覽器端經過調用：