Многопоточность производителя/потребителя

Фон

Не имея денег на учебу в школе, я работаю в ночных сменах на платной дороге и использую Интернет, чтобы научить себя некоторым навыкам кодирования, надеясь на лучшую работу завтра или онлайн-продажу какого-либо приложения, которое я делаю. Длинные ночи, немногие клиенты.

Я рассматриваю многопоточность как тему, поскольку я сталкиваюсь с большим количеством кода в литературе (например, Android SDK), который использует его, но я все еще нахожу его неясным.

Дух

Мой подход на этом этапе: попытайтесь закодировать самый простой пример многопоточности, о котором я могу думать, немного ударить головой о стену и посмотреть, могу ли я растянуть мой мозг, чтобы разместить новый образ мышления. Я подвергаю себя своим пределам, надеюсь, превзойти их. Не стесняйтесь критиковать дико, вплоть до nitpicking, и указывать лучшие способы делать то, что я пытаюсь сделать.

Цель

• Get some advice on how to do the above, based on my efforts so far (code provided)

Упражнение

Здесь определяется область действия:

Определение

Создайте два класса, которые работают в тандеме по созданию объектов данных и их потреблению. Один Thread создает объекты и доставляет их в разделяемое пространство для другого, чтобы забрать и потреблять. Позвольте вызвать производящий поток Producer, потребляющий поток Consumer и общее пространство SharedSpace. Акт создания объектов для потребления другими может быть ассимилирован с помощью аналогии с этим сценарием:

`Producer`    (a busy mum making chocolate-covered cakes for his child, up to a limit)
`Consumer`    (a hungry child waiting to eat all cakes the mum makes, until told to stop)
`SharedSpace` (a kitchen table on which the cakes are put as soon as they become ready)
`dataValue`   (a chocolate-dripping cake which MUST be eaten immediately or else...)

Чтобы упростить упражнение, я решил не разрешать маме готовить, когда ребенок ест свой торт. Она просто ждет, пока ребенок закончит свой торт и мгновенно сделает еще один, до определенного предела, для хорошего воспитания. Суть упражнения состоит в том, чтобы практиковать сигнализацию Thread по достижению любого concurrency вообще. Напротив, я сосредоточен на идеальной сериализации, без опроса или "я могу пойти еще?". чеки. Полагаю, мне придется процитировать последующее упражнение, в котором мать и ребенок "работают" параллельно.

Подход

• Попросите мои классы реализовать интерфейс Runnable, чтобы у них была точка ввода кода

• Используйте мои классы как аргументы конструктора для объектов Thread, которые создаются и запускаются из программы main entry точка

• Убедитесь, что программа main не заканчивается перед Thread с помощью Thread.join()

• Задайте ограничение на количество раз, когда Producer создаст данные для Consumer

• Согласитесь на значение дозорного значения, которое Produce будет использовать для завершения передачи данных.

• Логарифмическое получение блокировок на общих ресурсах и событиях производства/потребления данных, включая окончательное отключение рабочих потоков

• Создайте один SharedSpace объект из программы main и передайте его каждому работнику перед запуском

• Хранить private ссылку на объект SharedSpace внутри каждого рабочего

• Предоставить защиту и сообщения, чтобы описать условие готовности Consumer к употреблению до того, как будут созданы какие-либо данные.

• Остановите Producer после заданного количества итераций

• Остановите Consumer после того, как он прочитает значение часового

код

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
class Consumer extends Threaded {
  public Consumer(SharedSpace sharedSpace) {
    super(sharedSpace);
  }
  @Override
  public void run() {
    super.run();
    int consumedData = 0;
    while (consumedData != -1) {
      synchronized (sharedSpace) {
        logger.info("Acquired lock on sharedSpace.");
        consumedData = sharedSpace.dataValue;
        if (consumedData == 0) {
          try {
            logger.info("Data production has not started yet. "
                + "Releasing lock on sharedSpace, "
                + "until notification that it has begun.");
            sharedSpace.wait();
          } catch (InterruptedException interruptedException) {
            logger.error(interruptedException.getStackTrace().toString());
          }
        } else if (consumedData == -1) {
          logger.info("Consumed: END (end of data production token).");
        } else {
          logger.info("Consumed: {}.", consumedData);
          logger.info("Waking up producer to continue data production.");
          sharedSpace.notify();
          try {
            logger.info("Releasing lock on sharedSpace "
                + "until notified of new data availability.");
            sharedSpace.wait();
          } catch (InterruptedException interruptedException) {
            logger.error(interruptedException.getStackTrace().toString());
          }
        }
      }
    }
    logger.info("Signing off.");
  }
}
class Producer extends Threaded {
  private static final int N_ITERATIONS = 10;
  public Producer(SharedSpace sharedSpace) {
    super(sharedSpace);
  }
  @Override
  public void run() {
    super.run();
    int nIterations = 0;
    while (nIterations <= N_ITERATIONS) {
      synchronized (sharedSpace) {
        logger.info("Acquired lock on sharedSpace.");
        nIterations++;
        if (nIterations <= N_ITERATIONS) {
          sharedSpace.dataValue = nIterations;
          logger.info("Produced: {}", nIterations);
        } else {
          sharedSpace.dataValue = -1;
          logger.info("Produced: END (end of data production token).");
        }
        logger.info("Waking up consumer for data consumption.");
        sharedSpace.notify();
        if (nIterations <= N_ITERATIONS) {
          try {
            logger.info("Releasing lock on sharedSpace until notified.");
            sharedSpace.wait();
          } catch (InterruptedException interruptedException) {
            logger.error(interruptedException.getStackTrace().toString());
          }
        }
      }
    }
    logger.info("Signing off.");
  }
}
class SharedSpace {
  volatile int dataValue = 0;
}
abstract class Threaded implements Runnable {
  protected Logger logger;
  protected SharedSpace sharedSpace;
  public Threaded(SharedSpace sharedSpace) {
    this.sharedSpace = sharedSpace;
    logger = LoggerFactory.getLogger(this.getClass());
  }
  @Override
  public void run() {
    logger.info("Started.");
    String workerName = getClass().getName();
    Thread.currentThread().setName(workerName);
  }
}
public class ProducerConsumer {
  public static void main(String[] args) {
    SharedSpace sharedSpace = new SharedSpace();
    Thread producer = new Thread(new Producer(sharedSpace), "Producer");
    Thread consumer = new Thread(new Consumer(sharedSpace), "Consumer");
    producer.start();
    consumer.start();
    try {
      producer.join();
      consumer.join();
    } catch (InterruptedException interruptedException) {
      interruptedException.printStackTrace();
    }
  }
}

