Disruptor 源码分析(10) 位置信息是否ok的标识

位置信息是否ok的标识

在生成者中是通过RingBuffer的pulish方法来修改的
调用顺序是这样的:

public void publish(long sequence)
    {
        sequencer.publish(sequence);
    }

 public void publish(final long sequence)
    {
        setAvailable(sequence);
        waitStrategy.signalAllWhenBlocking();
    }

 private void setAvailable(final long sequence)
    {
        setAvailableBufferValue(calculateIndex(sequence), calculateAvailabilityFlag(sequence));
    }

 private void setAvailableBufferValue(int index, int flag)
    {
        long bufferAddress = (index * SCALE) + BASE;
        UNSAFE.putOrderedInt(availableBuffer, bufferAddress, flag);
    }


 private int calculateIndex(final long sequence)
    {
        return ((int) sequence) & indexMask;
    }

calculateIndex是获取数组的第几个位置,indexMask = bufferSize - 1;indexMask的2进制表示就是数组的长度位数全部为1.
((int) sequence) & indexMask其实就是去掉了不是数组长度的干扰.

 private int calculateAvailabilityFlag(final long sequence)
    {
        return (int) (sequence >>> indexShift);
    }

calculateAvailabilityFlag是获取当前是圈了.
因为indexShift = Util.log2(bufferSize);
这里的bufferSize和RingBuffer的bufferSize是一样的
>>> indexShift和log2操作是对应的.sequence右移indexShift位,其实即时除以了数组的大小,意思就是第几圈了.



在消费者中
getHighestPublishedSequence会调用isAvailable方法

public boolean isAvailable(long sequence)
    {
        int index = calculateIndex(sequence);
        int flag = calculateAvailabilityFlag(sequence);
        long bufferAddress = (index * SCALE) + BASE;
        return UNSAFE.getIntVolatile(availableBuffer, bufferAddress) == flag;
    }

可以看到isAvailable方法和生成者的逻辑是一样的.这样就能正常验证了

上面说的多个生产者的情况,可以看到用了个和RingBuffer一样大小的数组来存储标志.
如果是单个生产者的话是可以用一个变量来存就可以了的.