Java исправлено состояние гонки пула потоков? - PullRequest
Новая
       156

Java исправлено состояние гонки пула потоков?

2 голосов
/ 18 июня 2020

Рассмотрим следующий код: 

private static final Object LOCK = new Object();
private static final ExecutorService executorService = Executors.newFixedThreadPool(10); // Also used for a few other tasks.

public static void save(Object o) {
    String s = serialize(o);
    executorService.submit(() -> {
        // Do we have a race condition here?
        synchronized (LOCK) {
            saveToFile(s, new File("test.txt")); // overwrites the file
        }
    });
}

public static void main(String[] args) {
    save(args[0]);
    save(args[1]);
}

save(Object o) вызывается только в основном потоке. Я знаю, что пул потоков обрабатывает отправленные задачи из внутренней очереди по порядку, но может ли это случиться теоретически, что существует состояние гонки до достижения synchronized (LOCK), а вывод файла - args[0]?

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

Edit: Я думаю, что один из возможных способов - использовать Queue: 

private static final Queue<String> queue = new ConcurrentLinkedQueue<>();

public static void save(Object o) {
    queue.add(serialize(o));
    executorService.submit(() -> {
        synchronized (LOCK) {
            saveToFile(queue.remove(), new File("test.txt"));
        }
    });
}

1 Ответ

2 голосов
/ 18 июня 2020

В шаблоне производитель / потребитель, подобном вашему, вы обычно передаете задачи между (несколькими) производителями и (одним) потребителем через очередь блокировки . Тот факт, что существует единственный потребитель, обеспечивает порядок выполнения.

Таким образом, в псевдокоде это будет примерно так: 

val queue = ArrayBlockingQueue() //could be another BlockingQueue
val executor = ...

main {
  executor.submit(consumer)
  queue.put(args[0])
  queue.put(args[1])
}

consumer() {
  try {
    while (true) {
      val file = queue.take() //blocks until a file is added to the queue
      save(file)
    }
  } catch (InterruptedException e) {
    Thread.currentThread(interrupt()) //restore the interrupt flag
    //do nothing, just exit
  }
}

Если вы добавите больше потребителей, вы можете ' t гарантировать, что файлы будут обработаны по порядку. Но вы можете добавить столько производителей, сколько захотите (т.е. добавить в очередь из разных потоков).

...