qlhb/Ropin.Environmentally.AlarmService/Service/RabbitMQReceiveService.cs

using log4net;
using Microsoft.Extensions.Hosting;
using RabbitMQ.Client;
using System;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Ropin.Environmentally.AlarmService.Model;
using RabbitMQ.Client.Events;
using System.Net.Http;
using System.Net;
using Ropin.Inspection.Model;
using Newtonsoft.Json;
using Ropin.Inspection.Service;

namespace Ropin.Environmentally.AlarmService.Service
{
    public class RabbitMQReceiveService : IHostedService, IDisposable
    {
        private readonly IServiceProvider _provider;
        private static readonly ILog log = LogManager.GetLogger(typeof(RabbitMQReceiveService));
        private readonly IHttpClientFactory httpClientFactory;
        private readonly CofingSetModel _cofingSetModel;
        private readonly RabbitMQModel _rabbitMQModel;
        private readonly IPushMsgService _pushMsgService;
        public RabbitMQReceiveService(IServiceProvider provider, CofingSetModel cofingSetModel, IHttpClientFactory httpClientFactory, IPushMsgService pushMsgService)//, IPushMsgService pushMsgService
        {
            this._provider = provider;
            _cofingSetModel = cofingSetModel;
            if (cofingSetModel != null)
            {
                _rabbitMQModel = cofingSetModel.RabbitMQ;
            }

            this.httpClientFactory = httpClientFactory;
            _pushMsgService = pushMsgService;
        }
        private IConnection con;
        private IModel channel;

        public Task StartAsync(CancellationToken cancellationToken)
        {
            Task.Run(async () =>
            {
                //TpushMsgModel msgModel = new TpushMsgModel();
                //msgModel.C_DevStoreCode = "8453d5ed-8a21-4880-88e7-f872e93551bf";
                //msgModel.C_MsgTypeCode = "strs";
                //msgModel.Subject = "str";
                //string msgData = JsonConvert.SerializeObject(msgModel);
                //await WebScadaAlarmExecute(msgData);
                await AddRabbitMQ();

                //string msg = "{\"C_DevStoreCode\":\"015ea688-0698-408c-94b7-0475b14a20fc\",\"C_MsgTypeCode\":\"MSG_TYPE_011\",\"Subject\":\"设备开启关闭报警\",\"Msg\":\"关闭\",\"DevNumber\":\"20230413\",\"DevName\":\"沸石转轮催化燃烧设备\",\"UserName\":\"设备\",\"UserMobile\":null,\"DevAddress\":null,\"DevOpsName\":null,\"CreateOn\":\"03/06/2025 16:38:43\",\"GenerationType\":1,\"msgStatus\":4}";
                //bool result= await WebScadaAlarmExecute(msg);

                //while (true)
                //{
                //    await Task.Delay(5000);//10000:10s
                //}
            });
            return Task.CompletedTask;
        }

