Использование InheritableThreadLocal с ThreadPoolExecutor - или - ThreadPoolExecutor, который не использует потоки повторно

Question

Я пытаюсь использовать как InheritableThreadLocal, так и ThreadPoolExecutor.

Это ломается, потому что ThreadPoolExecutor повторно использует потоки для каждого пула (в конце концов, это пул), то есть InheritableThreadLocal не работает, как ожидалось.Теперь эта проблема кажется мне очевидной, но ее было особенно сложно отследить.

Я использую InheritableThreadLocal, так что каждый из нескольких процессов верхнего уровня имеет свое собственное соединение с базой данных для себя и любых подпроцессов, которые оннерестится.Я не просто использую один общий пул соединений, потому что каждый процесс верхнего уровня будет выполнять многоэтапную работу со своим соединением перед фиксацией в базе данных и / или подготовкой множества PreparedStatements, которые используются снова и снова.

Я использую общий ThreadPoolExecutor между этими процессами верхнего уровня, потому что есть определенные варианты поведения, которые необходимо учитывать.Например, несмотря на то, что у меня может быть запущено 4 процесса верхнего уровня, у меня может быть только один процесс, записывающий данные в базу данных за раз (или система должна использовать какой-то другой общий ресурс).Так что я сделаю процесс верхнего уровня, создающий Runnable и отправляющий его на общий ThreadPoolExecutor, чтобы убедиться, что не более одного (или двух или трех, в зависимости от обстоятельств) работает одновременно во всемвся система.

Проблема в том, что поскольку ThreadPoolExecutor повторно использует свои потоки для пулов, InheritableThreadLocal выбирает исходное значение, которое было выполнено в этом пуле, а не значение, которое было впроцесс верхнего уровня, который отправил Runnable в ThreadPoolExecutor.

• Есть ли способ заставить рабочий пул в ThreadPoolExecutor использовать значение InheritableThreadLocal, которое было вконтекст процесса, который создал Runnable, а не в контексте повторно используемого пула потоков?

• Альтернативно, есть ли реализация ThreadPoolExecutor, которая создает новый поток каждый раз, когда онзапускает новый Runnable?Для моих целей я забочусь только о задании количества одновременно запущенных потоков фиксированного размера.

• Есть ли какие-либо другие решения или предложения, которые есть у меня, чтобы выполнить то, что я описал выше?

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

Существует предыдущий вопрос о StackOverflow, InheritableThreadLocal и пулах потоков , который также решает эту проблему.Тем не менее, решение этой проблемы, похоже, заключается в том, что это плохой вариант использования InheritableThreadLocal, который, я не думаю, применим к моей ситуации.

Спасибо за любые идеи.

Answer 1

Вместо использования ThreadPoolExecutor для защиты общих ресурсов, почему бы не использовать java.util.concurrent.Semaphore? Подзадачи, которые вы создаете, будут выполняться до конца в своих собственных потоках, но только после получения разрешения от семафора и, конечно же, освобождения разрешения по завершении.

Answer 2

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

Что касается ассоциируемого объекта Runnable с контекстом.Переопределите public void execute(Runnable command) для ExecutorPool и оберните Runnable в некоторый контекст, содержащий в первую очередь нужное вам значение из InheritedThreadLocal.

Класс обертки должен выглядеть примерно как

class WrappedRunnable extends Runnable{
  static final ThreadLocal<Ctx> context=new ThreadLocal<Ctx>();
  final Runnable target;
  final Ctx context;
  WrappedRunnable(Ctx context, Runnable target){...}

  public void run(){
    ctx.set(context);
    try{ 
      target.run();
    }finally{
      ctx.set(null);//or ctx.remove()
    }
  }
}

В качестве альтернативы, есть ли реализация ThreadPoolExecutor, которая создает новый> поток каждый раз, когда он запускает новый Runnable?Для моих целей я забочусь только о том, чтобы> фиксировать количество одновременно работающих потоков до фиксированного размера.

Хотя это очень плохо с точки зрения производительности, вы можете реализовать свои собственные, в основном вам нужно только execute(Runnable task) метод для Executor, который порождает новый поток и запускает его.

Answer 3

Почему бы просто не передать текущее соединение каким-либо подзадачам, созданным основной задачей? может быть, какой-то общий объект Context?

Answer 4

Ранее у нас была такая же проблема, и мы решили эту проблему, написав ThreadLocalContextMigrator, который в основном копирует локальный контекст потока в задачу, которая будет выполняться с использованием потока из пула. Задача при выполнении будет собирать больше контекстной информации, а по завершении задачи мы копируем ее обратно.