Безопасное удаление потока из-за тайм-аута

Question

У меня есть очередь задач, которые необходимо выполнить, и пул рабочих, которые выбирают задачи и выполняют их.Есть также класс «менеджер», который отслеживает работника, позволяет пользователю останавливать или перезапускать его, сообщает об их ходе и т. Д. Каждый работник делает что-то вроде этого:

public void doWork() {
    checkArguments();
    performCalculation();
    saveResultsToDatabase();
    performAnotherCalculation();
    saveResultsToDatabase();
    performYetAnotherCalculation();
    saveResultsToDatabase();
}

В этом случае,«база данных» не обязательно относится к базе данных Oracle.Это, безусловно, один из вариантов, но результаты также могут быть сохранены на диске, в Amazon SimpleDB и т. Д.

Пока все хорошо.Однако иногда код executeCalculation () блокируется периодически из-за множества факторов, но в основном из-за плохой реализации сетевого кода в куче сторонних библиотек (например, Socket.read () никогда не возвращается),Очевидно, что это плохо, потому что задача теперь застряла навсегда, а рабочий уже мертв.

Что я хотел бы сделать, это обернуть весь метод doWork () в какое-то время ожидания, и, если время ожидания истекло, передайте задание кому-то еще.

Как я могу это сделать?Допустим, исходный работник застрял в методе executeCalculation ().Затем я даю задание другому работнику, который его выполняет, а затем первоначальный работник решает проснуться и сохранить промежуточные результаты в базе данных ... тем самым испортив совершенно достоверные данные.Есть ли какой-то общий шаблон, который я могу использовать, чтобы избежать этого?

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

Answer 1

Вы пробовали использовать Future? Они полезны для запуска задачи и ожидания ее завершения с использованием таймаута и т. Д. Например:

private Runnable performCalc = new Runnable() { 
  public void run() {
    performCalculation();
  }
}

public void doWork() {
  try {
    ExecutorService executor = Executors.newFixedThreadPool(1);
    executor.submit(performCalc).get(); // Timeouts can be used here.
    executor.submit(anotherCalc).get();
  } catch(InterruptedException e) {
    // Asked to stop. Rollback out transactions.
  } catch(OtherExceptions here) {
  }
}

Answer 2

Если performCalculation застрял при блокировке ввода-вывода, вы мало что можете сделать, чтобы прервать его. Одно из решений состоит в том, чтобы закрыть базовый сокет или установить тайм-аут для операций с сокетом, используя Socket.setSoTimeout, но для этого необходимо иметь код, который читает из сокета.

В противном случае вы можете добавить некоторый механизм согласования перед сохранением данных в базе данных. Используйте какие-то временные метки, чтобы определить, являются ли данные в базе данных более новыми, чем данные, которые исходный работник извлек из сети.

Answer 3

Полагаю, проще всего было бы иметь отдельный поток таймера, запускаемый при запуске потока с executeCalculation (). Поток таймера может активироваться через определенный промежуток времени и Thread.interrupt() поток вычисления, который затем может выполнить любой необходимый откат при обработке исключения InterruptedException.

Конечно, это требует дополнительной сложности для управления другими проблемами и, следовательно, не самое элегантное решение.