0

Создаю 10 потоков: те, которые делают +1, и те, которые делают -1 синхронизированного числа. Ограничение до >= 0 и <= 5, в ином случает тот, кто забирает или тот, кто кладёт, блокируется.

Проблема в том что, рано или поздно зацикливается один "Producer" или "Consumer", и вечно выскакивает сообщение, что он был остановлен. Чаще всего это происходит, когда число доходит до 0.

Имею предположение, что я плохо знаком с concurrent.locks и использую его неправильно, либо где-то в логике ошибка. Также потоки почему-то не понимают, что когда число уже равно 5, его нельзя прибавлять и нужно блокироваться, но происходит так, что они всё равно увеличивают число и не блокируют.

import java.util.*;
import java.util.concurrent.locks.ReentrantLock;

public class stream {
    static int number=0;
    public static void main(String[] args) throws IllegalMonitorStateException, InterruptedException {
        ReentrantLock locker = new ReentrantLock();
        var consumer = new ArrayList<Thread>();
            var producer = new ArrayList<Thread>();
            for (int i = 0; i < 10; i++) {
                consumer.add(new Thread(new Consumer(locker)));
                producer.add(new Thread(new Producer(locker)));
            }
        consumer.forEach(Thread::start); //потребитель
        producer.forEach(Thread::start); //поставщик


            System.out.println("main finished");
        }
    public static synchronized void put(){
        number++;
    }
    public static synchronized void pick(){
        number--;
    }
    public static synchronized void print(){
        System.out.println(number);
    }
}
    class Consumer implements Runnable{
    ReentrantLock locker;


        public Consumer(ReentrantLock lock) {
        this.locker=lock;
        }


        @Override
    public void run() {
            try {
                for(;;) {
                    if (stream.number == 4){
                        Thread.sleep(1000);
                    }
                    Thread.sleep(500);
                    if (stream.number >= 5) {
                        System.out.println("Consumer " + Thread.currentThread().getName() + " was stopped ");
                        locker.lock();
                    }
                    else {
                        if(locker.tryLock()){

                            locker.unlock();
                            System.out.println("Consumer " + Thread.currentThread().getName() + " was continued");
                        }
                        
                        stream.put();
                        System.out.println("Consumer " + Thread.currentThread().getName() + " was put product ");
                        System.out.print("Status product: ");
                        stream.print();
                        Thread.sleep(1000);
                    }

                }
            } catch (Exception e) {

            }
        }
    }
    class Producer implements Runnable{
        ReentrantLock locker;


        public Producer(ReentrantLock lock) {
        this.locker=lock;
        }


        @Override
        public void run() {
            try {
                Thread.sleep(500);
                for(;;) {
                    if (stream.number == 1){
                        Thread.sleep(1000);
                    }
                    if (stream.number <= 0) {
                        System.out.println("Producer "+ Thread.currentThread().getName() + " was stopped ");
                        locker.lock();
                    }
                    else {
                        if(locker.tryLock()){
                            locker.unlock();
                            System.out.println("Producer "+ Thread.currentThread().getName() + " was continued");
                        }
                        Thread.sleep(1000);
                        stream.pick();
                        System.out.println("Producer " + Thread.currentThread().getName() + " was picked up product ");
                        System.out.print("Status product: ");
                        stream.print();

                    }

                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

введите сюда описание изображения

введите сюда описание изображения

1 ответ 1

1
if (stream.number == 1){

Нельзя вот так с не volatile полем работать без синхронизации.

добавь

public static synchronized void read(){
        return number;
}

Ну и с lock / unlock ты вообще не понял как работать.

if (stream.number >= 5) {
                        System.out.println("Consumer " + Thread.currentThread().getName() + " was stopped ");
                        locker.lock();
                    }

и

                 if (stream.number <= 0) {
                        System.out.println("Producer "+ Thread.currentThread().getName() + " was stopped ");
                        locker.lock();
                    }

Блокирует лок навечно, потому что другие потоки не могут сделать ему unclock, а в этом потоке условие уже никогда не изменится.

Если ты хотел, чтобы они ждали, когда кто-то добавит в пустую или уберёт из переполненной, то тебе надо было создать два условия

Condition nonEmpty = locker.newCondition();
Condition nonFull = locker.newCondition();

передать оба условия во все потоки, и (например) когда очередь пустая, то получатель должен делать

locker.lock(); 
if (empty) nonEmpty.await();
locker.unlock(); 

а отправитель должен извещать, когда добавляет

locker.lock(); 
stream.put(); 
nonEmpty.signal();
locker.unlock(); 

Ну и для переполненной очереди должно быть наоборот - отправитель ждёт nonFull, а получатель извещает, когда прочитал из очереди.

И вообщк советую открыть java.util.concurrent.locks.Condition и изучить примеры в JavaDoc

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge you have read our privacy policy.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.