netty源码阅读之编码之MessageToByteEncoder

你的名字 2022-05-09 05:40 194阅读 0赞

MessageToByteEncoder的write过程，我们分析以下几步：

1、匹配对象

2、分配内存

3、编码实现

4、释放对象

5、传播数据

6、释放内存

源码在这里：

@Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            ByteBuf buf = null;
            try {
                if (acceptOutboundMessage(msg)) {
                    @SuppressWarnings("unchecked")
                    I cast = (I) msg;
                    buf = allocateBuffer(ctx, cast, preferDirect);
                    try {
                        encode(ctx, cast, buf);
                    } finally {
                        ReferenceCountUtil.release(cast);
                    }
    
                    if (buf.isReadable()) {
                        ctx.write(buf, promise);
                    } else {
                        buf.release();
                        ctx.write(Unpooled.EMPTY_BUFFER, promise);
                    }
                    buf = null;
                } else {
                    ctx.write(msg, promise);
                }
            } catch (EncoderException e) {
                throw e;
            } catch (Throwable e) {
                throw new EncoderException(e);
            } finally {
                if (buf != null) {
                    buf.release();
                }
            }
        }

### 一、匹配对象 ###

匹配对象从acceptOutboundMessage这里进入，只要符合条件的对象才能进入下一步，否则继续往下传播，也就是

else {
                    ctx.write(msg, promise);
                }

我们看 acceptOutboundMessage这个的源码吧：

/**
         * Returns {@code true} if the given message should be handled. If {@code false} it will be passed to the next
         * {@link ChannelOutboundHandler} in the {@link ChannelPipeline}.
         */
        public boolean acceptOutboundMessage(Object msg) throws Exception {
            return matcher.match(msg);
        }

看看mater是何物：

private final TypeParameterMatcher matcher;

*/
        protected MessageToByteEncoder(boolean preferDirect) {
            matcher = TypeParameterMatcher.find(this, MessageToByteEncoder.class, "I");
            this.preferDirect = preferDirect;
        }

看match方法：

private static final class ReflectiveMatcher extends TypeParameterMatcher {
            private final Class<?> type;
    
            ReflectiveMatcher(Class<?> type) {
                this.type = type;
            }
    
            @Override
            public boolean match(Object msg) {
                return type.isInstance(msg);
            }
        }

也就是通过isInstance判断是否符合条件，稍微分析下就能知道结论了。

### 二、内存分配 ###

netty创建一个byteBuf，来存放传进来的信息，这里是为这个byteBuf分配内存。

内存分配就是这个函数：

buf = allocateBuffer(ctx, cast, preferDirect);

进入：

/**
         * Allocate a {@link ByteBuf} which will be used as argument of {@link #encode(ChannelHandlerContext, I, ByteBuf)}.
         * Sub-classes may override this method to returna {@link ByteBuf} with a perfect matching {@code initialCapacity}.
         */
        protected ByteBuf allocateBuffer(ChannelHandlerContext ctx, @SuppressWarnings("unused") I msg,
                                   boolean preferDirect) throws Exception {
            if (preferDirect) {
                return ctx.alloc().ioBuffer();
            } else {
                return ctx.alloc().heapBuffer();
            }
        }

一般情况下netty默认使用堆外内存。

### 三、编码实现 ###

编码实现呢，就是我们具体编码器做的东西，例如我们Encoder实现的方法：

public class Encoder extends MessageToByteEncoder<User> {
        @Override
        protected void encode(ChannelHandlerContext ctx, User user, ByteBuf out) throws Exception {
    
            byte[] bytes = user.getName().getBytes();
            out.writeInt(4 + bytes.length);
            out.writeInt(user.getAge());
            out.writeBytes(bytes);
        }
    }

还有很多netty自带的encoder，大家可以看里面的实现。

### 四、释放对象 ###

在上一步的编码的时候，我们注意到这里：

try {
                        encode(ctx, cast, buf);
                    } finally {
                        ReferenceCountUtil.release(cast);
                    }

finally的时候，还需要释放cast。cast就是我们传进来的msg，我们注意看源码，它只有是ReferenceCounted的时候，才释放对象，也就是传入进来的本身就是byteBuf，那么netty就会自动释放这个对象，不需要用户来释放。

/**
         * Try to call {@link ReferenceCounted#release()} if the specified message implements {@link ReferenceCounted}.
         * If the specified message doesn't implement {@link ReferenceCounted}, this method does nothing.
         */
        public static boolean release(Object msg) {
            if (msg instanceof ReferenceCounted) {
                return ((ReferenceCounted) msg).release();
            }
            return false;
        }

### 五、传播数据 ###

传播数据的代码是这一段：

if (buf.isReadable()) {
                        ctx.write(buf, promise);
                    } else {
                        buf.release();
                        ctx.write(Unpooled.EMPTY_BUFFER, promise);
                    }
                    buf = null;

如果buf数据写入成功，那就是可读的，ctx.write(buf,promise)就会传播，一直传播到headcontext节点。

否则，传播一个空的buf：

ctx.write(Unpooled.EMPTY_BUFFER, promise);

最后释放这个buf：buf=null

### 六、释放内存 ###

上一步最后buf=null的时候已经颇有释放内存的意思了，但是有可能抛出异常了走不到buf=null这一步，不管如何总是不能有内存泄漏，netty想得很周到，在fianlly的时候判断不为空也就是没有释放过那就需要释放：

finally {
                if (buf != null) {
                    buf.release();
                }
            }

最后说明一下，一般情况下不会走到了这里，只有出现异常才走到这里。