Можно безопасно увеличить BigInteger потоковым безопасным способом, возможно, с AtomicReference, без блокировки?

Много нашего кода устарело, но мы переходим к "Большим данным", и я пытаюсь разъяснять новые вызовы API, поощрять использование последних библиотек Spring и т.д. Один из наших проблемы - это генерация идентификатора прикладного уровня. По причинам, которые я не понимаю, высший орган требует последовательных BigInteger. Я бы сделал их случайными с повторным созданием и повторил попытку неудачных вставок, но я получил право вето.

Ворча в сторону, я в такой ситуации, когда мне нужно увеличивать и получать BigInteger по потокам и делать это безопасно и качественно. Я никогда раньше не использовал AtomicReference, но он выглядит довольно близок к совершенству для этого приложения. Прямо сейчас у нас есть синхронизированный блок кода, который сильно ухудшает нашу производительность.

Это правильный путь? Примеры синтаксиса?

Я должен упомянуть, что, как работает этот модуль, он попадает в базу данных с помощью хранимой процедуры, чтобы захватить диапазон значений для использования. Десятки тысяч за раз, так что это происходит только один раз в 20 минут. Это заставляет разные серверы наступать друг на друга, но также добавляет морщину необходимости устанавливать BigInteger на произвольно последующее значение. Конечно, это также должно быть потокобезопасным.

P.S. Я по-прежнему считаю, что моя идея о случайном поколении лучше, чем обработка всего этого потока. BigInteger - это смехотворно большое количество, и вероятность когда-либо генерировать одну и ту же секунду должна быть близка к нулю.

Ответ 1

В AtomicReference можно использовать быстрый черновик:

public final class AtomicBigInteger {

    private final AtomicReference<BigInteger> valueHolder = new AtomicReference<>();

    public AtomicBigInteger(BigInteger bigInteger) {
        valueHolder.set(bigInteger);
    }

    public BigInteger incrementAndGet() {
        for (; ; ) {
            BigInteger current = valueHolder.get();
            BigInteger next = current.add(BigInteger.ONE);
            if (valueHolder.compareAndSet(current, next)) {
                return next;
            }
        }
    }
}

Это в основном копия кода AtomicLong для incrementAndGet()

Ответ 2

Это становится более управляемым и понятным с помощью accumulateAndGet или getAndAccumulate, введенный в Java 8. Это позволяет вам атомизировать обновление значения, предоставляя функцию аккумулятора, которая устанавливает значение в результат функции, а также возвращает предыдущий или вычисленный результат в зависимости от того, что вы необходимость. Вот пример того, как может выглядеть этот класс, за которым следует простой пример, который я написал, который использует его:

import java.math.BigInteger;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicReference;

public final class AtomicBigInteger {

  private final AtomicReference<BigInteger> bigInteger;

  public AtomicBigInteger(final BigInteger bigInteger) {
    this.bigInteger = new AtomicReference<>(Objects.requireNonNull(bigInteger));
  }

  // Method references left out for demonstration purposes
  public BigInteger incrementAndGet() {
    return bigInteger.accumulateAndGet(BigInteger.ONE, (previous, x) -> previous.add(x));
  }

  public BigInteger getAndIncrement() {
    return bigInteger.getAndAccumulate(BigInteger.ONE, (previous, x) -> previous.add(x));
  }

  public BigInteger get() {
    return bigInteger.get();
  }
}

Пример использования:

import java.math.BigInteger;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class ABIExample {

  private static final int AVAILABLE_PROCS = Runtime.getRuntime().availableProcessors();
  private static final int INCREMENT_AMOUNT = 2_500_000;
  private static final int TASK_AMOUNT = AVAILABLE_PROCS * 2;
  private static final BigInteger EXPECTED_VALUE = BigInteger.valueOf(INCREMENT_AMOUNT)
                                                             .multiply(BigInteger
                                                                           .valueOf(TASK_AMOUNT));

  public static void main(String[] args)
      throws InterruptedException, ExecutionException {
    System.out.println("Available processors: " + AVAILABLE_PROCS);


    final ExecutorService executorService = Executors
        .newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    final AtomicBigInteger atomicBigInteger = new AtomicBigInteger(BigInteger.ZERO);

    final List<Callable<Void>> incrementTasks =  IntStream.rangeClosed(1, TASK_AMOUNT)
             .mapToObj(i -> incrementTask(i, atomicBigInteger))
             .collect(Collectors.toList());
    final List<Future<Void>> futures = executorService.invokeAll(incrementTasks);
    for (Future<Void> future : futures) {
      future.get();
    }
    executorService.shutdown();
    executorService.awaitTermination(30, TimeUnit.SECONDS);
    System.out.println("Final value: " + atomicBigInteger.get());
    final boolean areEqual = EXPECTED_VALUE.equals(atomicBigInteger.get());
    System.out.println("Does final value equal expected? - " + areEqual);
  }

  private static Callable<Void> incrementTask(
      final int taskNumber,
      final AtomicBigInteger atomicBigInteger
  ) {
    return () -> {
      for (int increment = 0; increment < INCREMENT_AMOUNT; increment++) {
        atomicBigInteger.incrementAndGet();
      }
      System.out.println("Task #" + taskNumber + " Completed");
      return null;
    };

  }
}

И результат выполнения примера на моей машине:

Available processors: 8
Task #3 Completed
Task #8 Completed
Task #7 Completed
Task #6 Completed
Task #5 Completed
Task #2 Completed
Task #4 Completed
Task #1 Completed
Task #9 Completed
Task #10 Completed
Task #11 Completed
Task #13 Completed
Task #16 Completed
Task #12 Completed
Task #14 Completed
Task #15 Completed
Final value: 80000000
Does final value equal expected? - true

Ответ 3

Это в основном пытается снова и снова, пока операция не будет атомарной.

Лично мне не нравится этот код, так как теоретически это может привести к голоданию нити (хотя тот, кто показал это мне, утверждает, что этого не происходит)

private AtomicReference<BigInteger> ref = new AtomicReference<BigInteger>(BigInteger.ZERO);

public BigInteger incrementAndGet() {
    BigInteger currVal, newVal;
    do {
        currVal = ref.get();
        newVal = currVal.clone();
        newVal.add(BigInteger.ONE);
    } while (!ref.compareAndSet(currVal, newVal));

}

Я бы пошел с AtomicLong, если это возможно.