java 8 Map.merge是线程安全的吗？

Question

import java.util.HashMap;
import java.util.Map;

public class MainTest
{
    static Map<String, Integer> map = new HashMap();

    public static void incr()
    {
        map.merge("counter", 1, Integer::sum);
    }

    public static void decr()
    {
        map.merge("counter", -1, Integer::sum);
    }

    public static void main(String[] args) throws Exception
    {
        map.put("counter", 0);
        for (int i = 0; i < 10000; i++)
        {
            Thread t1 = new Thread(new Runnable()
            {
                @Override
                public void run()
                {
                    incr();
                }
            });
            t1.join();
            t1.start();

            Thread t2 = new Thread(new Runnable()
            {
                @Override
                public void run()
                {
                    decr();
                }
            });
            t2.join();
            t2.start();
        }
      System.out.println(map);
    }

}

结果{counter=-2}是运行main方法时.为什么不是0？

Answer 1

接口merge上的Javadoc Map说:

默认实现不保证此方法的同步或原子性属性.提供原子性保证的任何实现都必须覆盖此方法并记录其并发属性.

虽然HashMap覆盖了默认实现,但它没有关于该实现的并发属性的文档,但它确实有这个一般声明:

请注意,此实现不同步.如果多个线程同时访问哈希映射,并且至少有一个线程在结构上修改了映射,则必须在外部进行同步

因此它不是线程安全的.

PS,你不清楚为什么你打电话t1.join(),t2.join()在你开始相应的线程之前.

如果您拨打电话

    t1.join();
    t1.start();

至

    t1.start();
    t1.join();

和

    t2.join();
    t2.start();

至

    t2.start();
    t2.join();

你会得到0的输出.当然,如果你这样做,根本就没有并发修改,因为每个线程将在前一个线程死后启动.

另一种方法是从map.merge外部同步调用:

public static void incr()
{
    synchronized(map) {map.merge("counter", 1, Integer::sum);}
}

public static void decr()
{
    synchronized(map) {map.merge("counter", -1, Integer::sum);}
}

Answer 2

要求修改数据结构的特定单一方法的线程安全性有点奇怪,因为一般来说HashMap,这些数据结构被记录为不是线程安全的.

如果要同时修改映射,则必须查找支持并发更新的实现,这通常通过实现来显示ConcurrentMap.对于这些映射,即使default实现也足够了,因为它是在其他接口方法上实现的,这些方法保证是线程安全的.但是,如果您希望保证原子性,则需要使用适当的实现覆盖该方法的实现,例如ConcurrentHashMap:

merge

...整个方法调用以原子方式执行.其他线程在此映射上的某些尝试更新操作可能在计算进行时被阻止,因此计算应该简短,并且不得尝试更新此Map的任何其他映射.

要修复和简化您的示例:

Map<String, Integer> map = new ConcurrentHashMap<>();
for (int i = 0; i < 10000; i++) {
    Thread t1 = new Thread(() -> map.merge("counter", -1, Integer::sum));
    Thread t2 = new Thread(() -> map.merge("counter",  1, Integer::sum));
    t1.start();
    t2.start();
    t1.join();
    t2.join();
}
System.out.println(map);

请注意,在原始代码中,您join() 在开始之前调用,这没有任何效果.由于您没有进行join()后续处理start(),因此您的代码可以map在所有线程完成之前打印,因此即使HashMap线程安全,它也可以打印非零值.

在上面的代码中执行join()after之后start()将正确等待完成,但最多允许两个并发更新操作.

要提高并发性,您应该放弃手动创建线程:

ExecutorService threadPool
    = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
Map<String, Integer> map = new ConcurrentHashMap<>();
for (int i = 0; i < 10000; i++) {
    threadPool.execute(() -> map.merge("counter", -1, Integer::sum));
    threadPool.execute(() -> map.merge("counter",  1, Integer::sum));
}
threadPool.shutdown();
threadPool.awaitTermination(1, TimeUnit.DAYS);
System.out.println(map);

这允许并发高于2,但由于工作线程可能能够以您的循环可以调度新作业的速度执行此简单任务,因此生成的并发可能仍然接近于两个.