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

private AtomicCounter fakeCounter()
{
    final AtomicLong value = new AtomicLong();
    final AtomicCounter atomicCounter = mock(AtomicCounter.class);
    final Answer<Long> get = inv -> value.get();
    final Answer<?> set = inv ->
    {
        value.set(inv.getArgument(0));
        return null;
    };

    final Answer<?> add = (inv) -> value.getAndAdd(inv.getArgument(0));

    when(atomicCounter.get()).then(get);
    when(atomicCounter.getWeak()).then(get);

    doAnswer(set).when(atomicCounter).set(anyLong());
    doAnswer(set).when(atomicCounter).setOrdered(anyLong());
    doAnswer(set).when(atomicCounter).setWeak(anyLong());

    when(atomicCounter.getAndAdd(anyLong())).then(add);
    when(atomicCounter.getAndAddOrdered(anyLong())).then(add);

    return atomicCounter;
}
 

/**
 * Add to the counter
 *
 * @param key
 * @param value
 */
public void add(K key, long value) {
    updateLock.lock();
    try {
        AtomicLong current = counts.get(key);
        
        if (current == null) {
            current = initCounter(key);
            numRecords.getAndIncrement();
        }
        
        current.getAndAdd(value);
        
    } finally {
        updateLock.unlock();
    }
}
 

@Override
public IDBlock getIDBlock(final int partition, final int idNamespace, Duration timeout) throws BackendException {
    //Delay artificially
    if (delayAcquisitionMS>0) {
        try {
            Thread.sleep(delayAcquisitionMS);
        } catch (InterruptedException e) {
            throw new TemporaryBackendException(e);
        }
    }
    Preconditions.checkArgument(partition>=0 && partition<=Integer.MAX_VALUE);
    Preconditions.checkArgument(idNamespace>=0 && idNamespace<=Integer.MAX_VALUE);
    Long p = (((long)partition)<<Integer.SIZE) + ((long)idNamespace);
    long size = blockSizer.getBlockSize(idNamespace);
    AtomicLong id = ids.get(p);
    if (id == null) {
        ids.putIfAbsent(p, new AtomicLong(1));
        id = ids.get(p);
        Preconditions.checkNotNull(id);
    }
    long lowerBound = id.getAndAdd(size);
    if (lowerBound >= blockSizeLimit) {
        throw new IDPoolExhaustedException("Reached partition limit: " + blockSizeLimit);
    }
    return new MockIDBlock(lowerBound,Math.min(size,blockSizeLimit-lowerBound));
}
 

/**
 * Adds steal count to pool stealCounter if it exists, and resets.
 */
final void transferStealCount(ForkJoinPool p) {
    AtomicLong sc;
    if (p != null && (sc = p.stealCounter) != null) {
        int s = nsteals;
        nsteals = 0;            // if negative, correct for overflow
        sc.getAndAdd((long)(s < 0 ? Integer.MAX_VALUE : s));
    }
}
 

/**
 * Adds steal count to pool stealCounter if it exists, and resets.
 */
final void transferStealCount(ForkJoinPool p) {
    AtomicLong sc;
    if (p != null && (sc = p.stealCounter) != null) {
        int s = nsteals;
        nsteals = 0;            // if negative, correct for overflow
        sc.getAndAdd((long)(s < 0 ? Integer.MAX_VALUE : s));
    }
}
 

/**
 * Adds steal count to pool stealCounter if it exists, and resets.
 */
final void transferStealCount(ForkJoinPool p) {
    AtomicLong sc;
    if (p != null && (sc = p.stealCounter) != null) {
        int s = nsteals;
        nsteals = 0;            // if negative, correct for overflow
        sc.getAndAdd((long)(s < 0 ? Integer.MAX_VALUE : s));
    }
}
 

/**
 * Adds steal count to pool stealCounter if it exists, and resets.
 */
final void transferStealCount(ForkJoinPool p) {
    AtomicLong sc;
    if (p != null && (sc = p.stealCounter) != null) {
        int s = nsteals;
        nsteals = 0;            // if negative, correct for overflow
        sc.getAndAdd((long)(s < 0 ? Integer.MAX_VALUE : s));
    }
}
 

/**
 * Set number of topics
 * @param val Value to set to
 */
public void updateNumTopics(long val) {
  long delta = val - _numTopics.getAndSet(val);
  AtomicLong aggregatedMetric = AGGREGATED_NUM_TOPICS.get(_className);
  if (aggregatedMetric != null) {
    aggregatedMetric.getAndAdd(delta);
  }
}
 

