Netty 源码分析（五）

版本 4.1.15

官网：https://netty.io/

Netty is an asynchronous event-driven network application framework  for rapid development of maintainable high performance protocol servers & clients.

ReferenceCounted

io.netty.util.ReferenceCounted

引用计数文档

public interface ReferenceCounted
A reference-counted object that requires explicit deallocation.
When a new ReferenceCounted is instantiated, it starts with the reference count of 1. retain() increases the reference count, and release() decreases the reference count. If the reference count is decreased to 0, the object will be deallocated explicitly, and accessing the deallocated object will usually result in an access violation.
If an object that implements ReferenceCounted is a container of other objects that implement ReferenceCounted, the contained objects will also be released via release() when the container’s reference count becomes 0.

我们看看其中的方法

public interface ReferenceCounted {
    /**
     * Returns the reference count of this object.  If {@code 0}, it means this object has been deallocated.
     */
    int refCnt();

    /**
     * Increases the reference count by {@code 1}.
     */
    ReferenceCounted retain();

    /**
     * Increases the reference count by the specified {@code increment}.
     */
    ReferenceCounted retain(int increment);

    /**
     * Records the current access location of this object for debugging purposes.
     * If this object is determined to be leaked, the information recorded by this operation will be provided to you
     * via {@link ResourceLeakDetector}.  This method is a shortcut to {@link #touch(Object) touch(null)}.
     */
    ReferenceCounted touch();

    /**
     * Records the current access location of this object with an additional arbitrary information for debugging
     * purposes.  If this object is determined to be leaked, the information recorded by this operation will be
     * provided to you via {@link ResourceLeakDetector}.
     */
    ReferenceCounted touch(Object hint);

    /**
     * Decreases the reference count by {@code 1} and deallocates this object if the reference count reaches at
     * {@code 0}.
     *
     * @return {@code true} if and only if the reference count became {@code 0} and this object has been deallocated
     */
    boolean release();

    /**
     * Decreases the reference count by the specified {@code decrement} and deallocates this object if the reference
     * count reaches at {@code 0}.
     *
     * @return {@code true} if and only if the reference count became {@code 0} and this object has been deallocated
     */
    boolean release(int decrement);
}

AbstractReferenceCountedByteBuf

io.netty.buffer.AbstractReferenceCountedByteBuf

我们先来看两个比较重要的方法，retain() 和 release() 方法

retain()

io.netty.buffer.AbstractReferenceCountedByteBuf#retain()

retain() 方法可以使引用计数加一

@Override
public ByteBuf retain() {
    return retain0(1);
}

@Override
public ByteBuf retain(int increment) {
    return retain0(checkPositive(increment, "increment"));
}

private ByteBuf retain0(int increment) {
    for (;;) {
        int refCnt = this.refCnt;
        final int nextCnt = refCnt + increment;
        // 如果 refCnt = 0 的时候 nextCont = increment，就就应该被回收
        // Ensure we not resurrect (which means the refCnt was 0) and also that we encountered an overflow.
        if (nextCnt <= increment) {
            throw new IllegalReferenceCountException(refCnt, increment);
        }
        // 这里使用到了自旋锁
        if (refCntUpdater.compareAndSet(this, refCnt, nextCnt)) {
            break;
        }
    }
    return this;
}

java.util.concurrent.atomic.AtomicIntegerFieldUpdater

public abstract class AtomicIntegerFieldUpdater
extends Object
A reflection-based utility that enables atomic updates to designated volatile int fields of designated classes. This class is designed for use in atomic data structures in which several fields of the same node are independently subject to atomic updates.
Note that the guarantees of the compareAndSet method in this class are weaker than in other atomic classes. Because this class cannot ensure that all uses of the field are appropriate for purposes of atomic access, it can guarantee atomicity only with respect to other invocations of compareAndSet and set on the same updater.

AtomicIntegerFieldUpdater要点的总结：

1. 更新器必须是int类型的变量，不能是其他包装类型
2. 更新器的更新必须是volatile类型的变量，确保线程之间的共享变量时的立即可见性
3. 变量不能是static的，必须是实例变量，因为Unsafe.objectFieldOffset() 方法不支持静态变量（CAS操作本质是通过对象实例的偏移来直接进行赋值）
4. 更新器只能修改它可见范围内的变量，因为更新器是通过反射来得到这个变量，如果变量不可见就会报错

