Java 类java.util.concurrent.atomic.LongAccumulator 实例源码

public static void main(String[] args) throws Exception{
    LongAccumulator accumulator=new LongAccumulator(Long::max,Long.MIN_VALUE);
    Thread[] ts=new Thread[100];

    for(int i=0;i<100;i++){
        ts[i]=new Thread(()->{
            Random random=new Random();
            long value=random.nextLong();
            accumulator.accumulate(value);
        });
        ts[i].start();
    }

    for(int i=0;i<100;i++){
        ts[i].join();
    }
    System.out.println(accumulator.longValue());
}

public static void main(String[] args) throws Exception {
    LongAccumulator accumulator = new LongAccumulator(Long::max, Long.MIN_VALUE);
    Thread[] ts = new Thread[1000];
    for (int i = 0; i < 1000; i++) {
        ts[i] = new Thread(() -> {
            Random random = new Random();
            long value = random.nextLong();
            accumulator.accumulate(value);
        });
        ts[i].start();

    }
    for (int i = 0; i < 1000; i++) {
        ts[i].join();

    }
    System.out.println(accumulator.longValue());

}

/**
 * accumulates by multiple threads produce correct result
 */
public void testAccumulateAndGetMT() {
    final int incs = 1000000;
    final int nthreads = 4;
    final ExecutorService pool = Executors.newCachedThreadPool();
    LongAccumulator a = new LongAccumulator(Long::max, 0L);
    Phaser phaser = new Phaser(nthreads + 1);
    for (int i = 0; i < nthreads; ++i)
        pool.execute(new AccTask(a, phaser, incs));
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndAwaitAdvance();
    long expected = incs - 1;
    long result = a.get();
    assertEquals(expected, result);
    pool.shutdown();
}

public static void main(String[] args) throws Exception{
    LongAccumulator accumulator=new LongAccumulator(Long::max,Long.MIN_VALUE);
    Thread[] ts=new Thread[100];

    for(int i=0;i<100;i++){
        ts[i]=new Thread(()->{
            Random random=new Random();
            long value=random.nextLong();
            accumulator.accumulate(value);
        });
        ts[i].start();
    }

    for(int i=0;i<100;i++){
        ts[i].join();
    }
    System.out.println(accumulator.longValue());
}

static void testLongAccumulator() {
    LongBinaryOperator plus = (LongBinaryOperator & Serializable) (x, y) -> x + y;
    LongAccumulator a = new LongAccumulator(plus, -2);
    a.accumulate(34);
    LongAccumulator result = echo(a);
    if (result.get() != a.get())
        throw new RuntimeException("Unexpected value");
    a.reset();
    result.reset();
    if (result.get() != a.get())
        throw new RuntimeException("Unexpected value after reset");

    checkSerialClassName(a, "java.util.concurrent.atomic.LongAccumulator$SerializationProxy");
}

static void testLongAccumulator() {
    LongBinaryOperator plus = (LongBinaryOperator & Serializable) (x, y) -> x + y;
    LongAccumulator a = new LongAccumulator(plus, -2);
    a.accumulate(34);
    LongAccumulator result = echo(a);
    if (result.get() != a.get())
        throw new RuntimeException("Unexpected value");
    a.reset();
    result.reset();
    if (result.get() != a.get())
        throw new RuntimeException("Unexpected value after reset");

    checkSerialClassName(a, "java.util.concurrent.atomic.LongAccumulator$SerializationProxy");
}

/**
 * accumulate accumulates given value to current, and get returns current value
 */
public void testAccumulateAndGet() {
    LongAccumulator acc = new LongAccumulator(Long::max, 0L);
    acc.accumulate(2);
    assertEquals(2, acc.get());
    acc.accumulate(-4);
    assertEquals(2, acc.get());
    acc.accumulate(4);
    assertEquals(4, acc.get());
}

/**
 * reset() causes subsequent get() to return zero
 */
