此外,我想要有两个(以及以后)消费者在不同的任务上工作并在不同的机器上运行. (目前我只有一个,需要扩大它)

我们来命名这些任务(只是例子)：

FIBONACCI(产生斐波那契数)
> RANDOMBOOKS(生成随机句子来写一本书)

这些任务长达2-3个小时,应平均分配给每个消费者.

每个消费者都可以使用x个并行线程来处理这些任务.
所以我说：(这些数字只是例子,将被变量所取代)

>机器1可以为FIBONACCI消耗3个并行作业,为RANDOMBOOKS消耗5个并行作业
>机器2可以为FIBONACCI消耗7个并行作业,并为RANDOMBOOKS消耗3个并行作业

我该如何实现？

我必须启动每个频道的X线程来听每个消费者？

我什么时候要确认？

我目前只针对一个消费者的方法是：为每个任务启动x线程 – 每个线程都是实现Runnable的Defaultconsumer.在handleDelivery方法中,我调用basicAck(deliveryTag,false)然后执行工作.

进一步：我想把一些任务送给一个特殊的消费者.如何结合上述公平分配实现？

这是我的出版代码

String QUEUE_NAME = "FIBONACCI";

Channel channel = this.clientManager.getRabbitMQConnection().createChannel();

channel.queueDeclare(QUEUE_NAME,true,false,null);

channel.basicpublish("",QUEUE_NAME,MessageProperties.BASIC,Control.getBytes(this.getArgument()));

channel.close();

这是我的消费者代码

public final class Worker extends DefaultConsumer implements Runnable {
    @Override
    public void run() {

        try {
            this.getChannel().queueDeclare(this.jobType.toString(),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

    }
}

在这种情况下,工作人员启动了两次：一次为FIBUNACCI,一次为RANDOMBOOKS

UPDATE

正如答案所述,RabbitMQ不是最好的解决方案,但是Couchbase或MongoDB pull方法将是最好的.我对这些系统很新,有没有人可以向我解释,这将如何实现？