扩展java的ThreadLocal以允许在所有线程中重置值

Question

看了这个问题之后,我想我想要包装ThreadLocal来添加重置行为.

我想要一个类似于ThreadLocal的东西,我可以从任何线程调用一个方法将所有值设置回相同的值.到目前为止我有这个:

public class ThreadLocalFlag {

    private ThreadLocal<Boolean> flag;
    private List<Boolean> allValues = new ArrayList<Boolean>();

    public ThreadLocalFlag() {
        flag = new ThreadLocal<Boolean>() {
            @Override protected Boolean initialValue() {
                Boolean value = false;
                allValues.add(value);
                return value;
            }
        };
    }

    public boolean get() {
        return flag.get();
    }

    public void set(Boolean value) {
        flag.set(value);
    }

    public void setAll(Boolean value) {
        for (Boolean tlValue : allValues) {
            tlValue = value;
        }
    }
}

我担心原语的自动装箱可能意味着我存储在列表中的副本将不会引用ThreadLocal引用的相同变量(如果我尝试设置它们).我还没有测试过这段代码,并且在我继续沿着这条路走下去之前,我正在寻找一些专家建议.

有人会问"你为什么这样做？".我正在一个框架中工作,其他线程回调到我的代码中,我没有对它们的引用.我想定期更新他们使用的ThreadLocal变量中的值,因此执行该更新需要使用该变量的线程进行更新.我只需要一种方法来通知所有这些线程他们的ThreadLocal变量是陈旧的.

---

我很高兴最近对这个三年前的问题有了新的批评,尽管我觉得它的语气比专业还要小.我提供的解决方案在此期间没有发生任何事故.然而,必然有更好的方法来实现提出这个问题的目标,我邀请评论家提供明显更好的答案.为此,我将尝试更清楚地解决我试图解决的问题.

正如我之前提到的,我正在使用一个框架,其中多个线程正在使用我的代码,在我的控制之外.该框架是QuickFIX/J,我正在实现Application接口.该接口定义了用于处理FIX消息的钩子,在我的使用中,框架被配置为多线程,因此可以同时处理与应用程序的每个FIX连接.

但是,QuickFIX/J框架仅为所有线程使用该接口的单个​​实例实例.我无法控制线程如何启动,并且每个都在为不同的连接提供不同的配置细节和其他状态.很自然地让一些经常访问但很少更新的状态存在于ThreadLocal框架启动线程后加载其初始值的各种s中.

在组织的其他地方,我们有库代码,允许我们注册回调,以通知在运行时更改的配置详细信息.我想注册那个回调,当我收到它时,我想让所有的线程知道是时候重新加载那些ThreadLocals 的值,因为它们可能已经改变了.该回调来自我无法控制的线程,就像QuickFIX/J线程一样.

我在下面的解决方案仅使用ThreadLocalFlag(一个包装ThreadLocal<AtomicBoolean>)来向其他线程发出信号,告知它可能是时候更新它们的值.回调调用setAll(true),以及QuickFIX/J线程set(false)在开始更新时调用.我已经淡化了并发问题,ArrayList因为列表添加到的唯一时间是在启动期间,并且我的用例小于列表的默认大小.

我想可以用其他的线程通信技术完成同样的任务,但是对于它正在做的事情,这看起来更实用.我欢迎其他解决方案.

Answer 1

跨线程与ThreadLocal中的对象进行交互

我会事先说这是一个坏主意.ThreadLocal是一个特殊的类,如果使用正确,它提供速度和线程安全的好处.尝试跨线程进行通信,首先会ThreadLocal破坏使用该类的目的.

如果您需要跨多个线程访问对象,则有为此目的设计的工具,特别是java.util.collect.concurrent诸如的线程安全集合ConcurrentHashMap,您可以使用ThreadLocal这些Thread集合作为键来复制a ,如下所示:

ConcurrentHashMap<Thread, AtomicBoolean> map = new ConcurrentHashMap<>();

