ConcurrentMap VS SynchronizedMap

ConcurrentMap VS SynchronizedMap

2010-05-06 星期四 晴

最近在做BOPS搜索排序项目二期，主要是性能优化。根据分析，一期的主要性能消耗在异步计算第P页的"前面有N条"的N值，因为需要遍历前P-1页，并且计算某个公司的产品数。通过测试，这是一个NIO密集型，而不是CPU密集型。对于NIO的优化，主要有本地缓存，但是由于key的变化性（客户随意输入），缓存的hit概率太低，不适合。因此考虑通过并发访问来缩短时间，前提是任务可以进行相对独立的划分。设计如下：
在第P页计算"前面有N条"，如果P<60,那么采用单线程足以，因为前60页每一页的加载时间大概为150ms, 150ms * 60 = 9s,这个是完全可以接受的。
如果P>60，那么根据越往后面越慢的规律（前60页每页大概150ms；60-100页，每页大概180ms；100-150每页大概330ms，150-200每页大概450ms,200-260每页大概600ms）。
采用如下线程分配机制：（保证所有的线程大概同时完成（控制在10s内），避免木桶原理）。
线程1：负责前60页
线程2：负责前61-XX（具体数字需要进一步的测试）
线程3：负责前XX-XX
...
线程N：负责前XX-XX (N<=12)
每个线程查询之后，各自进行计算，并将计算结果存放在一个公用的Map中。以company_id为key，以count为value。由于并非问题，将使用ConcurrentMap保证共享数据的一致性。
备注：考虑到由于并发加锁和解锁可能导致性能变差，可以采用每个线程有自己的Map，当所有子线程都计算完成之后，主线程进行合并。

补记：其实没有必要做这样的划分，将每一页的请求作为一个job放入任务池，让所有子线程去执行就可以了。

根据这个设计，笔者对ConcurrentMap和SynchronizedMap分别进行了测试。结果发现了一些问题。
首先测试ConcurrentMap:

package alibaba.b2b.forrest;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ConcurrentMapTest {

    public static void main(String args[]) throws InterruptedException {
        long begin = System.currentTimeMillis();
        ConcurrentMap<String, Integer> map = new ConcurrentHashMap<String, Integer>();

        // 在线程池中创建 10 个线程
        ExecutorService exec = Executors.newFixedThreadPool(10);
        // 创建 1000 个线程目标对象
        for (int index = 0; index < 1000; index++) {

            Runnable run = new Runner(index, map);
            // 执行线程目标对象
            exec.execute(run);
        }
        // shutdown
        exec.shutdown();

        while (true) {
            try {
                boolean finish = exec.awaitTermination(600, TimeUnit.SECONDS);
                if (finish) {
                    long end = System.currentTimeMillis();
                    System.err.println("execute time: " + (end - begin) + " milliseconds");
                    System.out.println(map);
                    break;
                }
            } catch (InterruptedException e) {
                // nothing
            }
        }

    }
}

// 线程目标对象
class Runner implements Runnable {

    ConcurrentMap<String, Integer> map;

    int                            index = 0;

    public Runner(int index, ConcurrentMap<String, Integer> map) {
        this.map = map;
        this.index = index;
    }

    @Override
    public void run() {
        // 输出线程的名字和使用目标对象
        System.out.println("线程名:" + Thread.currentThread().getName() + "(目标对象" + index + ")");

        Integer count = map.putIfAbsent("key", 1);
        if (count != null && !map.replace("key", count, count + 1)) {
            System.err.println("replace failed!");
        }
    }
}

如果正常的话，map中的最终结果应该是key=1000，但是运行多次却发现有时候不是：

线程名:pool-1-thread-2(目标对象1)
线程名:pool-1-thread-3(目标对象2)
线程名:pool-1-thread-1(目标对象0)
// ... 省略N条 ...
线程名:pool-1-thread-3(目标对象11)
线程名:pool-1-thread-10(目标对象9)
replace failed!
线程名:pool-1-thread-2(目标对象20)
线程名:pool-1-thread-2(目标对象21)
// ... 省略N条 ...
线程名:pool-1-thread-2(目标对象602)
线程名:pool-1-thread-5(目标对象592)
线程名:pool-1-thread-5(目标对象604)
replace failed!
线程名:pool-1-thread-5(目标对象605)
线程名:pool-1-thread-5(目标对象606)
线程名:pool-1-thread-5(目标对象607)
// ... 省略N条 ...
线程名:pool-1-thread-5(目标对象642)
线程名:pool-1-thread-5(目标对象643)
线程名:pool-1-thread-5(目标对象644)
线程名:pool-1-thread-5(目标对象645)replace failed!