        public async Task AddRabbitMQ()
        {
            try
            {
                var factory = new ConnectionFactory()
                {
                    HostName = _rabbitMQModel.HostName,//"60.204.212.71",//IP地址
                    Port = _rabbitMQModel.Port,//5672,//端口号
                    UserName = _rabbitMQModel.UserName,//"guest",//用户账号
                    VirtualHost = _rabbitMQModel.VirtualHost,//"/",                    
                    Password = _rabbitMQModel.Password,// "guest"//用户密码
                };
                if (con == null || con.IsOpen == false)
                {
                    con = factory.CreateConnection();//创建连接对象
                }
                if (channel == null || channel.IsOpen == false)
                {
                    channel = con.CreateModel();//创建连接会话对象
                }
                channel.ExchangeDeclare(_rabbitMQModel.ExchangeName, type: ExchangeType.Direct); // Direct 交换机示例
                //声明队列
                var queueName = channel.QueueDeclare(
                   queue: _rabbitMQModel.QueueName, //消息队列名称
                    durable: false, //是否缓存
                    exclusive: false,
                    autoDelete: false,
                    arguments: null
                    ).QueueName;
                channel.QueueBind(queueName, _rabbitMQModel.ExchangeName, _rabbitMQModel.RoutingKey);
                //告诉Rabbit每次只能向消费者发送一条信息,再消费者未确认之前,不再向他发送信息
                channel.BasicQos(0, 1, false);
                channel.ConfirmSelect(); // 开启消息确认模式
                //创建消费者对象
                var consumer = new EventingBasicConsumer(channel);
                consumer.Received += async (model, ea) =>
                {
                    try
                    {
                        var body = ea.Body.ToArray();
                        var message = Encoding.UTF8.GetString(body);
                        log.Info("【RabbitMQ】" + message);
                        bool result = await WebScadaAlarmExecute(message);
                        if (result)
                        {
                            // 确认消息已处理
                            channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                        }
                        else
                        {
                            channel.BasicNack(ea.DeliveryTag, false, true); // 重新入队或发送到死信队列的逻辑需要自定义实现
                        }
                    }
                    catch (Exception ex)
                    {
                        channel.BasicNack(ea.DeliveryTag, false, true); // 重新入队或发送到死信队列的逻辑需要自定义实现
                        log.Info("【RabbitMQ-接收消息处理异常了】" + ex.Message);
                    }
                };
                //消费者开启监听  将autoAck设置false 手动确认关闭；true：自动关闭；
                channel.BasicConsume(queue: _rabbitMQModel.QueueName, autoAck: false, consumer: consumer);
            }
            catch (Exception ex)
            {
                log.Info("【异常-RabbitMQ】" + ex.Message);
            }
        }
        /// <summary>
        /// WebScada服务的设备报警
        /// </summary>
        /// <param name="msgData"></param>
        /// <returns></returns>
        public async Task<bool> WebScadaAlarmExecute(string msgData)
        {
            bool result = false;
            try
            {
                var pushMsg = JsonConvert.DeserializeObject<TpushMsgModel>(msgData);
                Inspection.Common.Helper.RabbitMQModel rabbitMQModels =new Inspection.Common.Helper.RabbitMQModel();
                rabbitMQModels.QueueName = "rab.video.record.mqtt"; //消息队列名称
                rabbitMQModels.UserName = _rabbitMQModel.UserName;
                rabbitMQModels.Password = _rabbitMQModel.Password;
                rabbitMQModels.HostName = _rabbitMQModel.HostName;
                rabbitMQModels.Port = _rabbitMQModel.Port;
                rabbitMQModels.VirtualHost = _rabbitMQModel.VirtualHost;
                rabbitMQModels.ExchangeName = rabbitMQModels.QueueName+".DirectExchange";
                rabbitMQModels.RoutingKey = rabbitMQModels.QueueName + ".key";
                bool bols = await _pushMsgService.PushAlarmMsgAsync(pushMsg, pushMsg.Subject, rabbitMQModels);
                if (bols)
                {
                    result = true;
                }
            }
            catch (Exception ex)
            {
                log.Info("【异常-WebScadaAlarmExecute】" + ex.Message);
            }
            return result;
        }

        /// <summary>
        /// WebScada服务的设备报警
        /// </summary>
        /// <param name="msgData"></param>
        /// <returns></returns>
        public async Task<bool> WebScadaAlarmExecuteAPI(string msgData)
        {
            bool result = false;
            try
            {
                using (HttpClient httpClient = httpClientFactory.CreateClient())
                {
                    var httpRequestMessage = new HttpRequestMessage
                    {
                        Method = HttpMethod.Post,
                        RequestUri = new Uri(_cofingSetModel.PublicPushMessageAPI),
                        Content = new StringContent(msgData, Encoding.UTF8, "application/json")
                    };
                    var response = await httpClient.SendAsync(httpRequestMessage);
                    string responseResult = await response.Content.ReadAsStringAsync();
                    if (response.StatusCode != HttpStatusCode.OK)
                    {
                        log.Info($"【错误-WebScadaAlarmExecute】WebScada服务的设备报警 发送数据失败【{msgData}】");
                    }
                    else
                    {
                        //"{\"code\":-1001,\"message\":\"参数错误。\"}"
                        receiveModel resultReturn = JsonConvert.DeserializeObject<receiveModel>(responseResult);
                        if (resultReturn.code == "0") { result = true; }
                    }
                }
            }
            catch (Exception ex)
            {
                log.Info("【异常-WebScadaAlarmExecute】" + ex.Message);
            }
            return result;
        }
        public Task StopAsync(CancellationToken cancellationToken)
        {
            Dispose();
            return Task.CompletedTask;
        }

        public void Dispose()
        {
            if (channel!=null)
            {
                channel.Close();
            }
            if (con!=null)
            {
                con.Close();
            }
        }
    }
    public class receiveModel
    {
        public string code { get; set; } 
        public string message { get; set; }
    }
}