使用 Map 计算解决 Java 线程可见性和并发错误 [英] Resolving Java thread visibility and concurrency error using Map compute

问题描述

我使用 Java 8.我有一个事件处理程序，它以高速率(每秒 n 个)接受事件，当我收到这么多事件(在这个简化的示例中为 1000)时，我想将它们刷新到存储中/p>

我在第 25 行 myCache.get(event.getKey()).add(event.getBean()); 上有可见性错误吗?我应该在 handleEvent() 方法上同步吗?

public class myClass extends MySimpleEventHanlder {私有映射<字符串，列表>我的缓存；私有的 ScheduledExecutorService 调度器；公共无效我的类(){myCache = new ConcurrentHashMap>();调度程序 = Executors.newSingleThreadScheduledExecutor();scheduler.scheduleAtFixedRate(() -> {for (Iterator>> it = myCache.entrySet().iterator(); it.hasNext();) {Map.Entry>entry = it.next();如果 (entry.getValue().size() >= 1000) {it.remove();//做一些更多的处理，刷新到存储}}}, 0, 60, TimeUnit.SECONDS);}@覆盖公共无效handleEvent(事件事件){如果 (myCachetCache.containsKey(event.getKey())) {myCache.get(event.getKey()).add(event.getBean());}别的{列表beans = new ArrayList();beans.add(event.getBean());myCache.put(event.key, beans);}}}

解决方案

你肯定有可见性问题:你在一个线程中将项添加到 ArrayList 中，并在另一个线程中从该 ArrayList 中读取 size()，没有同步之间.

另一个问题是密钥可能会在调用 myCache.containsKey 和 myCache.get 之间被删除.这将导致 NullPointerException.这可以通过使用计算来解决，它保证是原子的.

 myCache.compute(event.getKey(), (key, value) -> {如果(值 == 空){值 = 新的 ArrayList();}value.add(event.getBean());返回值；});

I use Java 8. I have an event handler that accepts events with a high rate, (n per second) and I want to flush them out to storage when I get so many of them (in this simplified example 1000)

Do I have a visibility error on line 25 myCache.get(event.getKey()).add(event.getBean()); ? Should I synchronize on handleEvent() method?

public class myClass extends MySimpleEventHanlder {
    private Map<String, List<MyBean>> myCache;
    private ScheduledExecutorService scheduler;

    public void MyClass() {
        myCache = new ConcurrentHashMap<String, List<MyBean>>();
        scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(() -> {

            for (Iterator<Map.Entry<String, List<MyBean>>> it = myCache.entrySet().iterator(); it.hasNext();) {
                Map.Entry<String, List<MyBean>> entry = it.next();
                if (entry.getValue().size() >= 1000) {
                    it.remove();
                    //do some more processing , flush to storage
                }
            }
        }, 0, 60, TimeUnit.SECONDS);
    }

    @Override
    public void handleEvent(Event event) {

        if (myCachetCache.containsKey(event.getKey())) {
            myCache.get(event.getKey()).add(event.getBean());
        }
        else{
            List<MyBean> beans = new ArrayList<MyBeans>();
            beans.add(event.getBean());
            myCache.put(event.key, beans);
        }
    }
}

解决方案

You definitely have visibility problems: you add items into an ArrayList in one thread, and read the size() from that ArrayList in another thread, with no synchronization in between.

Another problem is that the key may get removed between the calls to myCache.containsKey and myCache.get. This would cause a NullPointerException. That could be solved by using compute, which is guaranteed to be atomic.

    myCache.compute(event.getKey(), (key, value) -> {
        if (value == null) {
            value = new ArrayList<>();
        }
        value.add(event.getBean());
        return value;
    });

这篇关于使用 Map 计算解决 Java 线程可见性和并发错误的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！