[Pulsar] Cumulative Ack 原理

private void doCumulativeAckAsync(MessageIdImpl msgId, BitSetRecyclable bitSet) {    // Handle concurrent updates from different threads    LastCumulativeAck currentCumulativeAck = LastCumulativeAck.create(msgId, bitSet);    while (true) {        LastCumulativeAck lastCumulativeAck = this.lastCumulativeAck;        if (msgId.compareTo(lastCumulativeAck.messageId) > 0) {            if (LAST_CUMULATIVE_ACK_UPDATER.compareAndSet(this, this.lastCumulativeAck, currentCumulativeAck)) {                if (lastCumulativeAck.bitSetRecyclable != null) {                    try {                        lastCumulativeAck.bitSetRecyclable.recycle();                    } catch (Exception ignore) {                        // no-op                    }                    lastCumulativeAck.bitSetRecyclable = null;                }                lastCumulativeAck.recycle();                // Successfully updated the last cumulative ack. Next flush iteration will send this to broker.                cumulativeAckFlushRequired = true;                return;            }        } else {            currentCumulativeAck.recycle();            // message id acknowledging an before the current last cumulative ack            return;        }    }}