我的代码线程不安全吗？答案

【问题标题】：Is my code thread-unsafe?我的代码线程不安全吗？
【发布时间】：2015-11-03 11:48:54
【问题描述】：

我编写了代码来理解 CyclicBarrier。

我的应用程序模拟选举。每一轮都会选出票数较少的候选人，该候选人将被淘汰出局。

来源：

class ElectoralCommission {
    public volatile boolean hasWinner;
    public volatile String winner;
    private List<String> candidates;
    private Map<String, Integer> results = new ConcurrentHashMap<>();

    ElectoralCommission(List<String> candidates) {
        this.candidates = candidates;
    }

    public void acceptVote(int index) {
        Integer currentResult = results.get(candidates.get(index));
        results.put(candidates.get(index), currentResult == null ? 1 : currentResult + 1);
    }

    public void clearResults() {
        results.clear();
    }

    public synchronized List<String> getCandidates() {
        return candidates;
    }

    public String getWinner() {
        return winner;
    }

    public boolean isHasWinner() {
        return hasWinner;
    }

    public void printResults() {
        System.out.println("result:");
        for (Map.Entry<String, Integer> entry : results.entrySet()) {
            System.out.println("key=" + entry.getKey() + ", value=" + entry.getValue());
        }
    }

    public void removeWeakCandidates() {
        int minVoteValue = Collections.min(results.values());
        int maxVoteValue = Collections.max(results.values());
        if (maxVoteValue == minVoteValue) {
            System.out.println("all candidates got same votes count");
        } else {
            List<String> loosers = results.entrySet().stream().filter(i -> i.getValue() == minVoteValue).map(i -> i.getKey()).collect(Collectors.toList());
            candidates.removeAll(loosers);
            if (candidates.size() == 1) {
                hasWinner = true;
                winner = candidates.get(0);
            }
        }
        clearResults();
    }
}

class Voter implements Runnable {
    ElectoralCommission electoralCommission;
    CyclicBarrier cyclicBarrier;

    Voter(CyclicBarrier cyclicBarrier, ElectoralCommission electoralCommission) {
        this.cyclicBarrier = cyclicBarrier;
        this.electoralCommission = electoralCommission;
    }

    @Override
    public void run() {
        while (!electoralCommission.hasWinner) {
            List<String> candidates = electoralCommission.getCandidates();
            int voteIndex = new Random().nextInt(candidates.size());
            electoralCommission.acceptVote(voteIndex);
            try {
                cyclicBarrier.await(); //wait while all voters vote
            } catch (InterruptedException e) {
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            } catch (BrokenBarrierException e) {
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            }
        }
    }
}

class Main {
    public static final int VOTE_COUNT = 10000;

    public static void main(String[] args) {
        List<String> candidates = new ArrayList<>();
        candidates.add("candidate_1");
        candidates.add("candidate_2");
        candidates.add("candidate_3");
        candidates.add("candidate_4");
        candidates.add("candidate_5");
        candidates.add("candidate_6");
        candidates.add("candidate_7");
        candidates.add("candidate_8");
        candidates.add("candidate_9");
        candidates.add("candidate_10");
        ElectoralCommission electoralCommission = new ElectoralCommission(candidates);
        CyclicBarrier cyclicBarrier = new CyclicBarrier(VOTE_COUNT, new Summarizer(electoralCommission));
        for (int i = 0; i < VOTE_COUNT; i++) {
            new Thread(new Voter(cyclicBarrier, electoralCommission)).start();
        }

    }
}

class Summarizer implements Runnable {
    ElectoralCommission electoralCommission;

    Summarizer(ElectoralCommission electoralCommission) {
        this.electoralCommission = electoralCommission;
    }

    @Override
    public void run() {
        System.out.println("summarizer");
        electoralCommission.printResults();
        electoralCommission.removeWeakCandidates();
        if (electoralCommission.hasWinner) {
            String winnerName = electoralCommission.getWinner();
            System.out.println("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");
            System.out.println("candidate " + winnerName + " won!");
            System.out.println("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");
        }
    }
}

样本输出：