如果更新的变量时包装类型，那么可以使用AtomicReferenceFieldUpdater来进行更新

java.util.concurrent.atomic.AtomicIntegerFieldUpdater#compareAndSet

public abstract boolean compareAndSet(T obj,
int expect,
int update)
Atomically sets the field of the given object managed by this updater to the given updated value if the current value == the expected value. This method is guaranteed to be atomic with respect to other calls to compareAndSet and set, but not necessarily with respect to other changes in the field.
Parameters:
obj - An object whose field to conditionally set
expect - the expected value
update - the new value

一个不安全的更新

/**
 * @Author: cuzz
 * @Date: 2019/1/19 15:40
 * @Description:
 */
public class AtomicUpdateTest {
    public static void main(String[] args) {
        Person person = new Person();
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                try {
                    Thread.sleep(20);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.print(person.age++ + " "); // 1 6 7 5 4 2 3 1 8 9
            }).start();
        }

    }
}
class Person{
    int age = 1;
}

使用AtomicIntegerFieldUpdater

/**
 * @Author: cuzz
 * @Date: 2019/1/19 15:40
 * @Description:
 */
public class AtomicUpdateTest {
    public static void main(String[] args) {
        AtomicIntegerFieldUpdater<Person> fieldUpdater = AtomicIntegerFieldUpdater.newUpdater(Person.class, "age");
        Person person = new Person();
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                try {
                    Thread.sleep(20);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.print(fieldUpdater.getAndIncrement(person) + " "); // 1 4 3 2 5 6 7 10 9 8 
            }).start();
        }
    }
}
class Person{
    volatile int age = 1;
}

大概有以下两种字段适合用Atomic*FieldUpdater:

大多数用到这个字段的代码是在读取字段的值, 但仍然有通过CAS更新字段值的需求. 这个时候用AtomicInteger的话每个直接读取这个字段的地方都要多一次.get()调用, 用volatile又满足不了需求, 所以就用到了AtomicIntegerFieldUpdater

这个字段所属的类会被创建大量的实例对象, 如果用AtomicInteger, 每个实例里面都要创建AtomicInteger对象, 从而多出内存消耗. 比如一个链表类的Node, 用AtomicReference保存next显然是不合适的.

原文：https://blog.csdn.net/u012415542/article/details/80646605

private static final AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> refCntUpdater =
        AtomicIntegerFieldUpdater.newUpdater(AbstractReferenceCountedByteBuf.class, "refCnt");

Reference counted objects

引用计数文档：Reference counted objects

Netty 处理器

重要概念

• Netty 的处理器可以分为两类：入站处理器和出站处理器
• 入站处理器的顶层是 ChannelnboundHandler，出站处理器的顶层是 ChannelOutboundHandler
• 数据处理时常用的各种解码器本质上都是处理器
• 编解码器：无论我们向网络中写入的数据是什么类型，数据在网络中传递时，其都是以字节流的形式呈现，将数据有原本的字节流的操作成为编码（encode），将数据有字节转化为它原本的格式或是其它的操作成为解码（decode），编码统一称为（codec）
• 编码：本质上是一种出站处理器，因此，编码一定是一种 ChannelOutboundHandler
• 解码：本质上是一种入站处理器，因此，解码一定是一种 ChannelInboundHandler
• 在 Netty 中，编码器通常以 xxxEncoder命名；解码器通常以xxxDecoder命名

编写一个Long类型的解码器

编写一个解码器在客服端与服务端传输一个 Long 型的数据，Netty 为我们提供了 ByteToMessageDecoder

io.netty.handler.codec.ByteToMessageDecoder

io.netty.handler.codec
public abstract class ByteToMessageDecoder
extends ChannelInboundHandlerAdapter
ChannelInboundHandlerAdapter which decodes bytes in a stream-like fashion from one ByteBuf to an other Message type. For example here is an implementation which reads all readable bytes from the input ByteBuf and create a new ByteBuf.
public class SquareDecoder extends ByteToMessageDecoder {
@Override
public void decode(ChannelHandlerContext ctx, ByteBuf in, List