MassTransit与RabbitMQ:MT超时第二个请求/响应
使用MassTransit和RabbitMQ处理请求/响应场景。 在做简单的请求/回复时,它会多次运行。 如果我在请求处理程序中发布消息,它将在第一个请求上工作,但请求处理程序永远不会在第二个请求上被调用,并最终超时并且消息停留在服务器队列中。
似乎我错过了一些东西; 配置可能?
该项目位于https://bitbucket.org/joeyoung/enterprise-rabbitmq
客户端配置:
ObjectFactory.Configure(cfg =>
{
cfg.AddRegistry<WebRegistry>();
cfg.For<IServiceBus>().Singleton().Use(o => ServiceBusFactory.New(sbc =>
{
// configure the bus
sbc.UseRabbitMqRouting();
sbc.ReceiveFrom("rabbitmq://localhost/entrprise_client");
// finds all the consumers in the container and register them with the bus
sbc.Subscribe(x => x.LoadFrom(ObjectFactory.Container));
}));
});
服务器配置:
var container = new Container(cfg =>
{
cfg.Scan(scan =>
{
scan.Assembly("Server.MessageHandlers");
scan.AddAllTypesOf<IConsumer>();
});
});
var bus = ServiceBusFactory.New(sbc =>
{
// configure the bus
sbc.UseRabbitMqRouting();
sbc.ReceiveFrom("rabbitmq://localhost/enterprise_server");
// finds all the consumers in the container and register them with the bus
sbc.Subscribe(x => x.LoadFrom(container));
});
// finally inject the bus into the container
container.Inject(bus);
发送请求:
bus.PublishRequest(new CreateProductCommand(correlationId, model.Name, model.Description, model.Price), x =>
{
x.HandleTimeout(10.Seconds(), () => { timedOut = true; });
x.Handle<CreateProductCommand.Response>(response => { productId = response.Id; });
});
消费请求:
public void Consume(IConsumeContext<CreateProductCommand> context)
{
Console.Out.WriteLine("Consuming Create Product");
// simulate creating a product
var productId = "products/1234";
bus.Publish(new ProductCreatedEvent(productId));
context.Respond(new CreateProductCommand.Response(context.Message.CorrelationId) { Id = productId});
}
消息:
public class CreateProductCommand : CorrelatedBy<Guid>
{
public Guid CorrelationId { get; private set; }
public string Name { get; set; }
public string Description { get; set; }
public decimal Price { get; set; }
public CreateProductCommand(Guid correlationId, string name, string description, decimal price)
{
CorrelationId = correlationId;
Name = name;
Description = description;
Price = price;
}
public class Response : CorrelatedBy<Guid>
{
public Guid CorrelationId { get; private set; }
public string Id { get; set; }
public Response(Guid correlationId)
{
CorrelationId = correlationId;
}
}
}
感谢Chris建议在IConsumeContext上使用总线。 这似乎修复了它。
因此,不要在处理程序的构造函数中注入IServiceBus,而要从上下文中获取总线。
public class CreateProductCommandHandler : Consumes<CreateProductCommand>.Context
{
public void Consume(IConsumeContext<CreateProductCommand> context)
{
Console.Out.WriteLine("Consuming Create Product");
// simulate creating a product
var productId = "products/1234";
context.Bus.Publish(new ProductCreatedEvent(productId));
context.Respond(new CreateProductCommand.Response(context.Message.CorrelationId) { Id = productId});
}
}
我知道你想和RabbitMQ一起使用MT,我放弃了,把我的爱变成了EasyNetQ,试试看( - :
链接地址: http://www.djcxy.com/p/61217.html上一篇: MassTransit with RabbitMQ: MT timing out on second Request/Response