Столкнулся с проблемой реализации паттерна 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 должен получать откуда-то данные. Точку вхождения я придумать никак не могу.