线程名:pool-1-thread-5(目标对象647)
线程名:pool-1-thread-5(目标对象648)
// ... 省略N条 ...
线程名:pool-1-thread-8(目标对象620)
线程名:pool-1-thread-3(目标对象617)
线程名:pool-1-thread-10(目标对象611)
线程名:pool-1-thread-2(目标对象603)
线程名:pool-1-thread-5(目标对象654)
{key=997}
execute time: 94 milliseconds

看到有三条是插入失败的。所以最终结果是997条，而不是1000条。
原因在于ConcurrentMap采用的是类似于CSMA的冲突检测方式（确切地说，是ConcurrentMap提供了冲突检查接口），而不是锁机制（锁是一种类似于令牌网的方式）。这可以提高并发性，但是同时也不能保证一定会成功。我们这里使用
boolean java.util.concurrent.ConcurrentMap.replace(String key, Integer oldValue, Integer newValue);接口就是这种防御式更新方式。这个在人工翻译项目中对数据库的更性也采用了类似的处理方式。但是由于oldValue由于其他线程更新导致已经不一致时，这个更新（插入）操作就会失败。解决方式是像CSMA一样，随机等待一段时间再重试。
或者简单的一直重试：

// 线程目标对象
class Runner implements Runnable {

    ConcurrentMap<String, Integer> map;

    int                            index = 0;

    public Runner(int index, ConcurrentMap<String, Integer> map) {
        this.map = map;
        this.index = index;
    }

    @Override
    public void run() {
        // 输出线程的名字和使用目标对象
        System.out.println("线程名:" + Thread.currentThread().getName() + "(目标对象" + index + ")");

        Integer count = map.putIfAbsent("key", 1);
        while (count != null && !map.replace("key", count, count + 1)) {
            System.err.println(">>>>>>>>>>>>>Conflict!<<<<<<<<<<<<<<<<");
            count = map.get("key");
        }
    }
}

线程名:pool-1-thread-4(目标对象3)
线程名:pool-1-thread-7(目标对象6)
线程名:pool-1-thread-6(目标对象5)
线程名:pool-1-thread-3(目标对象2)
线程名:pool-1-thread-5(目标对象4)
线程名:pool-1-thread-2(目标对象1)
线程名:pool-1-thread-1(目标对象0)
线程名:pool-1-thread-8(目标对象7)
// ... 省略N条 ...
线程名:pool-1-thread-1(目标对象794)
线程名:pool-1-thread-1(目标对象795)
>>>>>>>>>>>>>Conflict!<<<<<<<<<<<<<<<<
>>>>>>>>>>>>>Conflict!<<<<<<<<<<<<<<<<
线程名:pool-1-thread-1(目标对象796)
线程名:pool-1-thread-1(目标对象797)
线程名:pool-1-thread-1(目标对象798)
// ... 省略N条 ...
线程名:pool-1-thread-1(目标对象998)
线程名:pool-1-thread-1(目标对象999)
execute time: 128 milliseconds
线程名:pool-1-thread-10(目标对象408)
线程名:pool-1-thread-5(目标对象551)
线程名:pool-1-thread-3(目标对象534)
线程名:pool-1-thread-7(目标对象526)
线程名:pool-1-thread-6(目标对象519)
线程名:pool-1-thread-9(目标对象550)
线程名:pool-1-thread-8(目标对象549)
线程名:pool-1-thread-4(目标对象456)
线程名:pool-1-thread-2(目标对象470)
{key=1000}

再来看看SynchronizedMap：

