Java - HashMap 值 - 易失性

Question

我的情况是，我有 N 个线程做相同的工作，而一个线程 X 做不同的事情。

每个线程 N 读取/写入一个静态对象，该对象属于一个类（称为 MMLCounter），它是 HashMap 的包装器，并且每个线程使用该 HashMap 的不同键/值对，因此所有线程都读取/写入值到HashMap 同时进行。 线程 X 需要定期访问所有值，并且在访问它们的同时（从访问第一个值到访问最后一个值，其他 N 个线程都不能更改 HashMap 中的值）。

HashMap在程序执行开始时在线程创建期间被初始化并添加键/值，之后不再添加新的键/值，只有HashMap中的值发生变化。

正因为如此，我没有使用 ConcurrentHashMap 或同步函数，而是创建了一个围绕 HashMap 的包装类，它还有一个标志，指示 N 个线程是否允许它们更改值，并且此标志是专门更改的通过线程 X。

这样所有 N 个线程都可以并行使用 HashMap，但是当线程 X 开始工作时，它只能使用 HashMap 直到它完成。

我的问题是我是否需要将任何内容声明为 volatile（例如 HashMap 中的值），如果需要，是什么以及如何声明？

我想避免的事情（不知道它是否可能发生）是 N 个线程中的一个更改了 HashMap 中的一个值，但该值的更改仅反映在该本地缓存中线程，当线程 X 从 HashMap 读取该值时，它将从其本地缓存内存中读取它，该内存与另一个 N 线程的本地缓存内存不同步，这意味着它将具有旧值。

代码如下：

