为什么 MassTransit 只在一个线程上消费消息
Why is MassTransit only consuming messages on one single thread
我在常规控制台应用程序中有以下简单代码,并希望(部分)命令并行使用。我认为 UseConcurrencyLimit
会设置并发线程数。我看到的是 RabbitMQ 确实有 10 条未确认的消息,但是消费者连续使用它们,每条 console.writeline 之间有一秒钟的暂停。我一定是遗漏了一些明显的东西,但我不明白。
public static class EventHandler
{
public static void Run()
{
var personQueueName = "RabbitMqPoc.Person";
var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
cfg.UseConcurrencyLimit(10);
var host = cfg.Host(new Uri("rabbitmq://localhost"), h =>
{
h.Username("guest");
h.Password("guest");
});
cfg.ReceiveEndpoint(host, personQueueName, e =>
{
e.UseConcurrencyLimit(10);
e.PrefetchCount = 10;
e.Consumer<PersonConsumer>();
});
});
var personSendEndpoint = busControl.GetSendEndpoint(new Uri($"rabbitmq://localhost/{personQueueName}")).Result;
busControl.Start();
foreach (var x in Enumerable.Range(1, 20))
{
personSendEndpoint.Send(new PersonUpdated() { Name = "Mina Ives", Key = Guid.NewGuid() });
}
Console.ReadLine();
busControl.Stop();
}
}
internal class PersonConsumer : IConsumer<IPersonUpdated>
{
public async Task Consume(ConsumeContext<IPersonUpdated> context)
{
Thread.Sleep(1000);
Console.WriteLine($"Updated {context.Message.Name}");
}
}
将 Thread.Sleep(1000);
更改为 await Task.Delay(1000);
可解决您遇到的问题。
Thread.Sleep 出于某种原因对 TPL 造成严重破坏。
我在常规控制台应用程序中有以下简单代码,并希望(部分)命令并行使用。我认为 UseConcurrencyLimit
会设置并发线程数。我看到的是 RabbitMQ 确实有 10 条未确认的消息,但是消费者连续使用它们,每条 console.writeline 之间有一秒钟的暂停。我一定是遗漏了一些明显的东西,但我不明白。
public static class EventHandler
{
public static void Run()
{
var personQueueName = "RabbitMqPoc.Person";
var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
cfg.UseConcurrencyLimit(10);
var host = cfg.Host(new Uri("rabbitmq://localhost"), h =>
{
h.Username("guest");
h.Password("guest");
});
cfg.ReceiveEndpoint(host, personQueueName, e =>
{
e.UseConcurrencyLimit(10);
e.PrefetchCount = 10;
e.Consumer<PersonConsumer>();
});
});
var personSendEndpoint = busControl.GetSendEndpoint(new Uri($"rabbitmq://localhost/{personQueueName}")).Result;
busControl.Start();
foreach (var x in Enumerable.Range(1, 20))
{
personSendEndpoint.Send(new PersonUpdated() { Name = "Mina Ives", Key = Guid.NewGuid() });
}
Console.ReadLine();
busControl.Stop();
}
}
internal class PersonConsumer : IConsumer<IPersonUpdated>
{
public async Task Consume(ConsumeContext<IPersonUpdated> context)
{
Thread.Sleep(1000);
Console.WriteLine($"Updated {context.Message.Name}");
}
}
将 Thread.Sleep(1000);
更改为 await Task.Delay(1000);
可解决您遇到的问题。
Thread.Sleep 出于某种原因对 TPL 造成严重破坏。