summarizer
result:
key=candidate_2, value=984
key=candidate_3, value=1004
key=candidate_1, value=1018
key=candidate_6, value=1002
key=candidate_7, value=969
key=candidate_4, value=1031
key=candidate_5, value=1038
key=candidate_10, value=915
key=candidate_8, value=1003
key=candidate_9, value=1034
summarizer
result:
key=candidate_2, value=1096
key=candidate_3, value=1133
key=candidate_1, value=1088
key=candidate_6, value=1144
key=candidate_7, value=1078
key=candidate_4, value=1136
key=candidate_5, value=1100
key=candidate_8, value=1147
key=candidate_9, value=1078
summarizer
result:
key=candidate_2, value=1429
key=candidate_3, value=1488
key=candidate_1, value=1430
key=candidate_6, value=1445
key=candidate_4, value=1410
key=candidate_5, value=1397
key=candidate_8, value=1399
summarizer
result:
key=candidate_2, value=1668
key=candidate_3, value=1697
key=candidate_1, value=1684
key=candidate_6, value=1661
key=candidate_4, value=1679
key=candidate_8, value=1610
summarizer
result:
key=candidate_2, value=2079
key=candidate_3, value=1954
key=candidate_1, value=2003
key=candidate_6, value=2022
key=candidate_4, value=1941
summarizer
result:
key=candidate_2, value=2504
key=candidate_3, value=2481
key=candidate_1, value=2500
key=candidate_6, value=2514
summarizer
result:
key=candidate_2, value=3299
key=candidate_1, value=3284
key=candidate_6, value=3417
summarizer
result:
key=candidate_2, value=4961
key=candidate_6, value=5036
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
candidate candidate_6 won!
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!

如您所见，我的最后一个汇总器调用输出如下：

summarizer
result:
key=candidate_2, value=4961
key=candidate_6, value=5036
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
candidate candidate_6 won!
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!

我有 10000 个选民，但 4961 + 5036 = 9997。

哪个输了 3 票？

附言

我尝试了带锁的解决方案：

class ElectoralCommission {
    public volatile boolean hasWinner;
    public volatile String winner;
    private List<String> candidates;
    private Map<String, AtomicInteger> results = new ConcurrentHashMap<>();

    ElectoralCommission(List<String> candidates) {
        this.candidates = candidates;
    }

    public void acceptVote(int index) {
        AtomicInteger currentResult = results.get(candidates.get(index));
       if(currentResult!=null){
           currentResult.incrementAndGet();
       }
    }

    public void clearResults() {
        results.clear();
    }

    public synchronized List<String> getCandidates() {
        return candidates;
    }

    public String getWinner() {
        return winner;
    }

    public boolean isHasWinner() {
        return hasWinner;
    }

    public void printResults() {
        System.out.println("result:");
        for (Map.Entry<String, AtomicInteger> entry : results.entrySet()) {
            System.out.println("key=" + entry.getKey() + ", value=" + entry.getValue());
        }
    }

    public void removeWeakCandidates() {
        int minVoteValue = Collections.min(results.values().stream().map(i->i.intValue()).collect(Collectors.toList()));
        int maxVoteValue = Collections.max(results.values().stream().map(i->i.intValue()).collect(Collectors.toList()));
        if (maxVoteValue == minVoteValue) {
            System.out.println("all candidates got same votes count");
        } else {
            List<String> loosers = results.entrySet().stream().filter(i -> i.getValue().intValue() == minVoteValue).map(i -> i.getKey()).collect(Collectors.toList());
            candidates.removeAll(loosers);
            if (candidates.size() == 1) {
                hasWinner = true;
                winner = candidates.get(0);
            }
        }
        clearResults();
    }
}

class Voter implements Runnable {
    ElectoralCommission electoralCommission;
    CyclicBarrier cyclicBarrier;

    Voter(CyclicBarrier cyclicBarrier, ElectoralCommission electoralCommission) {
        this.cyclicBarrier = cyclicBarrier;
        this.electoralCommission = electoralCommission;
    }

    @Override
    public void run() {
        while (!electoralCommission.hasWinner) {
            List<String> candidates = electoralCommission.getCandidates();
            int voteIndex = new Random().nextInt(candidates.size());
            electoralCommission.acceptVote(voteIndex);
            try {
                cyclicBarrier.await(); //wait while all voters vote
            } catch (InterruptedException e) {
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            } catch (BrokenBarrierException e) {
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            }
        }
    }
}

class Main {
    public static final int VOTE_COUNT = 10000;

    public static void main(String[] args) {
        List<String> candidates = new ArrayList<>();
        candidates.add("candidate_1");
        candidates.add("candidate_2");
        candidates.add("candidate_3");
        candidates.add("candidate_4");
        candidates.add("candidate_5");
        candidates.add("candidate_6");
        candidates.add("candidate_7");
        candidates.add("candidate_8");
        candidates.add("candidate_9");
        candidates.add("candidate_10");
        ElectoralCommission electoralCommission = new ElectoralCommission(candidates);
        CyclicBarrier cyclicBarrier = new CyclicBarrier(VOTE_COUNT, new Summarizer(electoralCommission));
        for (int i = 0; i < VOTE_COUNT; i++) {
            new Thread(new Voter(cyclicBarrier, electoralCommission)).start();
        }

    }
}

class Summarizer implements Runnable {
    ElectoralCommission electoralCommission;

