java.util.concurrent.atomic.AtomicLong#compareAndSet()源码实例Demo

/**
 * Atomically subtract the given number (positive, not validated) from the target field unless it contains Long.MAX_VALUE.
 *
 * @param requested the target field holding the current requested amount
 * @param emitted the produced element count, positive (not validated)
 * @return the new amount
 */
public static long subtract(AtomicLong requested, long emitted) {
    for (;;) {
        long current = requested.get();
        if (current == Long.MAX_VALUE) {
            return Long.MAX_VALUE;
        }
        long update = current - emitted;
        if (update < 0L) {
            update = 0L;
        }
        if (requested.compareAndSet(current, update)) {
            return update;
        }
    }
}
 

/**
 * Atomically subtract the given number (positive, not validated) from the target field unless it contains Long.MAX_VALUE.
 * @param requested the target field holding the current requested amount
 * @param n the produced element count, positive (not validated)
 * @return the new amount
 */
public static long produced(AtomicLong requested, long n) {
    for (;;) {
        long current = requested.get();
        if (current == Long.MAX_VALUE) {
            return Long.MAX_VALUE;
        }
        long update = current - n;
        if (update < 0L) {
            RxJavaCommonPlugins.onError(new IllegalStateException("More produced than requested: " + update));
            update = 0L;
        }
        if (requested.compareAndSet(current, update)) {
            return update;
        }
    }
}
 

/**
 * Removes and returns the value associated with {@code key}. If {@code key} is not in the map,
 * this method has no effect and returns zero.
 */

@CanIgnoreReturnValue
public long remove(K key) {
  AtomicLong atomic = map.get(key);
  if (atomic == null) {
    return 0L;
  }

  while (true) {
    long oldValue = atomic.get();
    if (oldValue == 0L || atomic.compareAndSet(oldValue, 0L)) {
      // only remove after setting to zero, to avoid concurrent updates
      map.remove(key, atomic);
      // succeed even if the remove fails, since the value was already adjusted
      return oldValue;
    }
  }
}
 

/**
 * If {@code (key, value)} is currently in the map, this method removes it and returns
 * true; otherwise, this method returns false.
 */
boolean remove(K key, long value) {
    AtomicLong atomic = map.get(key);
    if (atomic == null) {
        return false;
    }

    long oldValue = atomic.get();
    if (oldValue != value) {
        return false;
    }

    if (oldValue == 0L || atomic.compareAndSet(oldValue, 0L)) {
        // only remove after setting to zero, to avoid concurrent updates
        map.remove(key, atomic);
        // succeed even if the remove fails, since the value was already adjusted
        return true;
    }

    // value changed
    return false;
}
 

@Override
public long nextLong() {
  long oldseed, nextseed;
  AtomicLong seed = this._seed;
  do {
    oldseed = seed.get();
    nextseed = xorShift(oldseed);
  } while (!seed.compareAndSet(oldseed, nextseed));

  return nextseed;
}
 

/**
 * Associates {@code newValue} with {@code key} in this map, and returns the value previously
 * associated with {@code key}, or zero if there was no such value.
 */

@CanIgnoreReturnValue
public long put(K key, long newValue) {
  outer:
  while (true) {
    AtomicLong atomic = map.get(key);
    if (atomic == null) {
      atomic = map.putIfAbsent(key, new AtomicLong(newValue));
      if (atomic == null) {
        return 0L;
      }
      // atomic is now non-null; fall through
    }

    while (true) {
      long oldValue = atomic.get();
      if (oldValue == 0L) {
        // don't compareAndSet a zero
        if (map.replace(key, atomic, new AtomicLong(newValue))) {
          return 0L;
        }
        // atomic replaced
        continue outer;
      }
      if (atomic.compareAndSet(oldValue, newValue)) {
        return oldValue;
      }
      // value changed
    }
  }
}
 

protected void setBit(AtomicLong atomic, long lMask) {
    long lWord;
    do {
        lWord = atomic.get();
    } while (!atomic.compareAndSet(lWord, lWord | lMask));

    if ((atomic.get() & lMask) == 0L) {
        throw new InternalError();
    }
}
 

