ReentrantLock - lock.unlock () не снимает блокировку

Question

Я пытаюсь создать приложение с одним производителем задач (начальник) и несколькими потребителями задач (сотрудники). Босс будет добавлять новые задачи время от времени, и сотрудники должны очистить очередь задач. Проблема заключается в следующем: только один потребитель выполняет всю работу, и как только она заканчивается, другие начинают работать.

Я пробовал использовать ReentrantLock, но, похоже, ничего не изменилось.

package threads;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

public class Store {

    public static ReentrantLock lock = new ReentrantLock();


    public static void main(String[] args) {
        AtomicInteger tasksList = new AtomicInteger(7);
        Runnable boss = () -> {
            System.out.println("Boss: Good morning.");
            List<Runnable> employeeList = generateEmployeesList(5, tasksList);
            for (Runnable employee : employeeList) {
                employee.run();         
            }
            while(tasksList.get() > 0) {

                try {
                    lock.lock();
                    System.out.println("Work more!");
                    tasksList.incrementAndGet();

                } finally {
                    lock.unlock();
                    try {
                        Thread.sleep(1500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
            System.out.println("Boss: Time to go home.");
        };

        System.out.println("= Starting =");
        boss.run();

    }

    public static List<Runnable> generateEmployeesList(int amountOfEmployees, AtomicInteger tasksList){
        List<Runnable> employeeList = new ArrayList<Runnable>();
        for (int i = 0; i < amountOfEmployees; i++) {
            employeeList.add(createEmployee(i+1, tasksList));
        }
        return employeeList;
    }

    public static Runnable createEmployee(int employeeNumber, AtomicInteger tasksList) {
        return () -> {
            System.out.println("Thread #" + (employeeNumber) + " Started. Missing tasks: " + tasksList.get());
            while(tasksList.get() > 0) {

                try {
                    lock.lock();
                    System.out.println("Thread #" + (employeeNumber) + ". Task completed: " + tasksList.decrementAndGet() +" tasks left.");
                } finally {
                    lock.unlock();
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

            }
            System.out.println("Thread #" + (employeeNumber) +": Finished.");
        };
    }

}

Результаты (неверно):

= Starting =
Boss: Good morning.
Thread #1 Started. Missing tasks: 7
Thread #1. Task completed: 6 tasks left.
Thread #1. Task completed: 5 tasks left.
Thread #1. Task completed: 4 tasks left.
Thread #1. Task completed: 3 tasks left.
Thread #1. Task completed: 2 tasks left.
Thread #1. Task completed: 1 tasks left.
Thread #1. Task completed: 0 tasks left.
Thread #1: Finished.
Thread #2 Started. Missing tasks: 0
Thread #2: Finished.
Thread #3 Started. Missing tasks: 0
Thread #3: Finished.
Thread #4 Started. Missing tasks: 0
Thread #4: Finished.
Thread #5 Started. Missing tasks: 0
Thread #5: Finished.
Boss: Time to go home.

Ожидаемые результаты (случайный потребитель уменьшает переменную задачи):

= Starting =
Boss: Good morning.
Thread #4 Started. Missing tasks: 7
Thread #4. Task completed: 6 tasks left.
Thread #2 Started. Missing tasks: 6
Thread #2. Task completed: 5 tasks left.
Thread #4. Task completed: 4 tasks left.
Thread #1 Started. Missing tasks: 4
...
Thread #1: Finished.
Thread #2: Finished.
Thread #3: Finished.
...
Boss: Time to go home.

Есть идеи, что я делаю не так? Что я могу сделать, чтобы решить мою проблему?

Я изменил приложение, чтобы использовать потоки вместо Runnables. Но я все еще получаю ту же проблему.

class Lock {
    private static final ReentrantLock lock = new ReentrantLock();
    public static final ReentrantLock getLock() {
        return lock;
    }
}

public class Store {


    public static void main(String[] args) {    
        Boss b = new Boss(5, 7);
        b.start();
    }

    public static List<Thread> generateEmployeesList(int amountOfEmployees, AtomicInteger tasksList){
        List<Thread> employeeList = new ArrayList<Thread>();
        for (int i = 0; i < amountOfEmployees; i++) {
            employeeList.add(employeeFactory(i+1, tasksList));
        }
        return employeeList;
    }


    public static Thread employeeFactory(int employeeNumber, AtomicInteger tasksList) {
        return new Employee(employeeNumber, tasksList);
    }


}

class Employee extends Thread {
    int number = 0;
    AtomicInteger tasksList;
    public Employee(int number, AtomicInteger tasksList) {
        this.number = number;
        this.tasksList = tasksList;
    }
    @Override
    public void start() {
        System.out.println("Thread #" + number + " Started. Missing tasks: " + tasksList.get());
        while(tasksList.get() > 0) {

            try {
                Lock.getLock().lock();
                System.out.println("Thread #" + number + ". Task completed: " + tasksList.decrementAndGet() +" tasks left.");
            } finally {
                Lock.getLock().unlock();
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

        }
        System.out.println("Thread #" + number +": Finished.");
    }

}

class Boss extends Thread {
    int employees = 0;
    AtomicInteger tasks;

    public Boss(int employees, int numberOfTasks) {
        this.employees = employees;
        this.tasks = new AtomicInteger(numberOfTasks);
    }

    public void start() {
        List<Thread> employeeList = Store.generateEmployeesList(employees, tasks);
        System.out.println("Boss: Good Morning!");
        for (Thread employee : employeeList) {
            employee.start();
        }
        while(tasks.get() > 0 ) {
            Lock.getLock().lock();
            try {
            System.out.println("Keep working!");
            tasks.incrementAndGet();
            } finally {
                Lock.getLock().unlock();
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }

    }
}

Answer 1

Есть несколько проблем с этим кодом. Вы не понимаете, как использовать Runnables. Метод Runnable # run () просто запускает код, который вы создаете в потоке, в котором вы сейчас находитесь. Вам нужно создать отдельный поток для их запуска, если вы хотите некоторого параллелизма.

// Change this
employee.start();
// to this:
Thread t = new Thread(employee);
t.start();

Во-вторых, как отметил комментатор, вы неправильно блокируете доступ к Atomic. Таким образом, несмотря на то, что он является поточно-ориентированным в рамках отдельных операций сам по себе, он не согласован для нескольких вызовов, особенно если ваше поведение зависит от предыдущих результатов, которые вы делаете. Вот почему, если вы сделаете одно изменение выше, вы увидите что-то вроде следующего:

= Starting =
Boss: Good morning.
Work more!
Thread #2 Started. Missing tasks: 7
Thread #1 Started. Missing tasks: 8
Thread #4 Started. Missing tasks: 7
Thread #3 Started. Missing tasks: 7
Thread #1. Task completed: 7 tasks left.
Thread #5 Started. Missing tasks: 7
Thread #5. Task completed: 6 tasks left.
Thread #2. Task completed: 5 tasks left.
Thread #4. Task completed: 4 tasks left.
Thread #3. Task completed: 3 tasks left.
Thread #5. Task completed: 2 tasks left.
Thread #1. Task completed: 1 tasks left.
Thread #2. Task completed: 0 tasks left.
Thread #4. Task completed: -1 tasks left.
Thread #3. Task completed: -2 tasks left.
Boss: Time to go home.
Thread #5: Finished.
Thread #1: Finished.
Thread #4: Finished.
Thread #2: Finished.
Thread #3: Finished.

Между наблюдением, что остались задачи

tasksList.get() > 0

и «взятие» одного задания

tasksList.incrementAndGet();

другой поток мог выполнить задачу. Это то, что вы видите с отрицательными показателями. Обе операции должны существовать в цикле, или должен использоваться какой-то другой метод синхронизации, например встроенные методы синхронизации объекта:

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

public class Store {

    public static Object sync = new Object();

    public static void main(String[] args) {
        AtomicInteger tasksList = new AtomicInteger(7);
        Runnable boss = () -> {
            System.out.println("Boss: Good morning.");
            List<Runnable> employeeList = generateEmployeesList(5, tasksList);
            for (Runnable employee : employeeList) {
                Thread t = new Thread(employee);
                t.start();
            }

            synchronized (sync) {
                while(tasksList.get() > 0) {
                    try {                        
                        System.out.println("Work more!");
                        tasksList.incrementAndGet();
                    } finally {
                        try {
                            sync.wait(1500);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }

            System.out.println("Boss: Time to go home.");
        };

        System.out.println("= Starting =");
        boss.run();

    }

    public static List<Runnable> generateEmployeesList(int amountOfEmployees, AtomicInteger tasksList){
        List<Runnable> employeeList = new ArrayList<Runnable>();
        for (int i = 0; i < amountOfEmployees; i++) {
            employeeList.add(createEmployee(i+1, tasksList));
        }
        return employeeList;
    }

    public static Runnable createEmployee(int employeeNumber, AtomicInteger tasksList) {
        return () -> {
            System.out.println("Thread #" + (employeeNumber) + " Started. Missing tasks: " + tasksList.get());
            synchronized (sync) {
                while (tasksList.get() > 0) {
                    try {
                        System.out.println("Thread #" + (employeeNumber) + ". Task completed: "
                                + tasksList.decrementAndGet() + " tasks left.");
                    } finally {
                        try {
                            sync.wait(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
            System.out.println("Thread #" + (employeeNumber) +": Finished.");
        };
    }
}

Уступая:

= Starting =
Boss: Good morning.
Thread #2 Started. Missing tasks: 7
Thread #3 Started. Missing tasks: 7
Work more!
Thread #4 Started. Missing tasks: 7
Thread #1 Started. Missing tasks: 7
Thread #3. Task completed: 7 tasks left.
Thread #5 Started. Missing tasks: 8
Thread #2. Task completed: 6 tasks left.
Thread #5. Task completed: 5 tasks left.
Thread #1. Task completed: 4 tasks left.
Thread #4. Task completed: 3 tasks left.
Thread #2. Task completed: 2 tasks left.
Thread #1. Task completed: 1 tasks left.
Thread #3. Task completed: 0 tasks left.
Thread #4: Finished.
Thread #5: Finished.
Boss: Time to go home.
Thread #3: Finished.
Thread #2: Finished.
Thread #1: Finished.