Java 监视器和线程并发

Question

我正在尝试构建简单的多线程应用程序。但我对 Java 监视器感到困惑。我有很多线程想用他们的数据格式化一个数组。例如，我有超市线程（线程数据在 txt 文件中）所以第一个线程有这些产品（牛奶、奶酪、巧克力）和每个产品的国家代码 1,2,3

    SupermarketA
    Milk 1
    Cheese 2
    Chocolate 3
    SupermarketB
    Yogurt 1
    Orangle 2
    Bannana 3
    Tea 7
    Kiwi 9

我想格式化必须包含字段的数组（country_code 和 count） 所以我的数组应该是这样的

Country_code count
1              2
2              2
3              2
7              1
9              1

代码

public class SortedArray{
        private int num = 0; // num is country code
        private int count = 0;
}

这是我的 monitor 类

 public class SingleArray {
        private SortedArray[] array;
        private int arrayIndex;
        private static final int MAX_SIZE = 5;

    public SingleArray() {
        array = new SortedArray[MAX_SIZE];
        arrayIndex = 0;
        initArray();
    }

    private void initArray() {
        for (int i = 0; i < MAX_SIZE; i++) {
            array[i] = new SortedArray();
        }
    }

    public synchronized void inc(){
        awaitUnderMax();
        notifyAll();
    }

    private void awaitUnderMin(){
        while (arrayIndex == 0) try {
            wait();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public synchronized void dec(){
        awaitUnderMin();
        notifyAll();
    }

    public void add(ArrayList<Integer> count){
        for (int i = 0; i < count.size(); i++) {
            singleArray.inc();
            int num = count.get(i);
            if (singleArray.arrayIndex == 0) { // if array is empty add value to it
                singleArray.array[0].num = num;
                singleArray.array[0].count++;
                singleArray.arrayIndex++;
            } else {
                if (!isThere(num)) { // if num is a new value to array
                    singleArray.inc();
                    int index1 = singleArray.arrayIndex;
                    if (num > singleArray.array[index1 - 1].num) {
                        singleArray.inc();
                        singleArray.array[index1].num = num;
                        singleArray.inc();
                        singleArray.array[index1].count++;
                        singleArray.inc();
                        singleArray.arrayIndex++;

                        System.out.println(Thread.currentThread().getName() + " first " + singleArray.array[index1].num);
                    } else if (num < singleArray.array[index1 - 1].num) { // jei num mazesne uz paskutinia masyvo reiksme
                        int index = index1 - 1 < 0 ? index1 : index1 - 1;
                        while (index > 0 && num < singleArray.array[index].num) {
                            index--;
                        }
                        if (index != singleArray.arrayIndex) {
                            System.out.println(Thread.currentThread().getName() + " sec " + singleArray.array[index].num);
                            singleArray.array = addPos(singleArray.array, index + 1, num);
                        }
                    }
                }
            }
        }
    }

    public boolean isThere(int number){
        for(int i=0; i<singleArray.arrayIndex; i++){
            singleArray.inc();
            if(number == singleArray.array[i].num){
                singleArray.array[i].count++;
                return true;
            }
        }
        return false;
    }

    private void awaitUnderMax(){
        while (arrayIndex >= MAX_SIZE) try {
            wait();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public synchronized void removeValue(int number, int howManyItems){
        for(int i=0; i<arrayIndex; i++){
            dec();
            if(number == array[i].num){
                int numberToDelete = array[i].count - howManyItems >= 0 ? howManyItems : array[i].count;
                if(array[i].count >= numberToDelete){
                    array[i].count -= numberToDelete;
                }
                if(array[i].count == 0){
                    deleteItem(i);
                }
            }
            if(array[i].count == 0){
                deleteItem(i);
            }
        }
    }

每个线程调用add(ArrayList<Integer> count)方法 所以基本上 add 方法的作用是：

1. 查找插入新值的位置（取决于新值是大于还是小于前一个值）
2. 调用isThere(int num) 方法检查新值是否已经在数组中（如果是则增加计数singleArray.array[i].count++），否则将新值添加到数组中
3. 如果数组已满arrayIndex == MAX_SIZE 等待当前线程让其他线程递减arrayIndex（这只是代码的一部分，我还有其他基于县代码递减数组的线程）

所以最大的问题是多个线程需要同时更新单个数组（我知道在add方法中添加synchronized关键字应该可以解决这个问题但它只让一个线程运行这个方法一次！）所以有时一切都很好，但有时我得到了非常惊人的结果（例如，国家代码是 0（这是不可能的！！！），有时新的值被放置在错误的位置数组位置）。另外我认为semaphores 应该可以解决这个问题，但是monitors 可以做到这一点吗？感谢您的回答。

编辑 v2 致@Elyasin

 public Thread[] setUpShopsBuilderThreads(){
        int size = data.getSize();
        ArrayList<ArrayList<String>> a = new ArrayList<>();
        ArrayList<ArrayList<Integer>> b = new ArrayList<>();
        ArrayList<ArrayList<Double>> c = new ArrayList<>();
        Thread[] threads = new Thread[size];
        for (int i = 0; i < size; i++) {
            int tmp = data.getIndex(i);
            int range = i + 1 < size ? data.getIndex(i + 1) : data.getWaresSize();
            ArrayList<String> name = new ArrayList<>();
            ArrayList<Integer> count = new ArrayList<>();
            ArrayList<Double> price = new ArrayList<>();
            for (int j = tmp; j < range; j++) {
                name.add(data.getName(j));
                count.add(data.getCount(j));
                price.add(data.getPrice(j));
            }
            a.add(name);
            b.add(count);
            c.add(price);
        }

        procesas_1 p1 = new procesas_1(a.get(0), b.get(0), c.get(0));
        procesas_2 p2 = new procesas_2(a.get(1), b.get(1), c.get(1));
        procesas_3 p3 = new procesas_3(a.get(2), b.get(2), c.get(2));
        procesas_4 p4 = new procesas_4(a.get(3), b.get(3), c.get(3));
        procesas_5 p5 = new procesas_5(a.get(4), b.get(4), c.get(4));

        Thread worker1 = new Thread(p1);
        Thread worker2 = new Thread(p2);
        Thread worker3 = new Thread(p3);
        Thread worker4 = new Thread(p4);
        Thread worker5 = new Thread(p5);

        threads[0] = worker1;
        threads[1] = worker2;
        threads[2] = worker3;
        threads[3] = worker4;
        threads[4] = worker5;

        return threads;
    }

public static void main(String[] args) {
        Starter start = new Starter();
        start.read();
        start.printShopsData();
        start.printUserData();

        Thread[] builderThreads = start.setUpShopsBuilderThreads();

        for(int i=0; i<builderThreads.length; i++){
            builderThreads[i].start();
        }
}

Answer 1

如何使用 java 已经提供的并发安全数据集？

如果你想对其进行排序，这个看起来可能适合你：

http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ConcurrentSkipListSet.html

只需像在普通集合中一样添加它