protected void setBit(AtomicLong atomic, long lMask) {
    long lWord;
    do {
        lWord = atomic.get();
    } while (!atomic.compareAndSet(lWord, lWord | lMask));

    if ((atomic.get() & lMask) == 0L) {
        throw new InternalError();
    }
}
 

/**
 * Associates {@code newValue} with {@code key} in this map, and returns the value previously
 * associated with {@code key}, or zero if there was no such value.
 */

@CanIgnoreReturnValue
public long put(K key, long newValue) {
  outer:
  while (true) {
    AtomicLong atomic = map.get(key);
    if (atomic == null) {
      atomic = map.putIfAbsent(key, new AtomicLong(newValue));
      if (atomic == null) {
        return 0L;
      }
      // atomic is now non-null; fall through
    }

    while (true) {
      long oldValue = atomic.get();
      if (oldValue == 0L) {
        // don't compareAndSet a zero
        if (map.replace(key, atomic, new AtomicLong(newValue))) {
          return 0L;
        }
        // atomic replaced
        continue outer;
      }
      if (atomic.compareAndSet(oldValue, newValue)) {
        return oldValue;
      }
      // value changed
    }
  }
}
 

protected void setBit(AtomicLong atomic, long lMask) {
    long lWord;
    do {
        lWord = atomic.get();
    } while (!atomic.compareAndSet(lWord, lWord | lMask));

    if ((atomic.get() & lMask) == 0L) {
        throw new InternalError();
    }
}
 

/**
 * Adds {@code delta} to the value currently associated with {@code key}, and returns the old
 * value.
 */

@CanIgnoreReturnValue
public long getAndAdd(K key, long delta) {
  outer:
  while (true) {
    AtomicLong atomic = map.get(key);
    if (atomic == null) {
      atomic = map.putIfAbsent(key, new AtomicLong(delta));
      if (atomic == null) {
        return 0L;
      }
      // atomic is now non-null; fall through
    }

    while (true) {
      long oldValue = atomic.get();
      if (oldValue == 0L) {
        // don't compareAndSet a zero
        if (map.replace(key, atomic, new AtomicLong(delta))) {
          return 0L;
        }
        // atomic replaced
        continue outer;
      }
      long newValue = oldValue + delta;
      if (atomic.compareAndSet(oldValue, newValue)) {
        return oldValue;
      }
      // value changed
    }
  }
}
 

@Override
public final boolean offer(final M value,
                           final long timeout,
                           final TimeUnit unit)
{
  Objects.requireNonNull(value);

  final AtomicLong headRef = _head;
  final AtomicLong tailRef = _tail;
  final int capacity = _capacity;

  while (true) {
    final long tail = tailRef.get();
    final long head = headRef.get();
    final long nextHead = head + 1;

    if (capacity <= nextHead - tail) {
      long sequence = _blocker.nextOfferSequence();

      if (capacity <= headRef.get() + 1 - tailRef.get()
          && ! _blocker.offerWait(sequence, timeout, unit)) {
        // retest the capacity after the sequence is allocated because of
        // wake timing
        
        return false;
      }
    }
    else if (headRef.compareAndSet(head, nextHead)) {
      // _ring.setLazy(head, value);
      _ring.set(head, value);

      return true;
    }
  }
}
 

/**
 * Adds {@code n} to {@code requested} and returns the value prior to
 * addition once the addition is successful (uses CAS semantics). If
 * overflows then sets {@code requested} field to {@code Long.MAX_VALUE}.
 * 
 * @param requested
 *            atomic field updater for a request count
 * @param n
 *            the number of requests to add to the requested count
 * @return requested value just prior to successful addition
 */
public static long getAndAddRequest(AtomicLong requested, long n) {
    // add n to field but check for overflow
    while (true) {
        long current = requested.get();
        long next = current + n;
        // check for overflow
        if (next < 0) {
            next = Long.MAX_VALUE;
        }
        if (requested.compareAndSet(current, next)) {
            return current;
        }
    }
}
 