public void testReset() {
    LongAccumulator acc = new LongAccumulator(Long::max, 0L);
    acc.accumulate(2);
    assertEquals(2, acc.get());
    acc.reset();
    assertEquals(0, acc.get());
}

/**
 * getThenReset() returns current value; subsequent get() returns zero
 */
public void testGetThenReset() {
    LongAccumulator acc = new LongAccumulator(Long::max, 0L);
    acc.accumulate(2);
    assertEquals(2, acc.get());
    assertEquals(2, acc.getThenReset());
    assertEquals(0, acc.get());
}

/**
 * toString returns current value.
 */
public void testToString() {
    LongAccumulator acc = new LongAccumulator(Long::max, 0L);
    assertEquals("0", acc.toString());
    acc.accumulate(1);
    assertEquals(Long.toString(1), acc.toString());
}

/**
 * intValue returns current value.
 */
public void testIntValue() {
    LongAccumulator acc = new LongAccumulator(Long::max, 0L);
    assertEquals(0, acc.intValue());
    acc.accumulate(1);
    assertEquals(1, acc.intValue());
}

/**
 * longValue returns current value.
 */
public void testLongValue() {
    LongAccumulator acc = new LongAccumulator(Long::max, 0L);
    assertEquals(0, acc.longValue());
    acc.accumulate(1);
    assertEquals(1, acc.longValue());
}

/**
 * floatValue returns current value.
 */
public void testFloatValue() {
    LongAccumulator acc = new LongAccumulator(Long::max, 0L);
    assertEquals(0.0f, acc.floatValue());
    acc.accumulate(1);
    assertEquals(1.0f, acc.floatValue());
}

/**
 * doubleValue returns current value.
 */
public void testDoubleValue() {
    LongAccumulator acc = new LongAccumulator(Long::max, 0L);
    assertEquals(0.0, acc.doubleValue());
    acc.accumulate(1);
    assertEquals(1.0, acc.doubleValue());
}

static void testLongAccumulator() {
    LongBinaryOperator plus = (LongBinaryOperator & Serializable) (x, y) -> x + y;
    LongAccumulator a = new LongAccumulator(plus, -2);
    a.accumulate(34);
    LongAccumulator result = echo(a);
    if (result.get() != a.get())
        throw new RuntimeException("Unexpected value");
    a.reset();
    result.reset();
    if (result.get() != a.get())
        throw new RuntimeException("Unexpected value after reset");

    checkSerialClassName(a, "java.util.concurrent.atomic.LongAccumulator$SerializationProxy");
}

/**
 * accumulate accumulates given value to current, and get returns current value
 */
public void testAccumulateAndGet() {
    LongAccumulator ai = new LongAccumulator(Long::max, 0L);
    ai.accumulate(2);
    assertEquals(2, ai.get());
    ai.accumulate(-4);
    assertEquals(2, ai.get());
    ai.accumulate(4);
    assertEquals(4, ai.get());
}

/**
 * reset() causes subsequent get() to return zero
 */
public void testReset() {
    LongAccumulator ai = new LongAccumulator(Long::max, 0L);
    ai.accumulate(2);
    assertEquals(2, ai.get());
    ai.reset();
    assertEquals(0, ai.get());
}

/**
 * getThenReset() returns current value; subsequent get() returns zero
 */
public void testGetThenReset() {
    LongAccumulator ai = new LongAccumulator(Long::max, 0L);
    ai.accumulate(2);
    assertEquals(2, ai.get());
    assertEquals(2, ai.getThenReset());
    assertEquals(0, ai.get());
}

/**
 * toString returns current value.
 */
public void testToString() {
    LongAccumulator ai = new LongAccumulator(Long::max, 0L);
    assertEquals("0", ai.toString());
    ai.accumulate(1);
    assertEquals(Long.toString(1), ai.toString());
}

/**
 * intValue returns current value.
 */
public void testIntValue() {
    LongAccumulator ai = new LongAccumulator(Long::max, 0L);
    assertEquals(0, ai.intValue());
    ai.accumulate(1);
    assertEquals(1, ai.intValue());
}

