Поделиться одной переменной между потоками?

Question

У меня есть такой сценарий:

class MyClass {
   Producer p;
   Consumer c;
    public static void main(String[] args){
        BlockingQueue q = new LinkedBlockingQueue();
        p = new Producer(q);
        c = new Consumer(q);
        Thread t = new Thread(p);
        t.start();
        new Thread(c).start();
        while (true) {
            if (!p.getContinuer()) {
                c.setContinuer(false);
                System.out.println("here:"+p.getContinuer().toString());
                break;
            }
        }
        System.out.println("finish all");
     }
}

class Producer implements Runnable {
private final BlockingQueue queue;
private AtomicBoolean continuer = new AtomicBoolean(true);
   public Boolean getContinuer() {
      return continuer.get();
   }
    @Override
    public void run() {
      while(true){
        //open socket
        //read data from socket
        queue.put(data);
        if(end){
            System.out.println("Shutting down Producer");
            continuer.getAndSet(false);
        }
      }
    }
}

class Consumer implements Runnable {
    private final BlockingQueue queue;
    private static AtomicBoolean continuer = new AtomicBoolean(true);

    public void setContinuer(Boolean continuerr) {
        continuer = new AtomicBoolean(continuerr);
    }

    public Boolean getContinuer() {
        return continuer.get();
    }

    @Override
    public void run() {
          while (getContinuer()) {
             //Do some work
             consume(queue.take());
          }
          System.out.println("shut down Consumer");
    }
}

вот что я получаю:

Завершение работы производителя

здесь ложь

закончить все

это означает, что потребитель все еще работает, а переменная "continueer" не обновляется.

я также видел это и это сообщений, я пробовал, но ничего не изменилось.

в чем моя проблема?

РЕДАКТИРОВАТЬ: я изменил свой код, и, очевидно, потребитель заблокирован ( ожидание, если в этой очереди нет элементов. ) при попытке прочитать данные из BlockingQueue (см. потребление (queue.take ()); в классе потребителей).

Answer 1

Ну,

public void setContinuer(Boolean continuerr) {
    continuer = new AtomicBoolean(continuerr);
}

выглядит неправильно.Почему бы вам

public void setContinuer(boolean continuerr) {
    continuer.set(continuerr);
}

---

Кроме того, если переменная end не является энергозависимой, она может быть кэширована в потоках.

---

Нам нужно будет увидетьбольше кода(Или, по крайней мере, компиляция кода.) Поскольку этот код работает должным образом:

import java.util.concurrent.*;

public class MyClass {
    static Producer p;
    static Consumer c;

    public static void main(String[] args) {
        BlockingQueue q = new LinkedBlockingQueue();
        p = new Producer();
        c = new Consumer();
        Thread t = new Thread(p);
        t.start();
        new Thread(c).start();
        while (true) {
            if (!p.getContinuer()) {
                c.setContinuer(false);
                break;
            }
        }
        System.out.println("finish all");
    }
}

class Producer implements Runnable {
    private boolean end = true;
    private BlockingQueue queue;
    private AtomicBoolean continuer = new AtomicBoolean(true);

    public boolean getContinuer() {
        return continuer.get();
    }

    @Override
    public void run() {
        while (true) {
            // open socket
            // read data from socket
            if (end) {
                System.out.println("Shutting down Producer");
                continuer.getAndSet(false);
                break;
            }
        }
    }
}

class Consumer implements Runnable {
    private BlockingQueue queue;
    private static AtomicBoolean continuer = new AtomicBoolean(true);

    public void setContinuer(Boolean continuerr) {
        continuer = new AtomicBoolean(continuerr);
    }

    public Boolean getContinuer() {
        return continuer.get();
    }

    @Override
    public void run() {
        while (getContinuer()) {
            // Do some work
        }
        System.out.println("shut down Consumer");
    }
}

Он печатает

Shutting down Producer
finish all
shut down Consumer

---

Относительно вашего редактирования:

я изменил свой код, и, очевидно, потребитель блокируется (ожидает) при попытке прочитать данные из BlockingQueue (см. Потребление (queue.take ()); в классе потребителя).

Вы должны после вашего c.setContinuer(false); сделать consumerThread.interrupt().Поток-потребитель, заблокированный в методе read, сгенерирует исключение InterruptedException (которое можно игнорировать), а затем выйдет из цикла и завершит работу корректно.

Answer 2

проверяли ли вы

if (!p.getContinuer())

взято? getContinuer для продюсера не определен в вашем примере. Я бы рекомендовал вставить оператор отладки после

c.setContinuer(false);

чтобы получить значение c.getContinuer ()

Редактировать: Вы должны использовать один Atomicboolean для производителя и потребителя, как предлагают другие ответы.

Answer 3

Как сказал aioobe, вы создаете новый логический объект каждый раз, когда устанавливаете его.Это плохо.

Хуже того, у вас есть дискретные переменные-продолжатели для производителя и потребителя.Итак, если вы установите логическое значение источника для false, почему это повлияет на потребителя?Вы хотели сказать Consumer.getAndSet(false)?Код кажется немного запутанным.