/**
 * Adds steal count to pool stealCounter if it exists, and resets.
 */
final void transferStealCount(ForkJoinPool p) {
    AtomicLong sc;
    if (p != null && (sc = p.stealCounter) != null) {
        int s = nsteals;
        nsteals = 0;            // if negative, correct for overflow
        sc.getAndAdd((long)(s < 0 ? Integer.MAX_VALUE : s));
    }
}
 

/**
 * Set number of auto paused partitions awaiting destination topic creation
 * @param val Value to set to
 */
public void updateNumAutoPausedPartitionsAwaitingDestTopic(long val) {
  long delta = val - _numAutoPausedPartitionsAwaitingDestTopic.getAndSet(val);
  AtomicLong aggregatedMetric = AGGREGATED_NUM_AUTO_PAUSED_PARTITIONS_WAITING_FOR_DEST_TOPIC.get(_className);
  if (aggregatedMetric != null) {
    aggregatedMetric.getAndAdd(delta);
  }
}
 

/**
 *
 * @throws Exception If failed.
 */
@Test
public void testUsedSpaceSize() throws Exception {
    final int DIRS_COUNT = 5;
    final int DIRS_MAX_DEEP = 3;
    final int FILES_COUNT = 10;
    final AtomicLong totalSize = new AtomicLong();

    IgniteBiInClosure<Integer, IgfsPath> createHierarchy = new IgniteBiInClosure<Integer, IgfsPath>() {
        @Override public void apply(Integer level, IgfsPath levelDir) {
            try {
                for (int i = 0; i < FILES_COUNT; ++i) {
                    IgfsPath filePath = new IgfsPath(levelDir, "file" + Integer.toString(i));

                    createFile(igfs, filePath, true, chunk);

                    totalSize.getAndAdd(chunk.length);
                }

                if (level < DIRS_MAX_DEEP) {
                    for (int dir = 0; dir < DIRS_COUNT; dir++) {
                        IgfsPath dirPath = new IgfsPath(levelDir, "dir" + Integer.toString(dir));

                        igfs.mkdirs(dirPath);

                        apply(level + 1, dirPath);
                    }
                }
            } catch (Exception e) {
                fail(e.getMessage());
            }
        }
    };

    createHierarchy.apply(1, new IgfsPath("/dir"));

    assertEquals(totalSize.get(), igfs.metrics().secondarySpaceSize());
}
 

/**
 * Adds steal count to pool stealCounter if it exists, and resets.
 */
final void transferStealCount(ForkJoinPool p) {
    AtomicLong sc;
    if (p != null && (sc = p.stealCounter) != null) {
        int s = nsteals;
        nsteals = 0;            // if negative, correct for overflow
        sc.getAndAdd((long)(s < 0 ? Integer.MAX_VALUE : s));
    }
}
 

/**
 * Set number of auto paused partitions on in-flight messages
 * @param val Value to set to
 */
public void updateNumAutoPausedPartitionsOnInFlightMessages(long val) {
  long delta = val - _numAutoPausedPartitionsOnInFlightMessages.getAndSet(val);
  AtomicLong aggregatedMetric = AGGREGATED_NUM_AUTO_PAUSED_PARTITIONS_ON_INFLIGHT_MESSAGES.get(_className);
  if (aggregatedMetric != null) {
    aggregatedMetric.getAndAdd(delta);
  }
}
 

/**
 * Adds steal count to pool stealCounter if it exists, and resets.
 */
final void transferStealCount(ForkJoinPool p) {
    AtomicLong sc;
    if (p != null && (sc = p.stealCounter) != null) {
        int s = nsteals;
        nsteals = 0;            // if negative, correct for overflow
        sc.getAndAdd((long)(s < 0 ? Integer.MAX_VALUE : s));
    }
}
 

/**
 * Adds steal count to pool stealCounter if it exists, and resets.
 */
final void transferStealCount(ForkJoinPool p) {
    AtomicLong sc;
    if (p != null && (sc = p.stealCounter) != null) {
        int s = nsteals;
        nsteals = 0;            // if negative, correct for overflow
        sc.getAndAdd((long)(s < 0 ? Integer.MAX_VALUE : s));
    }
}
 

/**
 * Adds steal count to pool stealCounter if it exists, and resets.
 */
final void transferStealCount(ForkJoinPool p) {
    AtomicLong sc;
    if (p != null && (sc = p.stealCounter) != null) {
        int s = nsteals;
        nsteals = 0;            // if negative, correct for overflow
        sc.getAndAdd((long)(s < 0 ? Integer.MAX_VALUE : s));
    }
}
 