/**
 * longValue returns current value.
 */
public void testLongValue() {
    LongAccumulator ai = new LongAccumulator(Long::max, 0L);
    assertEquals(0, ai.longValue());
    ai.accumulate(1);
    assertEquals(1, ai.longValue());
}

/**
 * floatValue returns current value.
 */
public void testFloatValue() {
    LongAccumulator ai = new LongAccumulator(Long::max, 0L);
    assertEquals(0.0f, ai.floatValue());
    ai.accumulate(1);
    assertEquals(1.0f, ai.floatValue());
}

/**
 * doubleValue returns current value.
 */
public void testDoubleValue() {
    LongAccumulator ai = new LongAccumulator(Long::max, 0L);
    assertEquals(0.0, ai.doubleValue());
    ai.accumulate(1);
    assertEquals(1.0, ai.doubleValue());
}

public void run() {
    phaser.arriveAndAwaitAdvance();
    LongAccumulator a = acc;
    for (int i = 0; i < incs; ++i)
        a.accumulate(i);
    result = a.get();
    phaser.arrive();
}

static void testLongAccumulator() {
    LongBinaryOperator plus = (LongBinaryOperator & Serializable) (x, y) -> x + y;
    LongAccumulator a = new LongAccumulator(plus, -2);
    a.accumulate(34);
    LongAccumulator result = echo(a);
    if (result.get() != a.get())
        throw new RuntimeException("Unexpected value");
    a.reset();
    result.reset();
    if (result.get() != a.get())
        throw new RuntimeException("Unexpected value after reset");

    checkSerialClassName(a, "java.util.concurrent.atomic.LongAccumulator$SerializationProxy");
}

static void testLongAccumulator() {
    LongBinaryOperator plus = (LongBinaryOperator & Serializable) (x, y) -> x + y;
    LongAccumulator a = new LongAccumulator(plus, -2);
    a.accumulate(34);
    LongAccumulator result = echo(a);
    if (result.get() != a.get())
        throw new RuntimeException("Unexpected value");
    a.reset();
    result.reset();
    if (result.get() != a.get())
        throw new RuntimeException("Unexpected value after reset");

    checkSerialClassName(a, "java.util.concurrent.atomic.LongAccumulator$SerializationProxy");
}

/**
 * Use synchronized keyword and java.util.concurrent.atomic package to
 * control the order of thread execution
 */
synchronized private void synchronizedExample() {
    number = number + number;

    // Atomic
    AtomicInteger i = new AtomicInteger();
    i.incrementAndGet();
    i.getAndIncrement();

    i.decrementAndGet();
    i.getAndDecrement();

    i.getAndAdd(10);
    i.addAndGet(10);

    i.updateAndGet(v -> v * 2);

    i = new AtomicInteger(10);
    i.accumulateAndGet(10, (a, b) -> a * b);
    System.out.print("AtomicInteger : ");
    System.out.println(i);

    AtomicBoolean ab;
    AtomicInteger ai;
    DoubleAdder da;
    LongAdder la;
    DoubleAccumulator dac;
    LongAccumulator lac;
}

static void testLongAccumulator() {
    LongBinaryOperator plus = (LongBinaryOperator & Serializable) (x, y) -> x + y;
    LongAccumulator a = new LongAccumulator(plus, -2);
    a.accumulate(34);
    LongAccumulator result = echo(a);
    if (result.get() != a.get())
        throw new RuntimeException("Unexpected value");
    a.reset();
    result.reset();
    if (result.get() != a.get())
        throw new RuntimeException("Unexpected value after reset");

    checkSerialClassName(a, "java.util.concurrent.atomic.LongAccumulator$SerializationProxy");
}