public static void main(String[] args) throws ProtocolException { 

    int NUMBER_OF_THREADS = 400; 

    List<Future<?>> futureList = new ArrayList<Future<?>>(); 
    ExecutorService executor = Executors.newFixedThreadPool(NUMBER_OF_THREADS+1);
    futureList.add(executor.submit(new Runnable() {
        @Override
        public void run() { 
                int measureInterval = 10000;
                try {
                    Thread.sleep(measureInterval);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                } 
                System.out.println("--> MML rate is : " + MMLGenerator.MML_COUNTER.getMMLRate(measureInterval/1000) + " MML per second.");
        }
    }));

    //create and start sending threads. 
    for (int threadNmbr = 0; threadNmbr < NUMBER_OF_THREADS; threadNmbr++) {
        futureList.add(executor.submit(new Thread(new MMLGenerator(threadNmbr))));
        try {
            Thread.sleep(50);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    } 

    //wait for threads to finish. 
    for (Future<?> future : futureList) { 
        try { 
            future.get(); 
        } catch (InterruptedException e) {
        } 
        catch (ExecutionException e) { 
            throw (RuntimeException) e.getCause(); 
        } 
    } 
    executor.shutdown();
} 

class MMLGenerator implements Runnable {

    public static volatile MMLCounter MML_COUNTER = new MMLCounter();
    private int threadNmbr = 0;

    public MMLGenerator(int threadNmbr) {
        this.threadNmbr = threadNmbr;
        MMLGenerator.MML_COUNTER.put(this.threadNmbr, 0);

    }

    @Override
    public void run() {
        while(RUN_ACTIVE) {
            MML_COUNTER.increaseCounter(this.threadNmbr);
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
}

public class MMLCounter {

  private Map<Integer,Integer> MMLCounter = new HashMap<Integer, Integer>();
  private boolean MMLCounterLocked = false;

  public Integer get(Integer key) {
      return this.MMLCounter.get(key);
  }

  public Integer put(Integer key, Integer value) {
      while (this.MMLCounterLocked) {
          try {
              Thread.sleep(100);
          } catch (InterruptedException e) {
              // TODO Auto-generated catch block
              e.printStackTrace();
          }
      }
      return this.MMLCounter.put(key, value);
  }

  public void increaseCounter(Integer key) {
      while (this.MMLCounterLocked) {
          try {
              Thread.sleep(100);
          } catch (InterruptedException e) {
              // TODO Auto-generated catch block
              e.printStackTrace();
          }
      }
      this.MMLCounter.put(key,this.MMLCounter.get(key).intValue() + 1);
  }

  public int getMMLRate(int measurementTime) {
      this.MMLCounterLocked = true;
      int MMLCounterSum = 0;
      for (Integer counterID : this.MMLCounter.keySet()) {
          int counter = this.MMLCounter.get(counterID);
          MMLCounterSum += counter;
          this.MMLCounter.put(counterID, 0);
      }
      this.MMLCounterLocked = false;
      return MMLCounterSum/measurementTime;
  }

}

---

修改后

感谢大家的帮助。 我刚刚阅读了ReentrantReaderWriterLock 的描述，这确实是我需要的。下面是修改后的代码。

但是，我还有两个问题：

1) 如果我使用ReentrantReaderWriterLock 保护代码的关键部分，为什么还需要使用ConcurrentHashMap 而不是HashMap？

2) ReentrantReaderWriterLock 的这种用法只会从我之前的实现中替换我现在看到的标志的用法没有正确完成。但是，我仍然有HashMap 中的值对象不是易失性的问题，因此不同的线程将各自拥有自己的本地缓存值副本，该副本与来自其他线程的值的本地缓存副本不同步？

public static void main(String[] args) throws ProtocolException { 

    int NUMBER_OF_THREADS = 400; 

    List<Future<?>> futureList = new ArrayList<Future<?>>(); 
    ExecutorService executor = Executors.newFixedThreadPool(NUMBER_OF_THREADS+1);
    futureList.add(executor.submit(new Runnable() {
        @Override
        public void run() { 
                int measureInterval = 10000;
                try {
                    Thread.sleep(measureInterval);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                } 
                System.out.println("--> MML rate is : " + MMLGenerator.counter.getMMLRate(measureInterval/1000) + " MML per second.");
        }
    }));

    //create and start sending threads. 
    for (int threadNmbr = 0; threadNmbr < NUMBER_OF_THREADS; threadNmbr++) {
        futureList.add(executor.submit(new Thread(new MMLGenerator(threadNmbr))));
        try {
            Thread.sleep(50);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    } 

    //wait for threads to finish. 
    for (Future<?> future : futureList) { 
        try { 
            future.get(); 
        } catch (InterruptedException e) {
        } 
        catch (ExecutionException e) { 
            throw (RuntimeException) e.getCause(); 
        } 
    } 
    executor.shutdown();
} 

class MMLGenerator implements Runnable {

    public static MMLCounter counter = new MMLCounter();
    private int threadNmbr = 0;

    public MMLGenerator(int threadNmbr) {
        this.threadNmbr = threadNmbr;
        MMLCounter.counter.put(this.threadNmbr, 0);

    }

    @Override
    public void run() {
        while(RUN_ACTIVE) {
            MMLCounter.counter.increaseCounter(this.threadNmbr);
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
}

public class MMLCounter {

  private Map<Integer,Integer> counter = new HashMap<Integer, Integer>();
  public static final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);

  public Integer put(Integer key, Integer value) {
      lock.readLock().lock();
      Integer oldValue = this.counter.put(key, value);
      lock.readLock().unlock();
      return oldValue;
  }

  public void increaseCounter(Integer key) {
      lock.readLock().lock();
      this.counter.put(key,this.counter.get(key).intValue() + 1);
      lock.readLock().unlock();
  }

  public int getMMLRate(int measurementTime) {
      lock.writeLock().lock();
      int counterSum = 0;
      for (Integer counterID : this.counter.keySet()) {
          counterSum += this.counter.get(counterID);;
          this.counter.put(counterID, 0);
      }
      lock.writeLock().unlock();
      return counterSum/measurementTime;
  }
}

---

第二次修改后

我现在发现我需要实现的逻辑需要我从多个线程操作多个计数器，而不仅仅是一个，并且每个线程在任何时候都可以更改任何计数器。 下面是我的实现，但我不确定我在性能和数据一致性方面是否做得很好。

我除了有随机数量的计数器（计数器的数量将在执行开始时知道）将由字符串值标识，并且每个计数器必须计算两个值（第一个值总是会增加，第二个值有时只会增加，但如果它们增加，它们需要同时增加）。当我需要每个计数器的总和时，我需要在原子操作中获取两个计数器值，并且从我获取第一个计数器到我获取最后一个计数器的时间，其他线程都不能更改任何计数器。

出于演示目的，作为 Counter 标识（HashMap 中的键），我取了 Counter 的序号的 String 值，并确定在每个线程的每次迭代中需要增加哪个 Counter，以及确定是否只有一个或两个值计数器需要增加，我用的是随机生成器。

public static void main(String[] args) { 

    int NUMBER_OF_THREADS = 400;


    MMLGenerator.counterNmbr(2);
    List<Future<?>> futureList = new ArrayList<Future<?>>(); 
    ExecutorService executor = Executors.newFixedThreadPool(NUMBER_OF_THREADS+1); 

    futureList.add(executor.submit(new Runnable() {
        @Override
        public void run() { 
            while(true)
            {
                int measureInterval = 10;
                try {
                    TimeUnit.SECONDS.sleep(measureInterval);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
                MMLGenerator.lock.writeLock().lock();
                for (String counterId : MMLGenerator.counter.keySet()) {
                    MMLCounterSimple counter = MMLGenerator.counter.get(counterId).getCountAndReset();
                    System.out.println("--> For counter " + counterId + " total is : " + counter.getTotal() + ", and failed is : " + counter.getFailed());
                }
                MMLGenerator.lock.writeLock().unlock();
            }
        }
    }));

    //create and start sending threads. 
    for (int threadNmbr = 0; threadNmbr < NUMBER_OF_THREADS; threadNmbr++) {
        futureList.add(executor.submit(new Thread(new MMLGenerator())));
        try {
            TimeUnit.MILLISECONDS.sleep(50);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    } 

    //wait for threads to finish. 
    for (Future<?> future : futureList) { 
        try { 
            future.get(); 
        } catch (InterruptedException e) {
        } 
        catch (ExecutionException e) { 
            throw (RuntimeException) e.getCause(); 
        } 
    } 
    executor.shutdown();
} 



class MMLGenerator implements Runnable {

    public static volatile HashMap<String, MMLCounter> counter = new HashMap<String, MMLCounter>();
    public static final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);


    public static void counterNmbr(int counterNmbr) {
        lock.writeLock().lock();
        for(int i = 0; i < counterNmbr; i++) {
            counter.put(new Integer(i).toString(), new MMLCounter());
        }
        lock.writeLock().unlock();
    }

    @Override
    public void run() {
        while(RUN_PROVISIONING) {
            lock.readLock().lock();
            String counterID = new Integer(new Random().nextInt(counter.size())).toString();
            long failedInc = 0;
            if (new Random().nextInt(2) == 0) {
                failedInc = 1;
            }
            counter.get(counterID).increaseCounter(failedInc);
            lock.readLock().unlock();
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }

}


public class MMLCounter {

    private volatile long total = 0;
    private volatile long failed = 0;
    public static final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);


    public synchronized void increaseCounter(long failedInc) {
        lock.writeLock().lock();
        total++;;
        failed = failed + failedInc;
        lock.writeLock().unlock();
    }

    public synchronized MMLCounterSimple getCountAndReset() {
        lock.writeLock().lock();
        MMLCounterSimple simpleCounter = new MMLCounterSimple(total, failed);
        total = 0;
        failed = 0;
        lock.writeLock().unlock();
        return simpleCounter;
    }

}


public class MMLCounterSimple {

    private long total = 0;
    private long failed = 0;

    public MMLCounterSimple(long total, long failed) {
        this.total = total;
        this.failed = failed;
    }

    public long getTotal() {
        return this.total;
    }

    public long getFailed() {
        return this.failed;
    }
}

Answer 1

正如所写，不能保证按预期工作。 N 个线程执行的写入和 X 线程执行的读取之间没有同步点。甚至MMLCounterLocked 标志也可能被作者忽略。

除了使您的代码正常工作之外，使用像 ConcurrentMap 这样的更高级别的并发工具将大大简化您的代码。

---

由于您只需要总和，LongAccumulator 就足够了，它使代码非常简单和安全。

  public static void main(String[] args)
    throws Exception
  {
    int NUMBER_OF_THREADS = 400;
    ExecutorService executor = Executors.newFixedThreadPool(NUMBER_OF_THREADS);
    LongAccumulator sum = new LongAccumulator(Long::sum, 0);
    for (int i = 0; i < NUMBER_OF_THREADS; ++i) {
      executor.submit(new MMLGenerator(sum));
      TimeUnit.MILLISECONDS.sleep(50); /* Why??? */
    }
    int interval = 10;
    TimeUnit.SECONDS.sleep(interval);
    long rate = sum.getThenReset() / interval;
    System.out.println("--> MML rate is : " + rate + " MML per second.");
    executor.shutdownNow();
  }

  private static final class MMLGenerator
    implements Runnable
  {

    private final LongAccumulator counter;

    MMLGenerator(LongAccumulator counter)
    {
      this.counter = counter;
    }

    @Override
    public void run()
    {
      while (true) {
        counter.accumulate(1);
        try {
          TimeUnit.SECONDS.sleep(1);
        }
        catch (InterruptedException shutdown) {
          break;
        }
      }
    }

  }

---

关于你的两个新问题：

1. “如果我使用ReentrantReaderWriterLock 保护代码的关键部分，为什么还需要使用ConcurrentHashMap 而不是HashMap？”

您没有使用读写锁保护关键部分。当您 write 表时，您正在获取 read 锁，当您 read 时，您正在获取 write 锁。 HashMap 的行为未在并发修改下定义。您可能会导致正在使用该表的线程挂起。

此外，您应该使用try-finally 构造来确保无论发生任何错误都可以解锁。

如果您使用ConcurrentMap，线程可以在不获取整个表的锁的情况下进行更新，这就是您正确应用读写锁时要做的事情。

1. “ReentrantReaderWriterLock 的这种用法只会从我之前的实现中替代我现在看到的标志的用法没有正确完成。但是，我仍然有HashMap 中的值对象不是volatile 的问题，所以[将]不同的线程......每个都有自己的本地缓存值副本，该副本与其他线程的值的本地缓存副本不同步？”

不，Lock 获取确实与其他线程同步，并且在获取锁之前发生的其他线程的更改将是可见的。如果您修复了锁定，HashMap 将正常工作，但它会通过在更新期间锁定整个表来实现。

Answer 2

是的，你需要volatile，没有它就不能保证其他线程会看到变化； java 可能（并且通常会）在另一个线程中本地缓存该值，因此在本地更改它不会对等待它的其他线程产生任何影响。

例如，这一行：

this.MMLCounterLocked = true;

如果字段不为volatile，则无用。

---

您为什么要尝试实现自己的锁？只需通过在静态字段或静态方法上同步来使用 java 的锁定。

Answer 3

您可能希望使用允许多个读取器和一个写入器独占访问的锁定方案，例如java.util.concurrent.locks.ReentrantReadWriteLock，而不是尝试烘焙自己的同步机制。

The docs say

ReadWriteLock 维护一对关联的锁，一个用于 只读操作和一个用于写入的操作。读锁可能被持有 同时由多个阅读器线程，只要没有 作家。写锁是独占的。

Answer 4

结合罗布和埃里克森所说的话：

使用ConcurrentHashMap 这是线程安全的。

使用ReentrantReaderWriterLock，让所有正在编辑地图的线程获取ReadLock，这样他们就可以同时进行编辑。锁的读取端是共享锁。那么需要整个地图的一致视图的线程应该使用WriteLock来获得对地图的独占访问权。

您需要ConcurrentHashMap 的原因是所有线程都可以看到对地图的所有编辑，包括新条目。在这里使用ReentrantReaderWriterLock 的读取端是不够的。

您应该使用 ReentrantReaderWriterLock 的原因是，当线程 X 需要对映射的独占访问时，它可以通过使用锁的写入方来获得它。

这里最大的问题是在increaseCounter 中存在一个不能保证是原子的依赖项。你得到这个值然后你增加它。当两个线程调用此方法并且都获得相同的值然后增加它时会发生什么。如果这是一个银行账户，并且有人因为比赛条件损失了一分钱，那么......你明白了。这里有一个不变量没有得到支持，需要一些更细粒度的原子性。考虑使用AtomicInteger 作为值方，或者查看Atomic 包中的所有优秀类，这些类提供具有原子操作的单个变量。