У меня есть один Job Distributor, который публикует сообщения на разных Channels.
Кроме того, я хочу иметь два (и более в будущем) Consumers, которые работают над различными задачами и запускаются на разных машинах. (В настоящее время у меня есть только один и нужно его масштабировать)
Назовите эти задачи (просто примеры):
-
FIBONACCI(генерирует числа фибоначчи) -
RANDOMBOOKS(генерирует случайные предложения для написания книги)
Эти задачи работают до 2-3 часов, а должны быть разделены поровну на каждый Consumer.
Каждый пользователь может иметь x параллельные потоки для работы над этими задачами.
Поэтому я говорю: (эти числа являются просто примерами и будут заменены переменными)
- Машина 1 может использовать 3 параллельные задания для
FIBONACCIи 5 параллельных заданий дляRANDOMBOOKS - Машина 2 может использовать 7 параллельных заданий для
FIBONACCIи 3 параллельных заданий дляRANDOMBOOKS
Как я могу это достичь?
Нужно ли начинать x Threads для каждого Channel для прослушивания на каждом Consumer?
Когда мне это нужно?
Мой текущий подход только для одного Consumer: Start x Threads для каждой задачи - каждый поток - это реализация Defaultconsumer Runnable. В методе handleDelivery я вызываю basicAck(deliveryTag,false), а затем выполняю работу.
Далее: я хочу отправить некоторые задачи специальному потребителю. Как я могу достичь этого в сочетании с справедливым распределением, как указано выше?
Это мой код для publishing
String QUEUE_NAME = "FIBONACCI";
Channel channel = this.clientManager.getRabbitMQConnection().createChannel();
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
channel.basicPublish("", QUEUE_NAME,
MessageProperties.BASIC,
Control.getBytes(this.getArgument()));
channel.close();
Это мой код для Consumer
public final class Worker extends DefaultConsumer implements Runnable {
@Override
public void run() {
try {
this.getChannel().queueDeclare(this.jobType.toString(), true, false, false, null);
this.getChannel().basicConsume(this.jobType.toString(), this);
this.getChannel().basicQos(1);
} catch (IOException e) {
// catch something
}
while (true) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Control.getLogger().error("Exception!", e);
}
}
}
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] bytes) throws IOException {
String routingKey = envelope.getRoutingKey();
String contentType = properties.getContentType();
this.getChannel().basicAck(deliveryTag, false); // Is this right?
// Start new Thread for this task with my own ExecutorService
}
}
В этом случае класс Worker запускается дважды: один раз для FIBUNACCI и один раз для RANDOMBOOKS
UPDATE
Как сказано в ответах, RabbitMQ не будет лучшим решением для этого, но лучше всего использовать подход Couchbase или MongoDB. Я новичок в этих системах, есть ли кто-нибудь, кто мог бы объяснить мне, как это будет достигнуто?
