我正在关注RabbitMQ的这个指南：https://www.rabbitmq.com/tutorials/tutorial-two-java.html。我想在一个队列上使用多个线程来模拟此功能。
如果我在启动发件人之前启动接收器，它会按预期工作，如下所示：
[*] Rcvr1 Waiting for messages...
[*] Rcvr2 Waiting for messages...
[x] Rcvr1 Received 'Hello 0'
[x] Rcvr2 Received 'Hello 1'
[x] Rcvr1 Received 'Hello 2'
[x] Rcvr2 Received 'Hello 3'
[x] Rcvr1 Received 'Hello 4'
[x] Rcvr2 Received 'Hello 5'
[x] Rcvr1 Received 'Hello 6'
[x] Rcvr2 Received 'Hello 7'
[x] Rcvr1 Received 'Hello 8'
...
但是，首先启动我的Receiver会导致只有一个线程接收消息（最后一个要启动的线程）：
[*] Rcvr2 Waiting for messages...
[*] Rcvr1 Waiting for messages...
[x] Rcvr1 Received 'Hello 9'
[x] Rcvr1 Received 'Hello 10'
[x] Rcvr1 Received 'Hello 11'
[x] Rcvr1 Received 'Hello 12'
[x] Rcvr1 Received 'Hello 13'
[x] Rcvr1 Received 'Hello 14'
[x] Rcvr1 Received 'Hello 15'
...
有趣的是，如果我启动发件人，然后启动接收器，如上所述，然后再次启动发送器（接收器处理第一批）。发送的第一个消息是串行处理的，而第二个批处理是并行处理的，或至少与剩余的线程一起处理：
[*] Rcvr1 Waiting for messages...
[*] Rcvr2 Waiting for messages...
[x] Rcvr1 Received '[Batch 1] Hello 0'
[x] Rcvr1 Received '[Batch 1] Hello 1'
[x] Rcvr1 Received '[Batch 1] Hello 2'
[x] Rcvr1 Received '[Batch 1] Hello 3'
[x] Rcvr1 Received '[Batch 1] Hello 4'
[x] Rcvr1 Received '[Batch 1] Hello 5'
[x] Rcvr1 Received '[Batch 1] Hello 6'
[x] Rcvr1 Received '[Batch 1] Hello 7'
[x] Rcvr1 Received '[Batch 1] Hello 8'
[x] Rcvr2 Received '[Batch 2] Hello 1'
[x] Rcvr1 Received '[Batch 1] Hello 9'
[x] Rcvr2 Received '[Batch 2] Hello 3'
[x] Rcvr1 Received '[Batch 1] Hello 10'
[x] Rcvr2 Received '[Batch 2] Hello 5'
[x] Rcvr1 Received '[Batch 1] Hello 11'
[x] Rcvr2 Received '[Batch 2] Hello 7'
[x] Rcvr1 Received '[Batch 1] Hello 12'
[x] Rcvr2 Received '[Batch 2] Hello 9'
[x] Rcvr1 Received '[Batch 1] Hello 13'
[x] Rcvr2 Received '[Batch 2] Hello 11'
RabbitMQ显然可以做到这一点，我不确定我做错了什么。我的简单代码如下：
寄件人
public class Send {
private final static String QUEUE_NAME = "hello";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
try (Connection connection = factory.newConnection();
Channel channel = connection.createChannel()) {
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
for(int x=0; x<100; x++) {
String message = "Hello "+x;
channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
System.out.println(" [x] Sent '" + message + "'");
}
}
}
}
接收器
package com.mawv.ingest.rabbitmq;
import com.rabbitmq.client.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
public class Recv {
private final static String QUEUE_NAME = "hello";
public static void main(String[] argv) throws Exception {
ThreadPoolExecutor rcvrPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(10);
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Runnable rcvr1 = () -> {
try {
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
System.out.println(" [*] Rcvr1 Waiting for messages...");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
Envelope envelope = delivery.getEnvelope();
String message = new String(delivery.getBody(), "UTF-8");
System.out.println(" [x] Rcvr1 Received '" + message + "'");
long deliveryTag = envelope.getDeliveryTag();
channel.basicAck(deliveryTag, true);
try {
Thread.sleep(1000);
} catch (Exception ex) { }
};
channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });
} catch(Exception ex){
ex.printStackTrace();
}
};
Runnable rcvr2 = () -> {
try {
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
System.out.println(" [*] Rcvr2 Waiting for messages...");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
Envelope envelope = delivery.getEnvelope();
String message = new String(delivery.getBody(), "UTF-8");
System.out.println(" [x] Rcvr2 Received '" + message + "'");
long deliveryTag = envelope.getDeliveryTag();
channel.basicAck(deliveryTag, true);
try {
Thread.sleep(1000);
} catch (Exception ex) {
}
};
channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {
});
} catch(Exception ex){
ex.printStackTrace();
}
};
rcvrPool.execute(rcvr1);
rcvrPool.execute(rcvr2);
}
}
我也把这个例子绑定了，就像他们描述的那样，并看到相同的结果。 https://self-learning-java-tutorial.blogspot.com/2015/09/rabbitmq-one-producer-and-multiple.html
我假设我的设置不正确。
根据RabbitMQ api：
“虽然一个Channel可以被多个线程使用，但确保只有一个线程同时执行命令是很重要的。并发执行命令可能会导致抛出UnexpectedFrameError”
首先，我认为你应该为不同的线程使用不同的通道。
最后我认为第一个线程因为空闲而被终止，因此只有第二个线程处于活动状态并完成整个工作。在这种情况下，一个线程就够了。
看看java 8的ThreadPoolExecutor api：
https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ThreadPoolExecutor.html
例如，您可以找到：
“默认情况下，即使核心线程最初只在新任务到达时创建并启动，但可以使用方法prestartCoreThread（）或prestartAllCoreThreads（）动态覆盖。如果构造非空的池，可能需要预启动线程队列”
和
“如果池当前有多个corePoolSize线程，如果多余的线程空闲时间超过keepAliveTime（参见getKeepAliveTime（TimeUnit）），则会终止多余的线程。”
你应该使用prestartAllCoreThreads（）或prestartCoreThreads（）来获取即使空闲时的核心线程，也可以使用getKeepAliveTime（TimeUnit）使它们保持活动状态即使空闲也是如此。
看起来我错过了一个关键的频道配置。这解决了我的问题：
channel.basicQos（1）;
这就是RabbitMQ对它的看法。
公平派遣
您可能已经注意到调度仍然无法完全按照我们的意愿运行。例如，在有两个工人的情况下，当所有奇怪的消息都很重，甚至消息很轻时，一个工人将经常忙，而另一个工作人员几乎不会做任何工作。那么，RabbitMQ对此一无所知，仍然会均匀地发送消息。
发生这种情况是因为RabbitMQ只是在消息进入队列时调度消息。它不会查看消费者未确认消息的数量。它只是盲目地向第n个消费者发送每个第n个消息。
为了打败我们可以使用basicQos方法和prefetchCount = 1设置。这告诉RabbitMQ一次不向一个worker发送一条消息。或者，换句话说，在处理并确认前一个消息之前，不要向工作人员发送新消息。相反，它会将它发送给下一个仍然很忙的工人。