我正在用.net核心和rabbitMQ客户端测试一些东西。一个服务将发布数百万行数据,另一个服务将接收和解析这些数据。每行可能需要4-10秒的解析时间。所以重点是
1)能够利用更多的机器来运行接收服务,以及
2)使用运行服务的机器的所有cpu核心。
所以1看起来很明显。使用更多服务器来解析队列中的更多消息。但我想不出第二个是怎么做的。
我写了一个简单的.net核心程序,假设我想利用3个核心:
static async Task Main(string[] args)
{
var factory = new ConnectionFactory() { HostName = "localhost", DispatchConsumersAsync = true };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "hello1",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
var tasks = new List<Task>() {
ReceiveEvent(channel, 1),
ReceiveEvent(channel, 2),
ReceiveEvent(channel, 3)
};
await Task.WhenAll(tasks);
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
//
channel.Close();
connection.Close();
}
}
private static async Task ReceiveEvent(IModel channel, int consumerNo)
{
var consumer = new AsyncEventingBasicConsumer(channel);
int i = 1;
consumer.Received += async (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
Console.WriteLine($"Received {i++} lines in consumer {consumerNo} --- {message}");
Thread.Sleep(500 * consumerNo);
//
channel.BasicAck(ea.DeliveryTag, false);
await Task.Yield();
};
channel.BasicConsume(queue: "hello1",
autoAck: false,
consumer: consumer);
}不管我运行了多少次,只有1号消费者捕捉到了事件。我需要在Task.WhenAll线路上进行这样的呼叫吗?或者,如果事件是异步触发的,.net核心运行时会利用所有的cpu核心吗?
非常感谢
发布于 2018-08-14 16:02:57
根据我使用RabbitMQ的经验,每当收到消息时,都会创建一个消息侦听器线程。在处理此消息时,队列不会将另一条消息推送到listner。在您的情况下,您必须首先接收消息,然后才能在多个线程中处理。
https://stackoverflow.com/questions/51835958
复制相似问题