// pass map to threads, let them do work, using Thread.currentThread() as the key

// Update all known thread's flags
for(AtomicBoolean b : map.values()) {
    b.set(true);
}

更清晰,更简洁,并避免使用ThreadLocal它的设计方式.

通知线程他们的数据是陈旧的

我只需要一种方法来通知所有这些线程他们的ThreadLocal变量是陈旧的.

如果您的目标只是通知其他线程某些内容已发生变化,那么根本不需要ThreadLocal.只需使用一个AtomicBoolean并与您共享所有任务,就像您一样ThreadLocal<AtomicBoolean>.顾名思义AtomicBoolean是对原子和可见交叉线程的更新.更好的方法是使用真正的同步辅助工具,例如CyclicBarrier或Phaser,但对于简单的用例,只使用一个没有害处AtomicBoolean.

创建可更新的" ThreadLocal"

所有这些都说,如果你真的想要实现全局更新,ThreadLocal你的实现就会被打破.事实上,你没有遇到问题只是巧合,未来的重构可能会引入难以诊断的错误或崩溃.它" 无故障地工作 "只意味着你的测试不完整.

• 首先,a ArrayList不是线程安全的.当多个线程可能与它交互时,您根本无法使用它(没有外部同步),即使它们将在不同时间这样做.你现在没有看到任何问题只是巧合.
• 将对象存储为a List可防止我们删除过时的值.如果你调用ThreadLocal.set()它将附加到你的列表而不删除以前的值,如果你预计一旦线程终止后这些对象变得无法访问,这会引入内存泄漏和潜在的意外副作用,通常是ThreadLocal实例的情况.您的用例巧合地避免了这个问题,但仍然没有必要使用List.

这是一个IterableThreadLocal安全存储和更新值的所有现有实例的实现ThreadLocal,适用于您选择使用的任何类型:

import java.util.Iterator;
import java.util.concurrent.ConcurrentMap;

import com.google.common.collect.MapMaker;

/**
 * Class extends ThreadLocal to enable user to iterate over all objects
 * held by the ThreadLocal instance.  Note that this is inherently not
 * thread-safe, and violates both the contract of ThreadLocal and much
 * of the benefit of using a ThreadLocal object.  This class incurs all
 * the overhead of a ConcurrentHashMap, perhaps you would prefer to
 * simply use a ConcurrentHashMap directly instead?
 * 
 * If you do really want to use this class, be wary of its iterator.
 * While it is as threadsafe as ConcurrentHashMap's iterator, it cannot
 * guarantee that all existing objects in the ThreadLocal are available
 * to the iterator, and it cannot prevent you from doing dangerous
 * things with the returned values.  If the returned values are not
 * properly thread-safe, you will introduce issues.
 */
public class IterableThreadLocal<T> extends ThreadLocal<T>
                                    implements Iterable<T> {
    private final ConcurrentMap<Thread,T> map;

    public IterableThreadLocal() {
        map = new MapMaker().weakKeys().makeMap();
    }

    @Override
    public T get() {
        T val = super.get();
        map.putIfAbsent(Thread.currentThread(), val);
        return val;
    }

    @Override
    public void set(T value) {
        map.put(Thread.currentThread(), value);
        super.set(value);
    }

    /**
     * Note that this method fundamentally violates the contract of
     * ThreadLocal, and exposes all objects to the calling thread.
     * Use with extreme caution, and preferably only when you know
     * no other threads will be modifying / using their ThreadLocal
     * references anymore.
     */
    @Override
    public Iterator<T> iterator() {
        return map.values().iterator();
    }
}

正如您所希望的那样,这只不过是一个包装器ConcurrentHashMap,并且会产生与直接使用一个相同的开销,但隐藏在a的实现中ThreadLocal,用户通常希望它们快速且线程安全.我为了演示目的实现了它,但我真的不建议在任何设置中使用它.