0

Дали задачу, но чего-то никак не могу понять как ее реализовать.

Есть producer, например, 25 потоков должно быть. Есть consumer, например, 5 потоков должно быть.

producer посылает в очередь сообщение, получает оттуда String возвращается в producer.

consumer должен вычитывать эту очередь.

Размер сообщений, который будет оправлен, например, 500, после чего, видимо, ExecutorService/ThreadPool должен умереть для producer. Для consumer он должен работать до того момента, пока очередь не будет пуста.

Я всё не могу понять, как сделать так, чтобы 1) все потоки стартовали одновременно 2) как запустить новые потоки в consumer

Вот так у меня примерно в main

public class Main{

private ExecutorService prodService= Executors.newFixedThreadPool(25, prodFactory);
private ExecutorService conService= Executors.newFixedThreadPool(5, conFactory);

 public static void main(String[] args) {

        ApplicationContext context = new ClassPathXmlApplicationContext("beans.xml");
        Starter bean = (Starter) context.getBean("starter");
        bean.doit();
 }

 private void doit() {
        for (int i = 0; i < 10; i++) {
            prodService.execute(() -> {
                 for (int i = 0; i < 10; i++) {
                    System.out.println(Thread.currentThread().getName());
                    makeSmth();
                 }
            });
        conService.execute(() -> {doSmthTo();});
        }
 }

}

По факту, там у меня еще много чего страшного и не понятного, но я не знаю куда это всё внедрять)

7
  • producer посылает в очередь сообщение, получает оттуда String возвращается в producer. не понятно как и зачем producer'у что то получать
    – tym32167
    6 авг 2019 в 13:15
  • мне вот тут послений пример приглянулся
    – tym32167
    6 авг 2019 в 13:16
  • если придете к какому то нормальному результату, будет неплохо добавить пример здесь
    – tym32167
    6 авг 2019 в 13:18
  • @tym32167, знать бы, нормальный это пример или нет ) Получать он должен для того, что задача такая была поставлена. А реализацию из вашего примера я видел, но она мне не очень помогла( Всё еще ищу что-то путное. Кстати, почему тот пример тогда не добавить в stack-wiki?
    – Dred
    6 авг 2019 в 13:27
  • знать бы что такое stack-wiki )
    – tym32167
    6 авг 2019 в 13:48

2 ответа 2

0

Реализовал примерно так.

class Stater {

public static boolean STOP = false;
private Producer producer;
private Consumer consumer;
private ExecutorService prodService= Executors.newFixedThreadPool(PRODUCER_NUMBER, taxiFactory);
private ExecutorService conService= Executors.newFixedThreadPool(CONSUMER_NUMBER, clientFactory);


 public static void main(String[] args) {

        ApplicationContext context = new ClassPathXmlApplicationContext("beans.xml");
        Starter bean = (Starter) context.getBean("starter");
        bean.doit();
 }

 private void doit() {

    for (int i = 0; i < PRODUCER_NUMBER; i++) {
        prodService.execute(() -> producer.get());
        conService.execute(() -> consumer.run());
    }

    Starter.STOP = true;
    producerService.shutdown();
    consumerService.shutdown();
 }
}


class Common {

private Queue<Message> emergencyQueue;
private BlockingQueue<Message> blockingQueue;

   public void insertOrder(Message message) {
        if (!blockingQueue.offer(message)) {
            emergencyQueue.add(message);
        }
    }

   public Message getOrder() {

        if (emergencyQueue.isEmpty()) {
            if (!blockingQueue.isEmpty()) {
                return blockingQueue.poll();
            } else {
                return null;
            }
        } else {
            return emergencyQueue.poll();
        }
    }

   public boolean shouldStop() {
        return blockingQueue.isEmpty() && emergencyQueue.isEmpty() && Starter.STOP;
    }   
}


class Consumer implements Runnable{

private Common common;

   public void run(){
       common.insertOrder(new Message());
    }
}

class Producer implements Runnable{

private Common common;

   public void run(){

          while (!common.shouldStop()) {

              Message message=common.getOrder();

              if (message == null) {
               Thread.sleep(new Random().nextInt(TIME_TO_WAIT));
            }

          }
    }
}
-1

LinkedBloсkingQueue - один из вариантов быстрого решения вопроса

1
  • вообще не тянет на решение. В newFixedThreadPool итак используется очередь, неважно связанная или случайная.
    – Dred
    8 авг 2019 в 12:22

Ваш ответ

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

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