原文链接 Netty中ByteBuf的引用计数线程安全的实现原理

代码仓库地址

ByteBuf 实现了ReferenceCounted 接口，实现了引用计数接口，该接口的retain(int) 方法为了保证线程安全使用了自旋锁来确保操作安全，那么选择了比较重要的实现类AbstractReferenceCountedByteBuf 来查看这一特性.

在JDK 1.5 之后，JDK的并发包提供了Atomic* 的相关类，来帮助开发者更好的完成并发操作，这里我们学习使用CAS来实现线程安全，CAS就是ComposeAndSet,中文含义，比较And更新,我们来看一个例子,具体的说明在注释中！

static class Increment {

    // volatile 保证了线程可见性，但是没有保证按成安全性

    private volatile int i;

    // 声明CAS的抽象类

    private static AtomicIntegerFieldUpdater<Increment> updater;

    static {

      // 通过静态方法创建CAS的具体实例

      // 第一个参数表示维护的Integer所在的类的class

      // 第二个参数表示维护的Integer变量的名称

      updater = AtomicIntegerFieldUpdater.newUpdater(Increment.class, "i");

    }

    /**

     * 构造初始化函数

     *

     * @param i

     */

    public Increment(int i) {

      this.i = i;

    }

    /** 自增一个 */

    public void increment() {

      increment(1);

    }

    /**

     * 重载自增方法

     *

     * <p>该方法具体实现了CAS的代码逻辑，重在均在下面的for循环中

     *

     * @param increment

     */

    public void increment(int increment) {

      // 创建一个死循环，compareAndSet检查不通过的时候，重新获取新的值尝试更新

      for (; ; ) {

        // 保存当前线程更新的值的状态

        int oldI = this.i;

        final int newValue = oldI + increment;

        // 校验值是否更新完成

        if (newValue <= increment) {

          throw new IllegalArgumentException("非法的参数异常");

        }

        // 如果CAS更新完成，则退出循环，否则打印重试日志

        if (updater.compareAndSet(this, oldI, newValue)) {

          break;

        }

        System.out.println("发生多线程并发,CAS正常重新尝试");

      }

    }

  }

实现了自增的CAS代码之后，我们创建一个MyRunnable 继承Runnable 在run() 方法中实现Increment 实例的自增，具体代码如下:

  static class MyRunnable implements Runnable {

    private Increment increment;

    public MyRunnable(Increment increment) {

      this.increment = increment;

    }

    @Override

    public void run() {

      increment.increment();

    }

  }

在main() 方法中我们创建一个Increment 示例对象，构造参数传1，然后创建50个线程，同时启动，观察最终的结果, 多次运行观察是否能实现线程安全.

  public static void main(String[] args) throws InterruptedException {

    Increment increment = new Increment(1);

    for (int i = 0; i < 50; i++) {

      new Thread(new MyRunnable(increment)).start();

    }

    // 延时4s, 等待所有线程执完成后，输出i

    TimeUnit.SECONDS.sleep(4);

    System.out.println(increment.i);

  }

最后输出结果为51，可见CAS 能够实现多线程安全的问题，相对于RelockCount和重量级的synchronized 各有各的适合场景，下面回到Netty的自旋锁的问题上来，可以看到下面的代码和上面的示例是类似的，Netty依靠这个方法来完成ByteBuf的引用计数，确保多线程操作ByteBuf的时候引用计数不会出现多线程的问题,

下面的代码是引用了Netty5 的 io.netty.buffer.AbstractReferenceCountedByteBuf 类的部分代码，有兴趣的朋友可以下载该框架的源码，找到该文件看一下。

    private static final AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> refCntUpdater;

    static {

        AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> updater =

                PlatformDependent.newAtomicIntegerFieldUpdater(AbstractReferenceCountedByteBuf.class, "refCnt");

        if (updater == null) {

            updater = AtomicIntegerFieldUpdater.newUpdater(AbstractReferenceCountedByteBuf.class, "refCnt");

        }

        refCntUpdater = updater;

    }

    @Override

    public ByteBuf retain(int increment) {

        if (increment <= 0) {

            throw new IllegalArgumentException("increment: " + increment + " (expected: > 0)");

        }

        for (;;) {

            int refCnt = this.refCnt;

            if (refCnt == 0) {

                throw new IllegalReferenceCountException(0, increment);

            }

            if (refCnt > Integer.MAX_VALUE - increment) {

                throw new IllegalReferenceCountException(refCnt, increment);

            }

            if (refCntUpdater.compareAndSet(this, refCnt, refCnt + increment)) {

                break;

            }

        }

        return this;

    }