public static boolean compareAndIncreaseOnly(final AtomicLong target, final long value) {
    long prev = target.get();
    while (value > prev) {
        boolean updated = target.compareAndSet(prev, value);
        if (updated)
            return true;

        prev = target.get();
    }

    return false;
}
 

protected void setBit(AtomicLong atomic, long lMask) {
    long lWord;
    do {
        lWord = atomic.get();
    } while (!atomic.compareAndSet(lWord, lWord | lMask));

    if ((atomic.get() & lMask) == 0L) {
        throw new InternalError();
    }
}
 

/**
 * If {@code (key, expectedOldValue)} is currently in the map, this method replaces
 * {@code expectedOldValue} with {@code newValue} and returns true; otherwise, this method
 * returns false.
 * <p/>
 * <p>If {@code expectedOldValue} is zero, this method will succeed if {@code (key, zero)}
 * is currently in the map, or if {@code key} is not in the map at all.
 */
boolean replace(K key, long expectedOldValue, long newValue) {
    if (expectedOldValue == 0L) {
        return putIfAbsent(key, newValue) == 0L;
    } else {
        AtomicLong atomic = map.get(key);
        return (atomic == null) ? false : atomic.compareAndSet(expectedOldValue, newValue);
    }
}
 

public static void compareAndSetMax(final AtomicLong target, final long value) {
    long prev = target.get();
    while (value > prev) {
        boolean updated = target.compareAndSet(prev, value);
        if (updated)
            break;

        prev = target.get();
    }
}
 

protected void clearBit(AtomicLong atomic, long lMask) {
    long lWord;
    do {
        lWord = atomic.get();
    } while (!atomic.compareAndSet(lWord, lWord & ~lMask));

    if ((atomic.get() & lMask) != 0L) {
        throw new InternalError();
    }
}
 

@Override
public final M poll(long timeout, TimeUnit unit)
{
  // final AtomicLong tailAllocRef = _tailAlloc;
  final AtomicLong headRef = _head;
  final AtomicLong tailRef = _tail;

  final ArrayRing<M> ring = _ring;

  final RingBlocker blocker = _blocker;

  while (true) {
    long tail = tailRef.get();
    final long head = headRef.get();

    M value;

    if (tail == head) {
      blocker.offerWake();

      if (timeout <= 0) {
        return null;
      }

      long pollSequence = blocker.nextPollSequence();

      if (tailRef.get() == headRef.get()
          && ! blocker.pollWait(pollSequence, timeout, unit)) {
        return null;
      }
    }
    else if ((value = ring.pollAndClear(tail)) != null) {
      if (tailRef.compareAndSet(tail, tail + 1)) {
        blocker.offerWake();

        return value;
      }
      else {
        ring.set(tail, value);
      }
    }
  }
}
 

/**
 * Generates the next pseudorandom number. Subclasses should
 * override this, as this is used by all other methods.
 *
 * <p>The general contract of {@code next} is that it returns an
 * {@code int} value and if the argument {@code bits} is between
 * {@code 1} and {@code 32} (inclusive), then that many low-order
 * bits of the returned value will be (approximately) independently
 * chosen bit values, each of which is (approximately) equally
 * likely to be {@code 0} or {@code 1}. The method {@code next} is
 * implemented by class {@code Random} by atomically updating the seed to
 *  <pre>{@code (seed * 0x5DEECE66DL + 0xBL) & ((1L << 48) - 1)}</pre>
 * and returning
 *  <pre>{@code (int)(seed >>> (48 - bits))}.</pre>
 *
 * This is a linear congruential pseudorandom number generator, as
 * defined by D. H. Lehmer and described by Donald E. Knuth in
 * <i>The Art of Computer Programming,</i> Volume 3:
 * <i>Seminumerical Algorithms</i>, section 3.2.1.
 *
 * @param  bits random bits
 * @return the next pseudorandom value from this random number
 *         generator's sequence
 * @since  1.1
 */
protected int next(int bits) {
    long oldseed, nextseed;
    AtomicLong seed = this.seed;
    do {
        oldseed = seed.get();
        nextseed = (oldseed * multiplier + addend) & mask;
    } while (!seed.compareAndSet(oldseed, nextseed));
    return (int)(nextseed >>> (48 - bits));
}