public void updateNumPartitions(long val) {
  long delta = val - _numPartitions.getAndSet(val);
  AtomicLong aggregatedMetric = AGGREGATED_NUM_PARTITIONS.get(_className);
  if (aggregatedMetric != null) {
    aggregatedMetric.getAndAdd(delta);
  }
}
 

public void decrMonitorNum(MonitorNumType key,int num){
    AtomicLong atomicLong = monitorNumData.get(key);
    atomicLong.getAndAdd(-num);
}
 

@Test
public void test() {
    AtomicLong time = new AtomicLong(0);
    Clock wallClock = Clock.mock(time);
    Reservoir reservoir = new HdrBuilder(wallClock)
            .resetReservoirPeriodically(Duration.ofMillis(1000))
            .withBackgroundExecutor(MockExecutor.INSTANCE)
            .buildReservoir();

    reservoir.update(10);
    reservoir.update(20);
    Snapshot snapshot = reservoir.getSnapshot();
    assertEquals(10, snapshot.getMin());
    assertEquals(20, snapshot.getMax());

    time.getAndAdd(900); // 900
    reservoir.update(30);
    reservoir.update(40);
    snapshot = reservoir.getSnapshot();
    assertEquals(10, snapshot.getMin());
    assertEquals(40, snapshot.getMax());

    time.getAndAdd(99); // 999
    reservoir.update(8);
    reservoir.update(60);
    snapshot = reservoir.getSnapshot();
    assertEquals(8, snapshot.getMin());
    assertEquals(60, snapshot.getMax());

    time.getAndAdd(1); // 1000
    reservoir.update(70);
    reservoir.update(80);
    snapshot = reservoir.getSnapshot();
    assertEquals(70, snapshot.getMin());
    assertEquals(80, snapshot.getMax());

    time.getAndAdd(1001); // 2001
    reservoir.update(90);
    reservoir.update(100);
    snapshot = reservoir.getSnapshot();
    assertEquals(90, snapshot.getMin());
    assertEquals(100, snapshot.getMax());

    time.getAndAdd(1000); // 3001
    snapshot = reservoir.getSnapshot();
    assertEquals(0, snapshot.getMin());
    assertEquals(0, snapshot.getMax());

    time.getAndAdd(1); // 3002
    reservoir.update(42);
    snapshot = reservoir.getSnapshot();
    assertEquals(42, snapshot.getMin());
    assertEquals(42, snapshot.getMax());

    time.getAndAdd(2000); // 5002
    snapshot = reservoir.getSnapshot();
    assertEquals(0, snapshot.getMin());
    assertEquals(0, snapshot.getMax());
}
 

@Test
public void shouldCacheSnapshot() {
    AtomicLong time = new AtomicLong(System.currentTimeMillis());
    Clock wallClock = Clock.mock(time);
    Reservoir reservoir = new HdrBuilder(wallClock)
            .resetReservoirOnSnapshot()
            .withSnapshotCachingDuration(Duration.ofMillis(1000))
            .buildReservoir();

    reservoir.update(10);
    reservoir.update(20);
    Snapshot firstSnapshot = reservoir.getSnapshot();

    time.getAndAdd(900);
    reservoir.update(30);
    reservoir.update(40);
    Snapshot firstCachedSnapshot = reservoir.getSnapshot();
    assertSame(firstSnapshot, firstCachedSnapshot);
    assertEquals(10, firstCachedSnapshot.getMin());
    assertEquals(20, firstCachedSnapshot.getMax());

    time.getAndAdd(99);
    reservoir.update(50);
    reservoir.update(60);
    Snapshot secondCachedSnapshot = reservoir.getSnapshot();
    assertSame(firstSnapshot, secondCachedSnapshot);
    assertEquals(10, secondCachedSnapshot.getMin());
    assertEquals(20, secondCachedSnapshot.getMax());

    time.getAndAdd(1);
    reservoir.update(70);
    reservoir.update(80);
    Snapshot firstNewSnapshot = reservoir.getSnapshot();
    assertNotSame(firstSnapshot, firstNewSnapshot);
    assertEquals(30, firstNewSnapshot.getMin());
    assertEquals(80, firstNewSnapshot.getMax());

    time.getAndAdd(1001);
    reservoir.update(90);
    reservoir.update(100);
    Snapshot secondNewSnapshot = reservoir.getSnapshot();
    assertNotSame(firstNewSnapshot, secondNewSnapshot);
    assertEquals(90, secondNewSnapshot.getMin());
    assertEquals(100, secondNewSnapshot.getMax());
}