2

Столкнулся с проблемой реализации паттерна Consumer/Producer с несколькими Consumers; хотелось бы реализовать возможность отправить в очередь сообщение одним Producer'ом и возможность чтения из очереди множеством Worker'ов. Я бы сказал, что проблема исключительно в том, что не представляю, как выстроить правильно композицию классов.

В главном треде я решил оставить всего два вызова:

public static void main(String[] args) {
        CloudQueue sqs = new CloudQueue(5);
        for(int i=0; i<10; ++i){
            sqs.send("Data");
        }
    }

Соответственно объект CloudQueue выполняет некоторые setup операции, а также создает большую очередь, в которой будут общаться подпрограммы. Я понимаю, что все идет к созданию god-object, попытаюсь это устранить.

class CloudQueue{
    CloudQueueService queue = new CloudQueueService();
    public CloudQueue(int queues){
    }
    public void send(String s){
        queue.push(s);
    }
}

CloudQueue в свою очередь создает экземпляр CloudQueueService, цель которого - попытка реализовать собственно общение producer - queue - consumers;

class CloudQueueService{
    private BlockingQueue<String> queue = new ArrayBlockingQueue<String>(10);
    private WorkerPool pool;
    public CloudQueueService(){
        WorkerPool pool = new WorkerPool(this.queue);
    }
    public void push(String s){
        queue.add(s);
    }
}

CloudQueueService уже непосредственно создает очередь, а также пулл worker'ов, которые должны непосредственно вытягивать данные из очереди и работать с ними.

Привожу пример кода WorkersPool и Worker; особо что-то по их поводу сказать не могу.

class WorkerPool{
    public WorkerPool(BlockingQueue<String> q){
        ExecutorService executor = Executors.newFixedThreadPool(5);
        for(int i=0; i<5; i++){
            Runnable worker = new Worker(q);
            executor.execute(worker);
        }
        executor.shutdown();
        while (!executor.isTerminated()){

        }
        System.out.println("Done");
    }
}

class Worker implements Runnable{
    private BlockingQueue<String> queue;
    public Worker(BlockingQueue<String> queue){
        this.queue = queue;
    }
    @Override
    public void run(){
        try{
            System.out.println(Thread.currentThread().getName() + "->" + this.queue.take());
        }catch (Exception e){
            e.printStackTrace();
        }

    }
}

Соответственно, с какими я проблемами столкнулся?

  • Слабое представление, как организовать структуру данного решения. В главном треде я хочу как можно сильнее избавиться от системных подробностей, так как в дальнейшем CloudQueue будет решать довольно сложные задачи, поэтому хотелось бы для пользователя оставить только методы send, get и инициализацию.

  • Вероятнее всего неверное понимание принципов работы с потоками, так как Producer должен получать откуда-то данные. Точку вхождения я придумать никак не могу.

1 ответ 1

1

Как я понял Вы хотите создать такую реализацию паттерна, в котором потребители будут создаваться сразу (классы реализующие потребителя могут и передаваться при конструировании consumer-producer паттерна) и их количество заранее определено. Предлагаю вот такой НАБРОСОК его, нужно доработать под ваши цели.

public class Test {
    private static boolean runningFlag = true;
    private static final int CONSUMERS_NUMBER = 5;
    private static ExecutorService executor = Executors.newFixedThreadPool(CONSUMERS_NUMBER);
    private static BlockingQueue<String> queue = new ArrayBlockingQueue<String>(10);
    public final static Test INSTANCE = new Test();
    private Test(){
        startConsumers();
    }

    public static void startConsumers(){
        for(int worker = 0; worker < CONSUMERS_NUMBER; worker++){
            executor.submit(()->{
                try {
                    while(runningFlag) {
                        String job = queue.take();
                        Thread.sleep((int)(Math.random()*1000));// do some job
                        System.out.println("Job is done:"+job);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
    }

    public static void stopConsumers(){
        runningFlag = false;
        executor.shutdown();
    }

    public static void sendJob(String someJob) throws InterruptedException {
        queue.put(someJob); // или add и т.п.
    }

    public static void main(String[] args) throws InterruptedException {
        for(int i = 100; i> 0; i--){
            Test.INSTANCE.sendJob("sf"+i);
        }
        Test.stopConsumers();
    }
}

Ваш ответ

Нажимая на кнопку «Отправить ответ», вы соглашаетесь с нашими условиями использования и подтверждаете, что прочитали и поняли наши политику конфиденциальности и нормы поведения.

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