使用HashMap和ScheduledExecutorService进行线程安全

Question

我构建了这个“节流”任务运行器，它在HashMap中收集一些数据，同时（每分钟）将数据“带走”并清除HashMap。在我的测试中，我注意到executor部分可以停止scheduleAtFixedRate，并且永远不会再次清除HashMap。我假设这是因为我所做的HashMap修改不是线程安全的，它在run（）内部崩溃，没有恢复。我正在两个线程中修改HashMap。有人能告诉我如何优化HashMap修改的正确方向吗。

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class StatisticEventsDispatcher {
  public HashMap<String, AbstractStatisticsEvent> mappedCachedEvents = new HashMap<>();

  final Duration timeout = Duration.ofMinutes(1);

  final ScheduledExecutorService executor = Executors.newScheduledThreadPool(10);

  public StatisticEventsDispatcher(EventBus eventBus) {
    executor.scheduleAtFixedRate(
        new Runnable() {
          @Override
          public void run() {
            mappedCachedEvents.values().forEach(eventBus::post);
            mappedCachedEvents.clear();
            // i think this is not thread safe
          }
        },
        timeout.toMillis(),
        timeout.toMillis(),
        TimeUnit.MILLISECONDS);
  }

  public void applyChanges(String type, Map<String, Long> changes) {
    AbstractStatisticsEvent event;
    if (mappedCachedEvents.containsKey(type)) {
      event = mappedCachedEvents.get(type);
    } else {
      event = new AbstractStatisticsEvent(type);
      mappedCachedEvents.put(type, event);
      // i think this not thread safe
    }
    event.apply(changes);
  }
}

Answer 1

是的，如果两个线程在没有任何并发控制的情况下修改一个 HashMap，则它不会按预期运行。

虽然有< code>ConcurrentMap实现，但您似乎希望自动“发布”所收集事件的快照。也就是说，似乎您不希望在迭代集合时添加新事件或更新事件，即使这可以通过线程安全的方式来完成。

在迭代期间同步对共享映射的访问相对简单，但如果迭代需要一段时间，则在此期间将阻止来自主线程的更新，这似乎是不可取的。

或者，计划任务可以短暂锁定以与新地图交换：

private final Object lock = new Object();
private HashMap<String, AbstractStatisticsEvent> mappedCachedEvents = new HashMap<>();

...

    executor.scheduleAtFixedRate(() -> 
        {
            Map<String, AbstractStatisticsEvent> toPost;
            synchronized(lock) {
              toPost = mappedCachedEvents;
              mappedCachedEvents = new HashMap<>();
            }
            toPost.values().forEach(eventBus::post);
        },
        timeout.toMillis(), timeout.toMillis(), TimeUnit.MILLISECONDS);

...

public void applyChanges(String type, Map<String, Long> changes) {
    synchronized(lock) {
        AbstractStatisticsEvent event = 
          mappedCachedEvents.computeIfAbsent(type, AbstractStatisticsEvent::new);
        event.apply(changes);        
    } 
  }
}

Answer 2

我建议实现如下：

public class StatisticEventsDispatcher {
    private ConcurrentMap<String, AbstractStatisticsEvent> mappedCachedEvents = new ConcurrentHashMap<>();
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    final Duration timeout = Duration.ofMinutes(1);

    final ScheduledExecutorService executor = Executors.newScheduledThreadPool(10);

    public StatisticEventsDispatcher(EventBus eventBus) {
        executor.scheduleAtFixedRate(
                new Runnable() {
                    @Override
                    public void run() {
                        Map<String, AbstractStatisticsEvent> tmp;
                        lock.writeLock().lock();
                        try {
                            tmp = mappedCachedEvents;
                            mappedCachedEvents = new ConcurrentHashMap<>();
                        } finally {
                            lock.writeLock().unlock();
                        }
                        tmp.values().forEach(eventBus::post);
                    }
                },
                timeout.toMillis(),
                timeout.toMillis(),
                TimeUnit.MILLISECONDS);
    }

    public void applyChanges(String type, Map<String, Long> changes) {
        ConcurrentMap<String, AbstractStatisticsEvent> tmp;
        lock.readLock().lock();
        try {
            tmp = mappedCachedEvents;
        } finally {
            lock.readLock().unlock();
        }
        AbstractStatisticsEvent event = tmp.computeIfAbsent(type, AbstractStatisticsEvent::new);
        event.apply(changes);
    }
}

请注意，我将mappedCachedEvents设置为private，因为如果没有锁，就无法访问它。

Answer 3

您的代码绝对不是线程安全的。但从你的描述来看，没有理由这样。您真的需要此调度程序中的多个线程吗？

只需更新和复制地图就非常快。一个线程总是比多个线程争用访问更快，即使你有成千上万的消费者/生产者。

所以我的建议是保留一个“拥有”Map的单线程执行器(即从调度程序的线程内部进行所有的更新/读取)。

如果“发送”地图的内容需要输入输出，请确保使用Map.copyOf（mutableMap）将地图复制到不可变的地图中，然后将该副本提供给另一个线程进行处理。同时，“活动”地图可以继续更新/读取…你甚至可以编写一个“排水”操作，有效地将所有地图的内容放入另一个地图中，以避免同时存在重复的地图。