Как убедиться, что все потоки завершены перед вызовом notify?

Question

Я пытаюсь убедиться, что все мои потоки класса A закончились до вызова notify.В тот момент, когда один поток завершил свою работу, он вызывает notify (потоки класса B), пока другие потоки A все еще работают.Если это произойдет, B-потоки начнут работать, что приведет к изменению условия для остальных A-потоков.

Я пытался использовать блоки синхронизации, но, думаю, я их неправильно использую.

Идея состоит в том,что A заполняет массив до его полного заполнения и уведомляет B, чтобы он мог снова очистить массив.

public class A extends Thread {

    public static Object lockA = new Object();

    private ArrayList<String> list;

    public A(ArrayList<String> list) {
        this.list = list;
    }

    public void run(){

        while(true){
            synchronized (A.lockA){
                if(list.size() < 10){
                    list.add("A");
                    System.out.println(currentThread().getName() + " " + list);
                }else{
                    synchronized (B.lockB){
                        B.lockB.notifyAll();

                    }
                    return;
                }
            }

        }
    }

}

public class B extends Thread {

    public static Object lockB = new Object();

    private ArrayList<String> list;

    public B(ArrayList<String> list) {
        this.list = list;
    }

    public void run(){


            synchronized (B.lockB){
                try {
                    B.lockB.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                while (list.size() > 0){
                    list.remove("A");
                    System.out.println(currentThread().getName() + " " + list);
                }
                return;
            }
    }
}

public class Main {
    public static void main(String[] args) {
        ArrayList<String> list = new ArrayList<>();

        A a = new A(list);
        A aa = new A(list);
        A aaa = new A(list);
        B b = new B(list);
        B bb = new B(list);
        B bbb = new B(list);
        B bbbb = new B(list);

        a.start();
        aa.start();
        aaa.start();
        b.start();
        bb.start();
        bbb.start();
        bbbb.start();
    }
}

Answer 1

Я бы деформировал список в классе, который разрешает поточно-ориентированные обновления списка:

class ListWraper {
    //See also https://docs.oracle.com/javase/8/docs/api/java/util/Collections.html#synchronizedList-java.util.List-
    private volatile List<String> list;

    ListWraper() {
        list = new ArrayList<>();
    }

    //add to list
    synchronized boolean add(String s){
        return list.add(s);
    }

    //remove from list
    synchronized boolean remove(String s){
        return list.remove(s);
    }


    //get a defensive copy of list
    List<String> getList() {
        return new ArrayList<>(list);
    }
}

Поток, который использует ListWarper:

class ListUpdater extends Thread {

    private static String abc = "ABCDEFGHIJKLMNOPQRSTUVWXYZ";
    private static int threadsCouter = 0;             //used for
    private final int threadNumber = threadsCouter++; //printing only
    private final Random rd = new Random();
    private final ListWraper list;

    ListUpdater(ListWraper list) {
        this.list = list;
    }

    @Override
    public void run(){
         for(int i = 0; i < 10; i++){
             //add random character to list 
             list.add(String.valueOf( abc.charAt(rd.nextInt(abc.length()))));
             try {
                TimeUnit.MILLISECONDS.sleep(100 + rd.nextInt(500)); //simulate long process 
            } catch (InterruptedException ex) { ex.printStackTrace();   }
         }
         System.out.println("Thread number " + threadNumber + " finished");
    }
}

И протестируйте его, используя:

public class Main{

    public static void main(String args[]) throws InterruptedException {
        ListWraper list = new ListWraper();
        ListUpdater updater1 = new ListUpdater(list);
        ListUpdater updater2 = new ListUpdater(list);
        updater1.start();
        updater2.start();
        updater1.join(); //wait for thread to finish 
        updater2.join(); //wait for thread to finish 
        System.out.println("All finished ");
    }
}

Выход:

Тема № 1 завершена
Тема № 0 завершена
Все закончил

Полный исполняемый код можно найти здесь .

Другой простой альтернативой является использование общего CountDownLatch , как показано здесь .

Answer 2

Я пытался использовать блоки синхронизации, но, думаю, я их неправильно использую.

В общем, да. Для синхронизации между несколькими потоками лучше использовать высокоуровневые примитивы, которые можно найти в пакете java.util.concurrent .

Я пытаюсь убедиться, что все мои потоки класса A закончились до вызова notify.

CountDownLatch подходит непосредственно к вашему делу.

Answer 3

Если lockB не защищает какое-либо состояние, в котором поток может захотеть ждать изменений или может потребоваться уведомить другой поток об изменениях, использование его в сочетании с wait и notify не имеет никакого смысла.

        synchronized (B.lockB){
            try {
                B.lockB.wait();

Вы никогда не должны звонить wait, пока не проверите, чтобы убедиться, что то, чего вы ждете, еще не произошло. Причина, по которой этот код находится внутри блока synchronized, заключается именно в том, что вы можете выполнить эту проверку и ждать, если и только если вам нужно подождать.

                synchronized (B.lockB){
                    B.lockB.notifyAll();
                }

Это довольно сложно. Почему вы звоните lockB.notifyAll(), если вы не изменили ничего, защищенного этой блокировкой, о которой вам нужно уведомить другие темы?

Похоже, вы не понимаете, как на самом деле работает семантика wait / notify. Дело в том, что вам нужно только подождать, если что-то еще не произошло, и вы никогда не сможете быть уверены, что это уже произошло (потому что планировщики непредсказуемы) без удержания какой-либо блокировки. Но вам нужно снять блокировку, прежде чем ждать, создавая состояние гонки, если какой-то другой поток делает то, что вы ожидаете, после того, как вы сняли блокировку, но до того, как вы начали ждать. Функция wait - это атомарная функция, которая снимает блокировку и ждет, когда что-то произойдет. Атомность избегает этого состояния гонки.

Однако у вас нет разблокированного замка, защищающего то, что вы ждете. Это разрушает логику функции атомной разблокировки и ожидания и приводит к коду, который всегда может зайти в тупик.