Propagating ThreadLocal su un nuovo thread recuperato da un ExecutorService

Question

Sto eseguendo un processo in un thread separato con un timeout, utilizzando un ExecutorService e un Future (codice di esempio here) (il thread "spawning" si svolge in un Aspetto AOP).

Ora, il thread principale è una richiesta Resteasy. Resteasy usa una o più variabili ThreadLocal per memorizzare alcune informazioni di contesto che ho bisogno di recuperare ad un certo punto nella mia chiamata al metodo Rest. Il problema è che poiché il thread Resteasy è in esecuzione in un nuovo thread, le variabili ThreadLocal sono arelost.

Quale sarebbe il modo migliore per "propagare" qualsiasi variabile ThreadLocal utilizzata da Resteasy sul nuovo thread? Sembra che Resteasy usi più di una variabile ThreadLocal per tenere traccia delle informazioni sul contesto e vorrei trasferire "ciecamente" tutte le informazioni sul nuovo thread.

Ho esaminato la sottoclasse ThreadPoolExecutor e utilizzando il metodo beforeExecute per passare il thread corrente al pool, ma non sono riuscito a trovare un modo per passare le variabili ThreadLocal al pool.

Qualche suggerimento?

Grazie

Answer 1

L'insieme di ThreadLocal casi associati con un filo sono tenuti in soci privati di ogni Thread. La tua unica possibilità di elencarli è di fare qualche riflessione sullo Thread; in questo modo, è possibile ignorare le restrizioni di accesso sui campi del thread.

volta che è possibile ottenere il set di ThreadLocal, è possibile copiare nei thread in background utilizzando i beforeExecute() e afterExecute() ganci di ThreadPoolExecutor, o creando un wrapper Runnable per le attività che intercetta la chiamata run() per impostare un disinserire il necessario ThreadLocal le istanze. In realtà, quest'ultima tecnica potrebbe funzionare meglio, dato che ti darebbe un posto conveniente per memorizzare i valori ThreadLocal nel momento in cui l'attività viene accodata.

---

Aggiornamento: Ecco un esempio più concreto del secondo approccio. Contrariamente alla mia descrizione originale, tutto ciò che è memorizzato nel wrapper è il thread chiamante, che viene interrogato quando l'attività viene eseguita.

static Runnable wrap(Runnable task) 
{ 
    Thread caller = Thread.currentThread(); 
    return() -> { 
    Iterable<ThreadLocal<?>> vars = copy(caller); 
    try { 
     task.run(); 
    } 
    finally { 
     for (ThreadLocal<?> var : vars) 
     var.remove(); 
    } 
    }; 
} 

/** 
* For each {@code ThreadLocal} in the specified thread, copy the thread's 
* value to the current thread. 
* 
* @param caller the calling thread 
* @return all of the {@code ThreadLocal} instances that are set on current thread 
*/ 
private static Collection<ThreadLocal<?>> copy(Thread caller) 
{ 
    /* Use a nasty bunch of reflection to do this. */ 
    throw new UnsupportedOperationException(); 
} 

Answer 2

quanto ho capito il problema, si può avere uno sguardo a InheritableThreadLocal che è destinata a passare ThreadLocal variabili dal contesto Parent Discussione a figlio contesto del thread

Answer 3

Se si guarda il codice ThreadLocal si può vedere :

public T get() { 
     Thread t = Thread.currentThread(); 
     ... 
    } 

thread corrente non può essere sovrascritto.

Possibili soluzioni:

1. Guardate Java 7 forcella/unirsi al meccanismo (ma penso che sia un brutto modo)

2. Guardate endorsed meccanismo per sovrascrivere ThreadLocal classe nella JVM.

3. Provare a riscrivere RESTEasy (è possibile utilizzare strumenti di refactoring nel vostro IDE per sostituire tutti gli usi ThreadLocal, è sembrare facile)

Answer 4

Sulla base di @erickson risposta Ho scritto questo codice. Funziona con inheritableThreadLocals. Crea un elenco di inheritableThreadLocals utilizzando lo stesso metodo utilizzato nel contructor Thread. Naturalmente uso la riflessione per fare questo. Inoltre ho scavalcato la classe executor.

public class MyThreadPoolExecutor extends ThreadPoolExecutor 
{ 
    @Override 
    public void execute(Runnable command) 
    { 
     super.execute(new Wrapped(command, Thread.currentThread())); 
    } 
} 

Wrapper:

private class Wrapped implements Runnable 
    { 
     private final Runnable task; 

     private final Thread caller; 

     public Wrapped(Runnable task, Thread caller) 
     { 
     this.task = task; 
     this.caller = caller; 
     } 

     public void run() 
     { 
     Iterable<ThreadLocal<?>> vars = null; 
     try 
     { 
      vars = copy(caller); 
     } 
     catch (Exception e) 
     { 
      throw new RuntimeException("error when coping Threads", e); 
     } 
     try { 
      task.run(); 
     } 
     finally { 
      for (ThreadLocal<?> var : vars) 
       var.remove(); 
     } 
     } 
    } 

metodo di copia:

public static Iterable<ThreadLocal<?>> copy(Thread caller) throws Exception 
    { 
     List<ThreadLocal<?>> threadLocals = new ArrayList<>(); 
     Field field = Thread.class.getDeclaredField("inheritableThreadLocals"); 
     field.setAccessible(true); 
     Object map = field.get(caller); 
     Field table = Class.forName("java.lang.ThreadLocal$ThreadLocalMap").getDeclaredField("table"); 
     table.setAccessible(true); 

     Method method = ThreadLocal.class 
       .getDeclaredMethod("createInheritedMap", Class.forName("java.lang.ThreadLocal$ThreadLocalMap")); 
     method.setAccessible(true); 
     Object o = method.invoke(null, map); 

     Field field2 = Thread.class.getDeclaredField("inheritableThreadLocals"); 
     field2.setAccessible(true); 
     field2.set(Thread.currentThread(), o); 

     Object tbl = table.get(o); 
     int length = Array.getLength(tbl); 
     for (int i = 0; i < length; i++) 
     { 
     Object entry = Array.get(tbl, i); 
     Object value = null; 
     if (entry != null) 
     { 
      Method referentField = Class.forName("java.lang.ThreadLocal$ThreadLocalMap$Entry").getMethod(
        "get"); 
      referentField.setAccessible(true); 
      value = referentField.invoke(entry); 
      threadLocals.add((ThreadLocal<?>) value); 
     } 
     } 
     return threadLocals; 
    } 

Answer 5

Ecco un esempio di passare il LocaleContext corrente nel filo genitore per il thread figlio attraversato da CompletableFuture [Di default usato ForkJoinPool].

Basta definire tutte le cose che si desidera eseguire in un thread secondario all'interno di un blocco Runnable. Quindi, quando il CompletableFuture esegue il blocco Runnable, è il thread secondario che ha il controllo e voilà hai la roba ThreadLocal del genitore impostata in Child ThreadLocal.

Il problema qui non è l'intero ThreadLocal viene copiato. Solo il LocaleContext è copiato. Poiché ThreadLocal è di accesso privato solo al Thread a cui appartiene anche utilizzando Reflection e sta cercando di ottenere e impostare in Child è tutto troppo roba stravagante che potrebbe portare a perdite di memoria o hit delle prestazioni.

Quindi se si conoscono i parametri che ti interessano da ThreadLocal, allora questa soluzione funziona in modo più pulito.

public void parentClassMethod(Request request) { 
     LocaleContext currentLocale = LocaleContextHolder.getLocaleContext(); 
     executeInChildThread(() -> { 
       LocaleContextHolder.setLocaleContext(currentLocale); 
       //Do whatever else you wanna do 
      })); 

     //Continue stuff you want to do with parent thread 
} 


private void executeInChildThread(Runnable runnable) { 
    try { 
     CompletableFuture.runAsync(runnable) 
      .get(); 
    } catch (Exception e) { 
     LOGGER.error("something is wrong"); 
    } 
} 

Answer 6

Non mi piace l'approccio di riflessione. La soluzione alternativa sarebbe quella di implementare wrapper executor e passare l'oggetto direttamente come contesto ThreadLocal a tutti i thread figli che propagano un contesto genitore.

public class PropagatedObject { 

    private ThreadLocal<ConcurrentHashMap<AbsorbedObjectType, Object>> data = new ThreadLocal<>(); 

    //put, set, merge methods, etc 

} 

==>

public class ObjectAwareExecutor extends AbstractExecutorService { 

    private final ExecutorService delegate; 
    private final PropagatedObject objectAbsorber; 

    public ObjectAwareExecutor(ExecutorService delegate, PropagatedObject objectAbsorber){ 
     this.delegate = delegate; 
     this.objectAbsorber = objectAbsorber; 
    } 
    @Override 
    public void execute(final Runnable command) { 

     final ConcurrentHashMap<String, Object> parentContext = objectAbsorber.get(); 
     delegate.execute(() -> { 
      try{ 
       objectAbsorber.set(parentContext); 
       command.run(); 
      }finally { 
       parentContext.putAll(objectAbsorber.get()); 
       objectAbsorber.clean(); 
      } 
     }); 
     objectAbsorber.merge(parentContext); 
    }