package alibaba.b2b.forrest;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class SynchronizedMapTest {

    public static void main(String args[]) throws InterruptedException {
        long begin = System.currentTimeMillis();

        Map<String, Integer> m = new HashMap<String, Integer>();
        Map<String, Integer> map = Collections.synchronizedMap(m);

        // 在线程池中创建 10 个线程
        ExecutorService exec = Executors.newFixedThreadPool(10);
        // 创建 1000 个线程目标对象
        for (int index = 0; index < 1000; index++) {

            Runnable run = new Runner(index, map);
            // 执行线程目标对象
            exec.execute(run);
        }
        // shutdown
        exec.shutdown();

        while (true) {
            try {
                boolean finish = exec.awaitTermination(600, TimeUnit.SECONDS);
                if (finish) {
                    long end = System.currentTimeMillis();
                    System.err.println("execute time: " + (end - begin) + " milliseconds");
                    System.out.println(map);
                    break;
                }
            } catch (InterruptedException e) {
                // nothing
            }
        }

    }
}

// 线程目标对象
class Runner implements Runnable {

    Map<String, Integer> map;

    int                  index = 0;

    public Runner(int index, Map<String, Integer> map) {
        this.map = map;
        this.index = index;
    }

    @Override
    public void run() {
        // 输出线程的名字和使用目标对象
        System.out.println("线程名:" + Thread.currentThread().getName() + "(目标对象" + index + ")");

        Integer count = map.get("key");
        if (count != null) {
            Integer oldCount = map.put("key", count + 1);
            if (!oldCount.equals(count)) {
                System.err.println(">>>>>>>>>>not Consistent!<<<<<<<<<<<");
            }
        } else {
            Integer oldCount = map.put("key", 1);
            if (oldCount != null) {
                System.err.println(">>>>>>>>>>Oops!<<<<<<<<<<<");
            }
        }
    }
}

运行结果更糟糕：

线程名:pool-1-thread-2(目标对象1)
线程名:pool-1-thread-3(目标对象2)
线程名:pool-1-thread-4(目标对象3)
线程名:pool-1-thread-5(目标对象4)
线程名:pool-1-thread-6(目标对象5)
// ... 省略N条 ...
线程名:pool-1-thread-10(目标对象938)
线程名:pool-1-thread-9(目标对象937)
线程名:pool-1-thread-3(目标对象936)
线程名:pool-1-thread-4(目标对象935)
线程名:pool-1-thread-5(目标对象934)
线程名:pool-1-thread-1(目标对象673)
线程名:pool-1-thread-6(目标对象671)
{key=999}
>>>>>>>>>>not Consistent!<<<<<<<<<<<
execute time: 94 milliseconds

线程名:pool-1-thread-1(目标对象0)
线程名:pool-1-thread-3(目标对象2)
线程名:pool-1-thread-2(目标对象1)
>>>>>>>>>>Oops!<<<<<<<<<<<
>>>>>>>>>>Oops!<<<<<<<<<<<
线程名:pool-1-thread-4(目标对象3)
线程名:pool-1-thread-5(目标对象4)
线程名:pool-1-thread-6(目标对象5)
线程名:pool-1-thread-7(目标对象6)
// ... 省略N条 ...
线程名:pool-1-thread-7(目标对象378)
线程名:pool-1-thread-6(目标对象377)
线程名:pool-1-thread-5(目标对象376)
线程名:pool-1-thread-4(目标对象375)
线程名:pool-1-thread-2(目标对象374)
线程名:pool-1-thread-10(目标对象79)
execute time: 83 milliseconds
{key=998}

两种情况都可能发生，说明Integer count = map.get("key");取得key之后并没有加锁，看一下JDK的实现就知道了：

public V get(Object key) {
	    synchronized(mutex) {return m.get(key);}
        }

	public V put(K key, V value) {
	    synchronized(mutex) {return m.put(key, value);}
        }

所以千万不要以为SyncronizedMap带着个Syncronized定语就真的是同步的。

解决方法：
1. 将对公共Map的代码块放在一个Syncronized方法块中，保证同步。
2. 使用页码作为key，Map<String, Integer>作为value，这样不用的线程根据各自的key存取公共map的东西，就不会有冲突的问题了。
3. 主线程为每个子线程创建一个新的map传递给子线程。