static void testLongAccumulator() {
    LongBinaryOperator plus = (LongBinaryOperator & Serializable) (x, y) -> x + y;
    LongAccumulator a = new LongAccumulator(plus, -2);
    a.accumulate(34);
    LongAccumulator result = echo(a);
    if (result.get() != a.get())
        throw new RuntimeException("Unexpected value");
    a.reset();
    result.reset();
    if (result.get() != a.get())
        throw new RuntimeException("Unexpected value after reset");

    checkSerialClassName(a, "java.util.concurrent.atomic.LongAccumulator$SerializationProxy");
}

static void testLongAccumulator() {
    LongBinaryOperator plus = (LongBinaryOperator & Serializable) (x, y) -> x + y;
    LongAccumulator a = new LongAccumulator(plus, -2);
    a.accumulate(34);
    LongAccumulator result = echo(a);
    if (result.get() != a.get())
        throw new RuntimeException("Unexpected value");
    a.reset();
    result.reset();
    if (result.get() != a.get())
        throw new RuntimeException("Unexpected value after reset");

    checkSerialClassName(a, "java.util.concurrent.atomic.LongAccumulator$SerializationProxy");
}

/**
 * Provide lambda expressions for the peek() operations that enable you to detect
 * whether the stream is running in parallel, and using this information, provide
 * expressions for the stream1isParallel and stream2isParallel booleans to make
 * the assertions correct. You may also provide additional declarations
 * and statements anywhere before assertions. (There are an open-ended number of
 * solutions for this; the solutions file contains only one example.) Race conditions
 * will be tolerated if you're clever.
 */
@Test
public void ex27_parallelVsSequential() {
    //UNCOMMENT//IntConsumer ic1 = i -> { }; // TODO
    //UNCOMMENT//IntConsumer ic2 = i -> { }; // TODO
    //BEGINREMOVE
    LongAccumulator adder1 = new LongAccumulator((x, y) -> (x << 1) + y, 0L);
    LongAccumulator adder2 = new LongAccumulator((x, y) -> (x << 1) + y, 0L);
    IntConsumer ic1 = i -> adder1.accumulate(i);
    IntConsumer ic2 = i -> adder2.accumulate(i);
    //ENDREMOVE

    List<Integer> result1 = IntStream.range(0, 100)
                                     .peek(ic1)
                                     .boxed()
                                     .collect(Collectors.toList());

    List<Integer> result2 = IntStream.range(0, 100)
                                     .parallel()
                                     .peek(ic2)
                                     .boxed()
                                     .collect(Collectors.toList());

    //UNCOMMENT//boolean stream1isParallel = false; // TODO
    //UNCOMMENT//boolean stream2isParallel = false; // TODO
    //BEGINREMOVE
    boolean stream1isParallel = adder1.longValue() != -101L;
    boolean stream2isParallel = adder2.longValue() != -101L;
    //ENDREMOVE

    assertEquals(result1, result2);
    assertFalse(stream1isParallel);
    assertTrue(stream2isParallel);
}

@Theory
public void test(int[] fixture) {
    LongAccumulator la = LongAccumulators.max();
    la.accumulate(fixture[0]);
    la.accumulate(fixture[1]);
    assertEquals(fixture[2], la.get());
}

@Theory
public void test(int[] fixture) {
    LongAccumulator la = LongAccumulators.min();
    la.accumulate(fixture[0]);
    la.accumulate(fixture[1]);
    assertEquals(fixture[2], la.get());
}

private static int getTypeMark(final Object o1) {
    final Class<?> cls = o1.getClass();
    if (cls == String.class) {
        return STRING;
    } else if (cls == Integer.class) {
        return INTEGER;
    } else if (cls == Long.class) {
        return LONG;
    } else if (cls == Short.class) {
        return SHORT;
    } else if (cls == Double.class) {
        return DOUBLE;
    } else if (cls == Float.class) {
        return FLOAT;
    } else if (cls == Character.class) {
        return CHAR;
    } else if (cls == Byte.class) {
        return BYTE;
    } else if (o1 instanceof Number) {
        if (o1 instanceof BigInteger) {
            return BIG_INTEGER;
        } else if (o1 instanceof BigDecimal) {
            return BIG_DECIMAL;
        } else if (o1 instanceof AtomicInteger) {
            return INTEGER;
        } else if (o1 instanceof AtomicLong
                || o1 instanceof LongAdder
                || o1 instanceof LongAccumulator) {
            return LONG;
        } else if (o1 instanceof DoubleAdder) {
            return DOUBLE;
        } else {
            //Note: otherwise, treat as BigDecimal
            return BIG_DECIMAL;
        }
    }
    return OBJECT;
}

