HC_YFZX
/
LASAPlatform
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

修改rabbit

main
zhangbin 2025-11-26 15:32:18 +08:00
parent ddf1f27bf9
commit 2b18e0089b
1 changed files with 54 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

197
OpenAuth.WebApi/Model/RabbitMQService/RabbitMqListenerService.cs
View File

@ -9,173 +9,84 @@ namespace OpenAuth.WebApi.Model.RabbitMQService
{
public class RabbitMqListenerService : BackgroundService
{
private readonly ConnectionFactory _factory;
private readonly ILogger<RabbitMqListenerService> _logger;
private readonly DaHuaAiApp _app;
private IConnection _connection;
private IChannel _channel;
private readonly Dictionary<string, string> _queueRoutingMap = new();
private readonly object _lock = new();
private readonly List<QueueBinding> _bindings = new();
private readonly ILogger<RabbitMqListenerService> _logger;
private DaHuaAiApp _app;
private readonly ConnectionFactory _factory = new()
{
HostName = "123.132.248.154",
UserName = "DHCloudg1",
Password = "Cloud0#4fCraQrm",
Port = 9103
};
public RabbitMqListenerService(ILogger<RabbitMqListenerService> logger, DaHuaAiApp app)
{
_logger = logger;
_app = app;
_factory = new ConnectionFactory
{
HostName = "123.132.248.154",
UserName = "DHCloudg1",
Password = "Cloud0#4fCraQrm",
Port = 9103
//HostName = "localhost",
//UserName = "guest",
//Password = "guest",
//Port = 5672
};
}
public async Task AddQueueBindingAsync(string exchange, string exchangeType, string routingKey)
{
_bindings.Add(new QueueBinding
{
Exchange = exchange,
ExchangeType = exchangeType,
RoutingKey = routingKey
});
if (_channel != null)
{
await BindQueueAsync(_channel, exchange, exchangeType, routingKey);
}
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
_connection = await _factory.CreateConnectionAsync(stoppingToken);
_channel = await _connection.CreateChannelAsync();
//await BindQueueAsync(_channel, "processing_event", "topic", "event.trafficJunction.1.sdhc");
var queueOk = await _channel.QueueDeclareAsync(queue: "",
durable: false,
exclusive: true,
autoDelete: true);
//await _channel.QueueBindAsync(queueOk.QueueName, "processing_event", "event.eventNotAlarm.picture.1.#");
await _channel.QueueBindAsync(queueOk.QueueName, "processing_event", "event.behaviorAlarm.picture.1.#");
await _channel.QueueBindAsync(queueOk.QueueName, "storage_event", "event.trafficEvent.1.#");
var consumer = new AsyncEventingBasicConsumer(_channel);
consumer.ReceivedAsync += async (sender, ea) =>
_logger.LogInformation("RabbitMQ listener starting...");
while (!stoppingToken.IsCancellationRequested)
{
var body = ea.Body.ToArray();
var message = Encoding.UTF8.GetString(body);
//Console.WriteLine($"接收数据[processing_event]: {message}");
// Exchange 来做不同处理
switch (ea.Exchange)
try
{
case "processing_event":
await _app.AddDaHuaImg(message);
break;
case "storage_event":
await _app.AddDaHuaSmokeDetectionImg(message);
break;
default:
Console.WriteLine($"[Other:{ea.RoutingKey}] {message}");
break;
// 建立连接
_connection = await _factory.CreateConnectionAsync(stoppingToken);
_channel = await _connection.CreateChannelAsync();
_logger.LogInformation("RabbitMQ connected successfully.");
await StartConsumeAsync(_channel, stoppingToken);
// 阻塞直到断开
await Task.Delay(Timeout.Infinite, stoppingToken);
}
await Task.Yield();
};
await _channel.BasicConsumeAsync("", autoAck: true, consumer);
//foreach (var binding in _bindings)
//{
// await BindQueueAsync(_channel, binding.Exchange, binding.ExchangeType, binding.RoutingKey);
//}
catch (Exception ex)
{
_logger.LogError(ex, "RabbitMQ connection failed. Retrying in 5 seconds...");
await Task.Delay(5000, stoppingToken);
}
}
}
private async Task BindQueueAsync(IChannel channel, string exchange, string exchangeType, string routingKey)
private async Task StartConsumeAsync(IChannel channel, CancellationToken token)
{
//await channel.ExchangeDeclareAsync(exchange, exchangeType, durable: true);
//await channel.QueueDeclareAsync("", durable: true, exclusive: false, autoDelete: false);
//var queueOk = await _channel.QueueDeclareAsync(queue: "",
// durable: false,
// exclusive: true,
// autoDelete: true);
//await channel.QueueBindAsync(queueOk.QueueName, exchange, routingKey);
var q = await channel.QueueDeclareAsync("", false, true, true);
await channel.QueueBindAsync(q.QueueName, "processing_event", "event.behaviorAlarm.picture.1.#");
await channel.QueueBindAsync(q.QueueName, "storage_event", "event.trafficEvent.1.#");
var consumer = new AsyncEventingBasicConsumer(channel);
consumer.ReceivedAsync += async (sender, ea) =>
consumer.ReceivedAsync += async (_, ea) =>
{
var body = ea.Body.ToArray();
var message = Encoding.UTF8.GetString(body);
Console.WriteLine($"[processing_event]: {message}");
// Exchange 来做不同处理
switch (ea.Exchange)
var message = Encoding.UTF8.GetString(ea.Body.ToArray());
try
{
case "processing_event":
Console.WriteLine($"[processing_event]: {message}");
break;
case "thing_event":
Console.WriteLine($"[thing_event] {message}");
break;
default:
Console.WriteLine($"[Other:{ea.RoutingKey}] {message}");
break;
switch (ea.Exchange)
{
case "processing_event":
await _app.AddDaHuaImg(message);
break;
case "storage_event":
await _app.AddDaHuaSmokeDetectionImg(message);
break;
}
}
catch (Exception ex)
{
_logger.LogError(ex, "处理消息失败");
}
await Task.Yield();
};
await channel.BasicConsumeAsync("", autoAck: true, consumer);
}
//public async Task AddListenerAsync(string exchange, string exchangeType, string routingKey)
//{
// _connection = await _factory.CreateConnectionAsync();
await channel.BasicConsumeAsync(q.QueueName, autoAck: true, consumer);
// foreach (var binding in _bindings)
// {
// var channel = await _connection.CreateChannelAsync();
// // 声明交换机
// await channel.ExchangeDeclareAsync(binding.Exchange, binding.ExchangeType, durable: true);
// await channel.QueueDeclareAsync("", durable: true, exclusive: false, autoDelete: false);
// // 绑定队列到交换机
// await channel.QueueBindAsync(queue: "", exchange: binding.Exchange, routingKey: binding.RoutingKey);
// var consumer = new AsyncEventingBasicConsumer(channel);
// consumer.ReceivedAsync += async (model, ea) =>
// {
// var body = ea.Body.ToArray();
// var message = Encoding.UTF8.GetString(body);
// using var scope = _serviceProvider.CreateScope();
// foreach (var handler in handlers)
// {
// await handler.HandleMessageAsync(binding.Exchange, ea.RoutingKey, message);
// }
// await channel.BasicAckAsync(ea.DeliveryTag, multiple: false);
// };
// await channel.BasicConsumeAsync(queue: "", autoAck: false, consumer: consumer);
// _channels.Add(channel);
// }
//}
//public async ValueTask DisposeAsync()
//{
// foreach (var channel in _channels)
// {
// if (channel != null)
// await channel.CloseAsync();
// }
// if (_connection != null)
// await _connection.CloseAsync();
//}
private class QueueBinding
{
public string Exchange { get; set; } = "";
public string ExchangeType { get; set; } = "";
public string RoutingKey { get; set; } = "";
_logger.LogInformation("RabbitMQ Consumer started.");
}
}
}