Журнал выполнения

Consumer - Started.
Consumer - Acquired lock on sharedSpace.
Consumer - Data production has not started yet. Releasing lock on sharedSpace, until notification that it has begun.
Producer - Started.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 1
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 1.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 2
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 2.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 3
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 3.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 4
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 4.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 5
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 5.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 6
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 6.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 7
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 7.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 8
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 8.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 9
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 9.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 10
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 10.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: END (end of data production token).
Producer - Waking up consumer for data consumption.
Producer - Signing off.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: END (end of data production token).
Consumer - Signing off.

Вопрос

• Правильно ли это? (например, использует ли он правильные языковые инструменты, правильный подход, содержит ли он какой-то глупый код,...)

Но это выглядит "правильно"?

Я спрашиваю о правильности, даже если результат "выглядит хорошо", потому что вы не можете себе представить, сколько раз что-то пошло не так в моем тестировании "один раз", а не "другое" (например, когда потребитель начал сначала, когда Продюсер никогда не уходит после производства часового и т.д.). Я научился не требовать правильности от "успешного запуска". Напротив, я стал очень подозрительным в отношении псевдопараллельного кода! (по определению это не является даже параллельным! 0

Расширенные ответы

Хороший вопрос фокусируется только на one requested piece of advice (выше), но не стесняйтесь упоминать какие-либо другие темы в вашем ответе, если хотите:

• Как я могу протестировать параллельный код, когда буду вводить следующие мои попытки?

• Какие инструменты могут помочь мне в разработке и отладке? Рассмотрим, что я использую Eclipse

• Будет ли изменен подход, если я разрешаю Producer продолжать производить, причем каждое производство занимает некоторое переменное время, в то время как Consumer потребляет все, что становится доступным? Следует ли запереть замок в другом месте? Должна ли сигнализация измениться с этой парадигмы ожидания/уведомления?

• Этот метод делает вещи устаревшими и должен ли я изучать что-то еще? Из этой таблички я понятия не имею, что происходит "в реальном мире Java"

Следующие шаги

• Куда мне идти дальше? Я видел понятие "фьючерсы", упомянутое где-то, но я мог бы использовать нумерованный список тем для работы последовательно, с педалогически упорядоченным, со ссылками на связанные учебные ресурсы.

Тино Сино

Ответы

public class Consumer implements Runnable {
    private final BlockingQueue<Integer> queue;
    public Consumer(BlockingQueue<Integer> queue) {
       this.queue = queue;
    }
    @Override
    public void run() {
       while (true) {
          int consumedData = queue.take();
          if (consumedData ==  Producer.FINAL_VALUE) {
              logger.info("Consumed: END (end of data production token).");
              break;
          }
          logger.info("Consumed: {}.", consumedData);
       }
       logger.info("Signing off.");
    }
}

public class Producer implements Runnable {
    public static final int FINAL_VALUE = -1;
    private final BlockingQueue<Integer> queue;
    public Producer(BlockingQueue<Integer> queue) {
       this.queue = queue;
    }
    @Override
    public void run() {
       for (int nIterations = 0; nIterations <= N_ITERATIONS; nIterations++) {
          logger.info("Produced: {}", nIterations);
          queue.put(nIterations);
       }
       queue.put(FINAL_VALUE);
       logger.info("Produced: END (end of data production token).");
       logger.info("Signing off.");
    }
}

public class ProducerConsumer {
    public static void main(String[] args) {
       // you can add an int argument to the LinkedBlockingQueue constructor
       // to only allow a certain number of items in the queue at one time
       BlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer>();
       Thread producer = new Thread(new Producer(queue), "Producer");
       Thread consumer = new Thread(new Consumer(queue), "Consumer");
       // start and join go here
    }
}

void stateDependentMethod() throws InterruptedException {
      // condition predicate must be guarded by lock
      synchronized(lock) {
          while (!conditionPredicate())
            lock.wait();
          // object is now in desired state
       }
  }

class Producer implements Runnable { 
     private volatile boolean stopRequested ;

     public void run() {
        while(true){
           if(stopRequested )
                // get out of the loop
         }
     }

     public void stop(){
        stopRequested  = true;
        // arrange to  interrupt the Producer thread here.
     }
 }