我有一个WebAPI,应该也能接收来自RabbitMQ的消息。我使用了this教程,因为我知道有时IIS喜欢终止长时间运行的任务(尽管我还没有在服务器上测试过,也许它不起作用)。我有一个处理通过RabbitMQ接收到的消息的服务。我遇到的第一个问题是,我无法将其注入到
BackgroundService
类中,因此我使用了IServiceScopeFactory
。现在,我必须从两个队列中消耗消息,并且据我所知,最佳实践是为此使用两个通道。但是处理是在一个服务中完成的。BackgroundService:public class ConsumeRabbitMQHostedService : BackgroundService
{
private IConnection _connection;
private IModel _firstChannel;
private IModel _secondChannel;
private RabbitConfigSection _rabbitConfig;
public IServiceScopeFactory _serviceScopeFactory;
public ConsumeRabbitMQHostedService(IOptions<RabbitConfigSection> rabbitConfig, IServiceScopeFactory serviceScopeFactory)
{
_rabbitConfig = rabbitConfig.Value;
_serviceScopeFactory = serviceScopeFactory;
InitRabbitMQ();
}
private void InitRabbitMQ()
{
var factory = new ConnectionFactory { HostName = _rabbitConfig.HostName, UserName = _rabbitConfig.UserName, Password = _rabbitConfig.Password };
_connection = factory.CreateConnection();
_firstChannel = _connection.CreateModel();
_firstChannel.ExchangeDeclare(_rabbitConfig.DefaultExchange, ExchangeType.Topic);
_firstChannel.QueueDeclare(_rabbitConfig.Queues.ConsumeQueues.FirstItemsConsumeQueue, true, false, false, null);
_firstChannel.QueueBind(_rabbitConfig.Queues.ConsumeQueues.FirstItemsConsumeQueue, _rabbitConfig.DefaultExchange, "*.test.queue", null);
_firstChannel.BasicQos(0, 1, false);
_secondChannel = _connection.CreateModel();
_secondChannel.ExchangeDeclare(_rabbitConfig.DefaultExchange, ExchangeType.Topic);
_secondChannel.QueueDeclare(_rabbitConfig.Queues.ConsumeQueues.SecondItemsConsumeQueue, true, false, false, null);
_secondChannel.QueueBind(_rabbitConfig.Queues.ConsumeQueues.SecondItemsConsumeQueue, _rabbitConfig.DefaultExchange, "*.test.queue", null);
_secondChannel.BasicQos(0, 1, false);
_connection.ConnectionShutdown += RabbitMQ_ConnectionShutdown;
}
protected override Task ExecuteAsync(CancellationToken stoppingToken)
{
stoppingToken.ThrowIfCancellationRequested();
var firstConsumer = new EventingBasicConsumer(_firstChannel);
var secondConsumer = new EventingBasicConsumer(_secondChannel);
using (var scope = _serviceScopeFactory.CreateScope())
{
IIntegrationService scoped = scope.ServiceProvider.GetRequiredService<IIntegrationService>();
firstConsumer.Received += (ch, ea) =>
{
// received message
var content = System.Text.Encoding.UTF8.GetString(ea.Body.ToArray());
// handle the received message
HandleFirstMessage(content, scoped);
_firstChannel.BasicAck(ea.DeliveryTag, false);
};
firstConsumer.Shutdown += OnConsumerShutdown;
firstConsumer.Registered += OnConsumerRegistered;
firstConsumer.Unregistered += OnConsumerUnregistered;
firstConsumer.ConsumerCancelled += OnConsumerConsumerCancelled;
_firstChannel.BasicConsume(_rabbitConfig.Queues.ConsumeQueues.FirstItemsConsumeQueue, false, firstConsumer);
}
using (var scope = _serviceScopeFactory.CreateScope())
{
IIntegrationService scoped = scope.ServiceProvider.GetRequiredService<IIntegrationService>();
secondConsumer.Received += (ch, ea) =>
{
// received message
var content = System.Text.Encoding.UTF8.GetString(ea.Body.ToArray());
// handle the received message
HandleSecondMessage(content, scoped);
_secondChannel.BasicAck(ea.DeliveryTag, false);
};
secondConsumer.Shutdown += OnConsumerShutdown;
secondConsumer.Registered += OnConsumerRegistered;
secondConsumer.Unregistered += OnConsumerUnregistered;
secondConsumer.ConsumerCancelled += OnConsumerConsumerCancelled;
_secondChannel.BasicConsume(_rabbitConfig.Queues.ConsumeQueues.SecondItemsConsumeQueue, false, secondConsumer);
}
return Task.CompletedTask;
}
private void HandleFirstMessage(string content, IIntegrationService integrationService)
{
List<StockImportDto> dataToImport = JsonConvert.DeserializeObject<List<StockImportDto>>(content);
integrationService.ImportFirst(dataToImport);
}
private void HandleSecondMessage(string content, IIntegrationService integrationService)
{
List<Import901Data> importData = JsonConvert.DeserializeObject<List<Import901Data>>(content);
integrationService.ImportSecond(importData);
}
private void OnConsumerConsumerCancelled(object sender, ConsumerEventArgs e) { }
private void OnConsumerUnregistered(object sender, ConsumerEventArgs e) { }
private void OnConsumerRegistered(object sender, ConsumerEventArgs e) { }
private void OnConsumerShutdown(object sender, ShutdownEventArgs e) { }
private void RabbitMQ_ConnectionShutdown(object sender, ShutdownEventArgs e) { }
public override void Dispose()
{
_firstChannel.Close();
_connection.Close();
base.Dispose();
}
}
在服务中我遇到了以下问题:
System.ObjectDisposedException:“无法访问已释放的上下文实例。此错误的常见原因是释放了从依赖项注入解析的上下文实例,然后在应用程序的其他位置尝试使用同一上下文实例。如果您正在对上下文实例调用“Dispose”或将其包装在using语句中,则可能会发生这种情况。如果您正在使用依赖项注入,则应让依赖项注入容器处理上下文实例的释放。 对象名称:“IntegrationDbContext”。
DbContext
被注入到 IIntegrationService
中。如果我理解得正确,两个(甚至一个)服务实例共享 DbContext
,当其中一个完成时会释放 DbContext
。我尝试不创建两个实例(所有代码都在一个 using
内),尝试使 IIntegrationService
瞬态,尝试异步执行所有操作(这是初始版本,将其改为同步以进行测试) - 仍然出现相同的错误。我该怎么办?这样做正确吗?
更新1. 在 Startup
中的 ConfigureServices
:
public void ConfigureServices(IServiceCollection services)
{
var rabbitConfigSection =
Configuration.GetSection("Rabbit");
services.Configure<RabbitConfigSection>(rabbitConfigSection);
services.AddDbContext<SUNDbContext>(options =>
options.UseSqlServer(Configuration.GetConnectionString("DefaultConnection")));
services.AddCors();
services.AddSwaggerGen(c =>
{
c.SwaggerDoc("v1", new OpenApiInfo
{
Title = "My API",
Version = "v1"
});
});
services.AddRabbit(Configuration);
services.AddHostedService<ConsumeRabbitMQHostedService>();
services.AddControllers();
services.AddTransient<IIntegrationService, IntegrationService>();// it's transient now, same error with scoped
}