    Summarizer(ElectoralCommission electoralCommission) {
        this.electoralCommission = electoralCommission;
    }

    @Override
    public void run() {
        System.out.println("summarizer");
        electoralCommission.printResults();
        electoralCommission.removeWeakCandidates();
        if (electoralCommission.hasWinner) {
            String winnerName = electoralCommission.getWinner();
            System.out.println("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");
            System.out.println("candidate " + winnerName + " won!");
            System.out.println("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");
        }
    }
}

但我明白

java.util.concurrent.BrokenBarrierException
    at java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:250)
    at java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:362)
    at lection2.task3.Voter.run(Main.java:88)

第 88 行：cyclicBarrier.await();

【问题讨论】：

检查每个点的不变量，而不仅仅是结束。也许这会帮助你查明哪里出了问题。
您有共享变量，这些变量在多个线程执行时交错。使用原子变量或synchronized 来避免这种情况。
您在每次迭代中都失去了选票。第一个少了 3 个，倒数第二个只丢了 2 个。从我的角度来看，存在并发问题。正如@TimBiegeleisen 所说，您应该保护变量，因此 2 个线程不会同时访问它们。
@Tim Biegeleisen 是的 - 如果标记接受投票同步 - 一切正常

标签： java multithreading concurrency cyclicbarrier

【解决方案1】：

方法 acceptVote(int index)

似乎有错误

尝试使用 ConcurrentHashMap.putIfAbsent 和 AtomicInteger 方法重写它：

private final ConcurrentHashMap<String, AtomicInteger> results = new ConcurrentHashMap<>();

public void acceptVote(int index) {
    AtomicInteger currentResult = results.get(candidates.get(index));
    if (currentResult == null) {
        currentResult = results.putIfAbsent(candidates.get(index), new AtomicInteger(1));
        if (currentResult != null) {
            currentResult.getAndIncrement();
        }
    } else {
        currentResult.getAndIncrement();
    }
}

这样的……

【讨论】：

但是如果有票数的话，值也不会缺席吧？
@Sergey Morozov 请提供更多详细信息
如果在ElectoralCommission 构造函数中播种results 集合，则acceptVote 变为results.get(candidates.get(index)).getAndIncrement();
在这种情况下是否也可以使用 AtomicInteger？

【解决方案2】：

我尝试了几次同步候选人名单，结果一致且符合预期：

class Main {
    public static final int VOTE_COUNT = 10000;

    public static void main(String[] args) {
        List<String> candidates = Collections.synchronizedList(new ArrayList<>());
      .....
}

虽然不是最好的解决方案。

更好的解决方案是仅同步对已接受选民的访问。可以这样做：

class ElectoralCommission {
  //....

    public synchronized void acceptVote(int index) {
        Integer currentResult = results.get(candidates.get(index));
        results.put(candidates.get(index), currentResult == null ? 1 : currentResult + 1);
}

但这不是最好的方法，因为它会加锁整个对象。更好的方法是使用锁：

    class ElectoralCommission {
  //....
        private ReentrantLock lock = new ReentrantLock(); 

        public void acceptVote(int index) {
            try {
               lock.lock();
               Integer currentResult = results.get(candidates.get(index));
               results.put(candidates.get(index), currentResult == null ? 1 : currentResult + 1);
           }finally {
               lock.unlock();
           }
        }

输出 1：

result:
key=candidate_6, value=3315
key=candidate_7, value=3315
key=candidate_10, value=3370
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
candidate candidate_10 won!
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!

【讨论】：

这可能有效，但我们真的需要同步集合吗？
@TimBiegeleisen - 不是真的。你说的对。这是我尝试的第一件事，我正在完善解决方案。但是为了响应的完整性，我也添加了给定的解决方案。请编辑或提出改进建议:)。
我认为candidates 集合上的访问模式没有任何问题（在我看来，CyclicBarrier 实现在一个独占运行的单个线程中保护了对它的所有写访问）。但我确实看到了 acceptVote 实现的问题，我会鼓励 OP 研究 ConcurrentMap 和 AtomicInteger 的组合。

【解决方案3】：

如果您在 ElectoralCommission 构造函数中播种 results 集合，则基于 Sergey 的回答：

ElectoralCommission(List<String> candidates) {
    this.candidates = candidates;
    for (String candidate : candidates) {
        results.put(candidate, new AtomicInteger());
    }
}

那么acceptVote就变成了：

public void acceptVote(int index) {
    results.get(candidates.get(index)).getAndIncrement();
}

【讨论】：

results.get(candidates.get(index)) 可以返回null
正如我在回答中指出的那样，如果 results 使用默认值播种，则不会。
@Sean ：使用ExecutorService 可以保护您免受Thread API 的影响，但是看到Voter 实现要求所有10k Voters 立即处于活动状态（所有线程都需要命中CyclicBarrier)、ExecutorService 必须仔细选择才能实际生成 10k 并发线程。有人可能会争辩说，在这种情况下选择合适的ExecutorService 类似于手动生成Threads。例如。低于 10k 的 fixedSizeThreadPool 不会削减它。
啊，我错过了所有线程都需要同时处于活动状态。将删除该位。

【解决方案4】：

以下变体适用于原子：

class ElectoralCommission {
    public volatile boolean hasWinner;
    public volatile String winner;
    private List<String> candidates;
    private Map<String, AtomicInteger> results = new ConcurrentHashMap<>();

