集成MQ

Common/RabbitMQClient.cs

@@ -0,0 +1,75 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
+using System.Threading;
+using Infrastructure.Model;
+using Base;
+using Common;
+using Util;
+
+namespace MySystem
+{
+    public class RabbitMQClient
+    {
+        public readonly static RabbitMQClient Instance = new RabbitMQClient();
+        string UserName,Password,HostName,VirtualHostName;
+        private RabbitMQClient()
+        {
+            MqSettings mqSettings = new();
+            AppSettings.Bind("MqSettings", mqSettings);
+            UserName = mqSettings.UserName;
+            Password = mqSettings.Password;
+            HostName = mqSettings.HostName;
+            VirtualHostName = mqSettings.VirtualHostName;
+        }
+
+        #region 单对单接收
+        public static IConnection _connection;
+        public void CreateConn()
+        { 
+            var factory = new ConnectionFactory()
+            {
+                UserName = UserName,
+                Password = Password,
+                AutomaticRecoveryEnabled = true,  //如果connection挂掉是否重新连接
+                TopologyRecoveryEnabled = true,  //连接恢复后，连接的交换机，队列等是否一同恢复
+                RequestedHeartbeat = TimeSpan.FromMinutes(1),
+                VirtualHost = VirtualHostName,
+            };
+            List<AmqpTcpEndpoint> p = new List<AmqpTcpEndpoint>();
+            string[] HostNames = HostName.Split(',');
+            foreach (string subHostName in HostNames)
+            {
+                string[] subHostNameData = subHostName.Split(':');
+                p.Add(new AmqpTcpEndpoint(subHostNameData[0], int.Parse(subHostNameData[1])));
+            }
+            _connection = factory.CreateConnection(p);
+        }
+        public void StartReceive(string QueueName, string Exchange = "", string RoutingKey = "")
+        {
+            if (_connection == null)
+            {
+                CreateConn();
+            }
+            else if (!_connection.IsOpen)
+            {
+                CreateConn();
+            }
+            var channel = _connection.CreateModel();
+            channel.QueueBind(QueueName, Exchange, RoutingKey);
+            EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
+            consumer.Received += (a, e) =>
+            {
+                string MsgContent = Encoding.Default.GetString(e.Body.ToArray());
+                Function.WriteLog(MsgContent, "接收mq数据队列");
+                PrizeDo.sendPrize(MsgContent);
+                channel.BasicAck(e.DeliveryTag, true); //收到回复后，RabbitMQ会直接在队列中删除这条消息
+            };
+            channel.BasicConsume(QueueName, false, consumer);
+        }
+        #endregion
+
+    }
+}

Model/Base/OptionsSetting.cs

@@ -94,6 +94,29 @@ namespace Infrastructure.Model
         public string TokenType { get; set; } = "Bearer";
     }
 
+    /// <summary>
+    /// Mq
+    /// </summary>
+    public class MqSettings
+    {
+        /// <summary>
+        /// 账号
+        /// </summary>
+        public string UserName { get; set; }
+        /// <summary>
+        /// 密码
+        /// </summary>
+        public string Password { get; set; }
+        /// <summary>
+        /// 主机
+        /// </summary>
+        public string HostName { get; set; }
+        /// <summary>
+        /// 虚拟主机
+        /// </summary>
+        public string VirtualHostName { get; set; }
+    }
+
     public class CodeGen
     {
         public bool ShowApp { get; set; }

PrizeSet.csproj

@@ -34,6 +34,7 @@
     <PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
     <PackageReference Include="NLog" Version="5.2.8" />
     <PackageReference Include="Npgsql" Version="8.0.3" />
+    <PackageReference Include="RabbitMQ.Client" Version="6.8.1" />
     <PackageReference Include="SkiaSharp.NativeAssets.Linux.NoDependencies" Version="2.88.8" />
     <PackageReference Include="SkiaSharp.QrCode" Version="0.7.0" />
     <PackageReference Include="SqlSugar.IOC" Version="2.0.0" />

Util/PrizeDo.cs

@@ -13,10 +13,11 @@ namespace Util
     {
         public static string batchNo = "";
         //发奖入口
-        public static void sendPrize(int prizeObjectTableId, string content)
+        public static void sendPrize(string content)
         {
             batchNo = DateTime.Now.ToString("yyyyMMddHHmmssfff") + Function.get_Random(3);
-            string priObjString = prizeObject(prizeObjectTableId, content);
+            JsonData jsonObj = JsonMapper.ToObject(content);
+            string priObjString = prizeObject(jsonObj["prize_tag"].ToString(), jsonObj["content"].ToJson());
             if(!string.IsNullOrEmpty(priObjString))
             {
                 string[] priObjData = priObjString.Split('|');
@@ -326,13 +327,13 @@ namespace Util
         }
 
         //奖励发放对象（奖励发放入口）
-        public static string prizeObject(int prizeObjectTableId, string content)
+        public static string prizeObject(string prizeObjectTag, string content)
         {
             Dictionary<string, object> dic = new Dictionary<string, object>();
             var prizeObjectTableService = App.GetService<IPriPrizeObjectTableService>();
             var prizeObjectFieldService = App.GetService<IPriPrizeObjectFieldService>();
             //查询表
-            var prizeObjectTable = prizeObjectTableService.GetFirst(m => m.id == prizeObjectTableId) ?? new PriPrizeObjectTable();
+            var prizeObjectTable = prizeObjectTableService.GetFirst(m => m.prizeListIds == prizeObjectTag) ?? new PriPrizeObjectTable();
             //查询返回字段
             string field = prizeObjectTable.selectField;
             int projectId = prizeObjectTable.projectId;

appsettings.Development.json

@@ -40,6 +40,12 @@
     "RefreshTokenTime": 30, //分钟
     "TokenType": "Bearer"
   },
+  "MqSettings": {
+    "UserName": "admin",
+    "Password": "admin",
+    "HostName": "47.109.61.191:5672",
+    "VirtualHostName": "/"
+  },
   "InitDb": false, //是否初始化db
   "ApiKey": "*ga34|^7", //webapi解密的key
   "Host": "http://telap.yumuit.com"