/**
 * new instance initialized to supplied identity
 */
public void testConstructor() {
    for (long identity : new long[] { Long.MIN_VALUE, 0, Long.MAX_VALUE })
        assertEquals(identity,
                     new LongAccumulator(Long::max, identity).get());
}

/**
 * default constructed initializes to zero
 */
public void testConstructor() {
    LongAccumulator ai = new LongAccumulator(Long::max, 0L);
    assertEquals(0, ai.get());
}

AccTask(LongAccumulator acc, Phaser phaser, int incs) {
    this.acc = acc;
    this.phaser = phaser;
    this.incs = incs;
}

/**
 * Test many threads updating the reservoir.
 */
@Test
public void testManyThreads() throws Exception {
    final ExecutorService pool = Executors.newWorkStealingPool(4);

    // last possible bucket position according to current configuration
    final long lastBucket = THREAD_COUNT * (SAMPLE_SIZE / CLOCK_INTERVAL) - SIZE;

    for (long iteration = 0L; iteration < ITERATIONS; iteration++) {
        final Random random = new Random(0x1234123412341234L + iteration);

        final DeterministicClock clock = new DeterministicClock();

        final Reservoir delegate = new Reservoir() {
            @Override
            public int size() {
                return 0;
            }

            @Override
            public void update(final long value) {
            }

            @Override
            public Snapshot getSnapshot() {
                return DELEGATE_SNAPSHOT;
            }
        };

        final MinMaxSlidingTimeReservoir reservoir =
            new MinMaxSlidingTimeReservoir(clock, SIZE, STEP, TimeUnit.NANOSECONDS, delegate);

        final LongAccumulator min = new LongAccumulator(Math::min, Long.MAX_VALUE);
        final LongAccumulator max = new LongAccumulator(Math::max, Long.MIN_VALUE);

        final CountDownLatch latch = new CountDownLatch(THREAD_COUNT);

        for (int i = 0; i < THREAD_COUNT; i++) {
            pool.execute(() -> {
                for (int s = 0; s < SAMPLE_SIZE; s++) {
                    final long sample = random.nextLong();

                    if (s % CLOCK_INTERVAL == 0) {
                        clock.add(STEP);
                    }

                    // check if first bucket according to the clock is after the last possible
                    // bucket. if so, they should be taken into account.
                    if ((reservoir.calculateFirstBucket() + SIZE) > lastBucket) {
                        // start accumulating for reference comparison
                        min.accumulate(sample);
                        max.accumulate(sample);
                    }

                    reservoir.update(sample);
                }

                latch.countDown();
            });
        }

        // wait for all threads to complete
        latch.await();

        final Snapshot snapshot = reservoir.getSnapshot();

        assertArrayEquals("expected snapshot for iteration #" + iteration,
            new long[]{min.get(), 1, max.get()}, snapshot.getValues());

        assertEquals("expected max for iteration #" + iteration, max.get(), snapshot.getMax());
        assertEquals("expected min for iteration #" + iteration, min.get(), snapshot.getMin());
    }

    pool.shutdown();
}

@Before
public void setup() {
    maxRingBufferPending =  new LongAccumulator(Long::max, Long.MIN_VALUE);
    droppedCount = new AtomicLong(0);
    producerExecutor = Executors.newSingleThreadExecutor();
}

static LongAccumulator max() {
    return new LongAccumulator((x, y) -> x < y ? y : x, 0);
}