    private Map<String, AtomicInteger> initializeMap() {
        candidates.stream().forEach(i -> results.put(i, new AtomicInteger()));
        return results;
    }

    ElectoralCommission(List<String> candidates) {
        this.candidates = candidates;
        results = initializeMap();
    }

    public void acceptVote(int index) {
        results.get(candidates.get(index)).incrementAndGet();
    }

    public void clearResults() {
        results.clear();
        initializeMap();
    }

    public List<String> getCandidates() {
        return candidates;
    }

    public String getWinner() {
        return winner;
    }

    public void printResults() {
        System.out.println("result:");
        for (Map.Entry<String, AtomicInteger> entry : results.entrySet()) {
            System.out.println("key=" + entry.getKey() + ", value=" + entry.getValue());
        }
    }

    public void removeWeakCandidates() {
        int minVoteValue = Collections.min(results.values().stream().map(i->i.intValue()).collect(Collectors.toList()));
        int maxVoteValue = Collections.max(results.values().stream().map(i->i.intValue()).collect(Collectors.toList()));
        if (maxVoteValue == minVoteValue) {
            System.out.println("all candidates got same votes count");
        } else {
            List<String> loosers = results.entrySet().stream().filter(i -> i.getValue().intValue() == minVoteValue).map(i -> i.getKey()).collect(Collectors.toList());
            candidates.removeAll(loosers);
            if (candidates.size() == 1) {
                hasWinner = true;
                winner = candidates.get(0);
            }
        }
        clearResults();
    }
}

class Voter implements Runnable {
    ElectoralCommission electoralCommission;
    CyclicBarrier cyclicBarrier;

    Voter(CyclicBarrier cyclicBarrier, ElectoralCommission electoralCommission) {
        this.cyclicBarrier = cyclicBarrier;
        this.electoralCommission = electoralCommission;
    }

    @Override
    public void run() {
        while (!electoralCommission.hasWinner) {
            List<String> candidates = electoralCommission.getCandidates();
            int voteIndex = new Random().nextInt(candidates.size());
            electoralCommission.acceptVote(voteIndex);
            try {
                cyclicBarrier.await(); //wait while all voters vote
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }
}

class Main {
    public static final int VOTE_COUNT = 10000;

    public static void main(String[] args) {
        List<String> candidates = new ArrayList<>();
        candidates.add("candidate_1");
        candidates.add("candidate_2");
        candidates.add("candidate_3");
        candidates.add("candidate_4");
        candidates.add("candidate_5");
        candidates.add("candidate_6");
        candidates.add("candidate_7");
        candidates.add("candidate_8");
        candidates.add("candidate_9");
        candidates.add("candidate_10");
        ElectoralCommission electoralCommission = new ElectoralCommission(candidates);
        CyclicBarrier cyclicBarrier = new CyclicBarrier(VOTE_COUNT, new Summarizer(electoralCommission));
        for (int i = 0; i < VOTE_COUNT; i++) {
            new Thread(new Voter(cyclicBarrier, electoralCommission)).start();
        }

    }
}

class Summarizer implements Runnable {
    static int roundNumber = 1;
    ElectoralCommission electoralCommission;

    Summarizer(ElectoralCommission electoralCommission) {
        this.electoralCommission = electoralCommission;
    }

    @Override
    public void run() {
        System.out.println("Round " + roundNumber++ );
        electoralCommission.printResults();
        electoralCommission.removeWeakCandidates();
        if (electoralCommission.hasWinner) {
            String winnerName = electoralCommission.getWinner();
            System.out.println("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");
            System.out.println("candidate " + winnerName + " won!");
            System.out.println("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");
        }
    }
}

【讨论】：

请注意，您的acceptVote 方法效率可能会提高。在这里，每次调用acceptVote 时都会实例化一个AtomicInteger，不管它是否需要（而且大多数时候不需要）。也许您应该像其他 cmets 中所讨论的那样，在选举轮开始时将每个候选人的 Map 预初始化为 0。
@GPI 同意你的意见