How to copy ThreadLocal from parent thread to multiple child threads?

Question

I am developing a support library that declares concurrent aggregation through annotations.

But I have a problem that is difficult to solve.

When there is a large amount of ThreadLocal used in a project, concurrent aggregation will not work, because the value of ThreadLocal is lost in multithreading.

For Example

public class RequestContext {
    private static ThreadLocal<Long> TENANT_ID = new ThreadLocal<>();

    public static Long getTenantId() {
        return TENANT_ID.get();
    }

    public static void setTenantId(Long tenantId) {
        TENANT_ID.set(tenantId);
    }

    public static void removeTenantId() {
        TENANT_ID.remove();
    }
}

@Service
public class HomepageServiceImpl implements HomepageService {
    @DataProvider("topMenu")
    @Override
    public List<Category> topMenu() {
        /* will be null */
        Long tenantId = RequestContext.getTenantId();
        Assert.notNull(tenantId,"tenantId must be not null");
        // ... The content hereafter will be omitted.
    }

    @DataProvider("postList")
    @Override
    public List<Post> postList() {
        /* will be null */
        Long tenantId = RequestContext.getTenantId();
        Assert.notNull(tenantId,"tenantId must be not null");
        // ... The content hereafter will be omitted.
    }

    @DataProvider("allFollowers")
    @Override
    public List<User> allFollowers() {
        /* will be null */
        Long tenantId = RequestContext.getTenantId();
        Assert.notNull(tenantId,"tenantId must be not null");
        // ... The content hereafter will be omitted.
    }
}

Concurrent aggregation query

@Test
public void testThreadLocal() throws Exception {
    try {
        RequestContext.setTenantId(10000L);
        Object result = dataBeanAggregateQueryFacade.get(null,
                new Function3<List<Category>, List<Post>, List<User>, Object>() {
            @Override
            public Object apply(
                    @DataConsumer("topMenu") List<Category> categories,
                    @DataConsumer("postList") List<Post> posts,
                    @DataConsumer("allFollowers") List<User> users) {
                return new Object[] {
                        categories,posts,users
                };
            }
        });
    } finally {
        RequestContext.removeTenantId();
    }
}

The following methods will be called in different threads.

• topMenu()
• postList()
• allFollowers()

What's the problem?

The problem is that the above three methods do not get tenantId.

java.lang.reflect.InvocationTargetException
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:483)
    at io.github.lvyahui8.spring.aggregate.service.impl.DataBeanAggregateQueryServiceImpl.get(DataBeanAggregateQueryServiceImpl.java:85)
    at io.github.lvyahui8.spring.aggregate.service.impl.DataBeanAggregateQueryServiceImpl.get(DataBeanAggregateQueryServiceImpl.java:47)
    at io.github.lvyahui8.spring.aggregate.service.impl.DataBeanAggregateQueryServiceImpl.lambda$getDependObjectMap$0(DataBeanAggregateQueryServiceImpl.java:112)
    at io.github.lvyahui8.spring.aggregate.service.impl.DataBeanAggregateQueryServiceImpl$$Lambda$197/1921553024.call(Unknown Source)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.IllegalArgumentException: tenantId must be not null
    at org.springframework.util.Assert.notNull(Assert.java:198)
    at io.github.lvyahui8.spring.example.service.impl.HomepageServiceImpl.postList(HomepageServiceImpl.java:36)
    ... 12 more

I have several solutions and I want to known is there is any better solution?

• Pass by parameter
• Replace ThreadLocal with InheritableThreadLocal
• Copy ThreadLocal through reflection

Here is all the code that can be executed. https://github.com/lvyahui8/spring-boot-data-aggregator/blob/master/spring-boot-data-aggregator-example/src/test/java/io/github/lvyahui8/spring/example/DataBeanAggregateQueryFacadeTest.java

Answer 1

Your problem is that you declared HomepageServiceImpl as a singleton. But to function, it requires tenantId which can take different values. So in fact, you need different instances of HomepageServiceImpl with a field tenantId set at the moment of creation. Then, it is handy simplify the invocation of dataBeanAggregateQueryFacade.get(): pass data array instead of lambda. Something like this:

public class HomepageServiceImpl implements HomepageService {
    final Long tenantId;
    public HomepageServiceImpl(Long tenantId) {
       Assert.notNull(tenantId,"tenantId must be not null");
       this.tenantId=tenantId;
    }

    @Override
    public List<Category> topMenu() {
       /* will not be null */
       ... this.tenantId...
    }

    // declare also postList and allFollowers

    public Object[] getLists() {
       return new Object[]{topMenu(), postList(), allFollowers()};
    }
}

@Test
public void testThreadLocal() throws Exception {
    HomepageServiceImpl homepage = new HomepageServiceImpl(10000L);
    Object[] lists = homepage.getLists();
    Object result = dataBeanAggregateQueryFacade.get(null, lists);
}

Answer 2

I noticed that you are creating an instance of ThreadPoolExecutor within the aggregateExecutorService bean. Using InheritableThreadLocal won't help you because ThreadPoolExecutor uses its own pool of threads. These threads don't inherit anything from the caller thread. How to use MDC with thread pools? shows how to copy over Threadlocal values from caller thread to the thread pool thread. MDC is a type of a ThreadLocal object. Basically you want to extend ThreadPoolExecutor and use that in the aggregateExecutorService bean