feat: add rabbitmq demo

anjoy8 · Nov 26, 2023 · 7a07b85 · 7a07b85
1 parent 62cdfb3
commit 7a07b85
Show file tree

Hide file tree

Showing 3 changed files with 86 additions and 8 deletions.
diff --git a/Blog.Core.Api/Blog.Core.xml b/Blog.Core.Api/Blog.Core.xml
diff --git a/Blog.Core.Api/Controllers/ValuesController.cs b/Blog.Core.Api/Controllers/ValuesController.cs
@@ -14,8 +14,11 @@
 using Microsoft.AspNetCore.Authorization;
 using Microsoft.AspNetCore.Mvc;
 using Microsoft.Extensions.Options;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
 using System.ComponentModel.DataAnnotations;
 using System.Linq.Expressions;
+using System.Text;
 
 namespace Blog.Core.Controllers
 {
@@ -39,6 +42,7 @@ public class ValuesController : BaseApiController
         private readonly IPasswordLibServices _passwordLibServices;
         readonly IBlogArticleServices _blogArticleServices;
         private readonly IHttpPollyHelper _httpPollyHelper;
+        private readonly IRabbitMQPersistentConnection _persistentConnection;
         private readonly SeqOptions _seqOptions;
 
         /// <summary>
@@ -52,6 +56,7 @@ public class ValuesController : BaseApiController
         /// <param name="user"></param>
         /// <param name="passwordLibServices"></param>
         /// <param name="httpPollyHelper"></param>
+        /// <param name="persistentConnection"></param>
         /// <param name="seqOptions"></param>
         public ValuesController(IBlogArticleServices blogArticleServices
             , IMapper mapper
@@ -60,6 +65,7 @@ public ValuesController(IBlogArticleServices blogArticleServices
             , IRoleModulePermissionServices roleModulePermissionServices
             , IUser user, IPasswordLibServices passwordLibServices
             , IHttpPollyHelper httpPollyHelper
+            , IRabbitMQPersistentConnection persistentConnection
             , IOptions<SeqOptions> seqOptions)
         {
             // 测试 Authorize 和 mapper
@@ -77,9 +83,69 @@ public ValuesController(IBlogArticleServices blogArticleServices
             _blogArticleServices = blogArticleServices;
             // httpPolly
             _httpPollyHelper = httpPollyHelper;
+            _persistentConnection = persistentConnection;
             _seqOptions = seqOptions.Value;
         }
 
+        /// <summary>
+        /// 测试Rabbit消息队列发送
+        /// </summary>
+        [HttpGet]
+        [AllowAnonymous]
+        public void TestRabbitMqPublish()
+        {
+            if (!_persistentConnection.IsConnected)
+            {
+                _persistentConnection.TryConnect();
+            }
+            using var channel = _persistentConnection.CreateModel();
+            var message = " < i am a sender! > ";
+            var body = Encoding.UTF8.GetBytes(message);
+            var properties = channel.CreateBasicProperties();
+            channel.BasicPublish(
+                exchange: "blogcore",
+                routingKey: "eventName",
+                mandatory: true,
+                basicProperties: properties,
+                body: body);
+        }
+
+        /// <summary>
+        /// 测试Rabbit消息队列订阅
+        /// </summary>
+        [HttpGet]
+        [AllowAnonymous]
+        public void TestRabbitMqSubscribe()
+        {
+            if (!_persistentConnection.IsConnected)
+            {
+                _persistentConnection.TryConnect();
+            }
+
+            string QueueName = "testq";
+            using var channel = _persistentConnection.CreateModel();
+            var consumer = new AsyncEventingBasicConsumer(channel);
+
+            consumer.Received += new AsyncEventHandler<BasicDeliverEventArgs>(
+                async (a, b) =>
+                {
+                    var Headers = b.BasicProperties.Headers;
+                    var msgBody = b.Body.ToArray();
+                    bool Dealresult = await Dealer(b.Exchange, b.RoutingKey, msgBody, Headers);
+                    if (Dealresult) channel.BasicAck(b.DeliveryTag, false);
+                    else channel.BasicNack(b.DeliveryTag, false, true);
+                }
+                );
+            channel.BasicConsume(QueueName, false, consumer);
+        }
+
+        private async Task<bool> Dealer(string exchange, string routingKey, byte[] msgBody, IDictionary<string, object> headers)
+        {
+            await Task.CompletedTask;
+            Console.WriteLine("我是消费者，这里消费了一条信息是：" + Encoding.UTF8.GetString(msgBody));
+            return true;
+        }
+
         [HttpGet]
         public MessageModel<List<ClaimDto>> MyClaims()
         {

diff --git a/Blog.Core.Api/appsettings.json b/Blog.Core.Api/appsettings.json
@@ -23,11 +23,12 @@
     "InstanceName": "" //前缀
   },
   "RabbitMQ": {
-    "Enabled": false,
-    "Connection": "118.25.251.13",
-    "UserName": "",
-    "Password": "!",
-    "RetryCount": 3
+    "Enabled": true,
+    "Connection": "101.35.125.157",
+    "UserName": "admin",
+    "Password": "admin",
+    "Port": "5672",
+    "RetryCount": 2
   },
   "Kafka": {
     "Enabled": false,
@@ -181,8 +182,8 @@
   "Audience": {
     "Secret": "sdfsdfsrty45634kkhllghtdgdfss345t678fs", //不要太短，16位+
     "SecretFile": "C:\\my-file\\blog.core.audience.secret.txt", //安全。内容就是Secret
-    "Issuer": "Blog.Core",
-    "Audience": "wr"
+    "Issuer": "Blog.Core", //这个值一定要在自己的项目里修改！！
+    "Audience": "wr" //这个值一定要在自己的项目里修改！！
   },
   "Mongo": {
     "ConnectionString": "mongodb://nosql.data",