RabbitMQ学习ASP.NET Core中使用RabbitMQ

代码参考连接：https://www.bilibili.com/video/BV1GU4y1w7Yq?p=9&spm_id_from=pageDriver

1.项目结构

项目类型为ASP.NET Core Web API

 

 

   

   

2.Infrastructure类库代码 

nuget添加RabbitMQ.Client的引用

 (1) RabbitConnection.cs代码

using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Text;

namespace Infrastructure.Config
{
    public class RabbitConnection
    {
        private readonly RabbitOption _config;
        private IConnection _connection = null;
        public RabbitConnection(RabbitOption rabbitOption)
        {
            _config = rabbitOption;
        }

        public IConnection GetConnection()
        {
            if (_connection == null)
            {
                if (string.IsNullOrEmpty(_config.Address))
                {
                    ConnectionFactory factory = new ConnectionFactory();
                    factory.HostName = _config.HostName;
                    factory.Port = _config.Port;
                    factory.UserName = _config.UserName;
                    factory.Password = _config.Password;
                    factory.VirtualHost = _config.VirtualHost;
                    _connection = factory.CreateConnection();
                }
                else
                {
                    ConnectionFactory factory = new ConnectionFactory();
                    factory.UserName = _config.UserName;
                    factory.Password = _config.Password;
                    factory.VirtualHost = _config.VirtualHost;

                    var address = _config.Address;
                    List<AmqpTcpEndpoint> endpoints = new List<AmqpTcpEndpoint>();
                    foreach (var endpoint in address.Split(","))
                    {
                        endpoints.Add(new AmqpTcpEndpoint(endpoint.Split(":")[0],int.Parse(endpoint.Split(":")[1])));
                    }
                    _connection = factory.CreateConnection(endpoints);
                }
            }
            return _connection;
        }
    }
}

(2)RabbitConstant.cs代码

using System;
using System.Collections.Generic;
using System.Text;

namespace Infrastructure.Config
{
    public class RabbitConstant
    {
        public const string TEST_EXCHANGE = "test.exchange";
        public const string TEST_QUEUE = "test.queue";
    }
}

(3)RabbitOption.cs代码

using System;
using System.Collections.Generic;
using System.Text;

namespace Infrastructure.Config
{
    public class RabbitOption
    {
        /// <summary>
        /// 主机名称
        /// </summary>
        public string HostName { get; set; }
        public string Address { get; set; }
        /// <summary>
        /// 端口号
        /// </summary>
        public int Port { get; set; }
        public string UserName { get; set; }
        public string Password { get; set; }

        public string VirtualHost { get; set; }
    }
}

(4) QueueInfo.cs

using System;
using System.Collections.Generic;
using System.Text;

namespace Infrastructure.Consumer
{
    public class QueueInfo
    {
        /// <summary>
        /// 队列名称
        /// </summary>
        public string Queue { get; set; }
        /// <summary>
        /// 路由名称
        /// </summary>
        public string RoutingKey { get; set; }
        /// <summary>
        /// 交换机类型
        /// </summary>
        public string ExchangeType { get; set; }
        /// <summary>
        /// 交换机名称
        /// </summary>
        public string Exchange { get; set; }
        public IDictionary<string, object> props { get; set; } = null;
        public Action<RabbitMessageEntity> OnRecevied { get; set; }
    }
}

(5)RabbitChannelConfig.cs

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Text;

namespace Infrastructure.Consumer
{
    public class RabbitChannelConfig
    {
        public string ExchangeTypeName { get; set; }
        public string ExchangeName { get; set; }
        public string QueueName { get; set; }
        public string RoutingKeyName { get; set; }
        public IConnection Connection { get; set; }
        public EventingBasicConsumer Consumer { get; set; }

        /// <summary>
        /// 外部订阅消费者通知委托
        /// </summary>
        public Action<RabbitMessageEntity> OnReceivedCallback { get; set; }

        public RabbitChannelConfig(string exchangeType,string exchange,string queue,string routingKey)
        {
            this.ExchangeTypeName = exchangeType;
            this.ExchangeName = exchange;
            this.QueueName = queue;
            this.RoutingKeyName = routingKey;
        }

        public void Receive(object sender,BasicDeliverEventArgs args)
        {
            RabbitMessageEntity body = new RabbitMessageEntity();
            try
            {
                string content = Encoding.UTF8.GetString(args.Body.ToArray());
                body.Content = content;
                body.Consumer =(EventingBasicConsumer)sender;
                body.BasicDeliver = args;
            }
            catch (Exception e)
            {
                body.ErrorMessage = $"订阅出错{e.Message}";
                body.Exception = e;
                body.Error = true;
                body.Code = 500;
            }
            OnReceivedCallback?.Invoke(body);
        }
    }
}

(6)RabbitChannelManager.cs

using Infrastructure.Config;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Text;

namespace Infrastructure.Consumer
{
    public class RabbitChannelManager
    {
        public RabbitConnection Connection { get; set; }

        public RabbitChannelManager(RabbitConnection connection)
        {
            this.Connection = connection;
        }

        /// <summary>
        /// 创建接收消息的通道
        /// </summary>
        /// <param name="exchangeType"></param>
        /// <param name="exchange"></param>
        /// <param name="queue"></param>
        /// <param name="routingKey"></param>
        /// <param name="arguments"></param>
        /// <returns></returns>
        public RabbitChannelConfig CreateReceiveChannel(string exchangeType,string exchange,string queue,string routingKey,
            IDictionary<string,object>arguments = null)
        {
            IModel model = this.CreateModel(exchangeType,exchange,queue,routingKey,arguments);
            EventingBasicConsumer consumer = this.CreateConsumer(model,queue);
            RabbitChannelConfig channel = new RabbitChannelConfig(exchangeType,exchange,queue,routingKey);
            consumer.Received += channel.Receive;
            return channel;
        }

      
        /// <summary>
        /// 创建一个通道，包含交换机/队列/路由，并建立绑定关系
        /// </summary>
        /// <param name="exchangeType">交换机类型：Topic,Direct,Fanout</param>
        /// <param name="exchange">交换机名称</param>
        /// <param name="queue">队列名称</param>
        /// <param name="routingKey">路由名称</param>
        /// <param name="arguments"></param>
        /// <returns></returns>
        private IModel CreateModel(string exchangeType, string exchange, string queue, string routingKey, IDictionary<string, object> arguments)
        {
            exchangeType = string.IsNullOrEmpty(exchangeType) ? "default" : exchangeType;
            IModel model = this.Connection.GetConnection().CreateModel();
            model.BasicQos(0,1,false);
            model.QueueDeclare(queue,true,false,false,arguments);
            model.ExchangeDeclare(exchange, exchangeType);
            model.QueueBind(queue, exchange, routingKey);
            return model;
        }

        /// <summary>
        /// 创建消费者
        /// </summary>
        /// <param name="model"></param>
        /// <param name="queue"></param>
        /// <returns></returns>
        private EventingBasicConsumer CreateConsumer(IModel model, string queue)
        {
            EventingBasicConsumer consumer = new EventingBasicConsumer(model);
            model.BasicConsume(queue, false, consumer);
            return consumer;
        }
    }
}

(7)RabbitMessageEntity.cs

using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Text;

namespace Infrastructure.Consumer
{
    public class RabbitMessageEntity
    {
        public string Content { get; set; }
        public EventingBasicConsumer Consumer { get; set; }
        public BasicDeliverEventArgs BasicDeliver { get; set; }
        public string ErrorMessage { get; set; }
        public Exception Exception { get; set; }
        public bool Error { get; set; }
        public int Code { get; set; }
    }
}

(8)IRabbitProducer.cs

using System;
using System.Collections.Generic;

namespace Infrastructure.Producer
{
    public interface IRabbitProducer
    {
        public void Publish(string exchange,string routingKey,IDictionary<string,object> props,string content);
    }
}

(9)RabbitProducer.cs

using Infrastructure.Config;
using System;
using System.Collections.Generic;
using System.Text;

namespace Infrastructure.Producer
{
    public class RabbitProducer : IRabbitProducer
    {
        private readonly RabbitConnection _connection;
        public RabbitProducer(RabbitConnection connection)
        {
            _connection = connection;
        }
        public void Publish(string exchange, string routingKey, IDictionary<string, object> props, string content)
        {
            var channel = _connection.GetConnection().CreateModel();
            var prop = channel.CreateBasicProperties();
            if (props.Count > 0)
            {
                var delay = props["x-delay"];
                prop.Expiration = delay.ToString();
            }
            channel.BasicPublish(exchange, routingKey, false, prop, Encoding.UTF8.GetBytes(content));
        }
    }
}

3.RabbitMQ.Services类库代码

添加Infrastructure的项目引用

(1)IOrderService.cs

using System;
using System.Collections.Generic;
using System.Text;

namespace RabbitMQ.Services
{
    public interface IOrderService
    {
        void SendOrderMessage();
        void SendTestMessage(string message);
    }
}

(2) OrderService.cs

using Infrastructure.Config;
using Infrastructure.Producer;
using System;
using System.Collections.Generic;
using System.Text;

namespace RabbitMQ.Services
{
    public class OrderService : IOrderService
    {
        private readonly IRabbitProducer _rabbitProducer;
        public OrderService(IRabbitProducer rabbitProducer)
        {
            _rabbitProducer = rabbitProducer;
        }
        public void SendOrderMessage()
        {
            throw new NotImplementedException();
        }

        public void SendTestMessage(string message)
        {
            Console.WriteLine($"send message:{message}");
            _rabbitProducer.Publish(RabbitConstant.TEST_EXCHANGE,"",new Dictionary<string,object>(),message);
        }
    }
}

4. RabbitMQ.Test控制台代码

添加RabbitMQ.WebApi.Order项目引用

复制项目RabbitMQ.WebApi.Order下的appsettings.json到此项目中

(1)ProcessTest.cs

using Infrastructure.Config;
using Infrastructure.Consumer;
using Microsoft.Extensions.Hosting;
using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace RabbitMQ.Test
{
    public class ProcessTest : IHostedService
    {
        private readonly RabbitConnection _connection;
        public List<QueueInfo> Queues { get; } = new List<QueueInfo>();
        public ProcessTest(RabbitConnection connection)
        {
            _connection = connection;
            Queues.Add(new QueueInfo()
            { 
               ExchangeType = ExchangeType.Fanout,
               Exchange = RabbitConstant.TEST_EXCHANGE,
               Queue = RabbitConstant.TEST_QUEUE,
               RoutingKey = "",
               OnRecevied = this.Receive
            });
        }

        private void Receive(RabbitMessageEntity message)
        {
            Console.WriteLine($"Test Receive Message:{message.Content}");
            message.Consumer.Model.BasicAck(message.BasicDeliver.DeliveryTag,true);
        }

        public Task StartAsync(CancellationToken cancellationToken)
        {
            Console.WriteLine("RabbitMQ测试消息接收处理服务正在启动");
            RabbitChannelManager channelManager = new RabbitChannelManager(_connection);
            foreach (var queueInfo in Queues)
            {
                RabbitChannelConfig channel = channelManager.CreateReceiveChannel(queueInfo.ExchangeType,
                    queueInfo.Exchange,queueInfo.Queue,queueInfo.RoutingKey);
                channel.OnReceivedCallback = queueInfo.OnRecevied;
            }
            Console.WriteLine("RabbitMQ测试消息接收处理服务已经启动");
            return Task.CompletedTask;
        }

        public Task StopAsync(CancellationToken cancellationToken)
        {
            return Task.CompletedTask;
        }
    }
}

(2) Program.cs

using Infrastructure.Config;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using System;

namespace RabbitMQ.Test
{
    class Program
    {
        static void Main(string[] args)
        {
            var section = new ConfigurationBuilder()
                .AddJsonFile("appsettings.json")
                .Build()
                .GetSection("RabbitMQ");
            var host = new HostBuilder()
                .ConfigureServices(services =>
                 services.AddSingleton(new RabbitConnection(section.Get<RabbitOption>()))
                         .AddSingleton<IHostedService,ProcessTest>())
                .Build();
             host.Run();
        }
    }
}

5. RabbitMQ.WebApi.Order代码

nuget添加Swashbuckle.AspNetCore引用

添加项目Infrastructure、RabbitMQ.Services引用

(1) OrderController.cs

using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc;
using RabbitMQ.Services;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;

namespace RabbitMQ.WebApi.Order.Controllers
{
    [Route("api/[controller]")]
    [ApiController]
    public class OrderController : ControllerBase
    {
        private readonly IOrderService _orderService;

        public OrderController(IOrderService orderService)
        {
            _orderService = orderService;
        }

        //[HttpGet]
        //public IActionResult Order()
        //{
        //    _orderService.SendOrderMessage();
        //    return Ok();
        //}
        [HttpGet("test")]
        public IActionResult Test(string message)
        {
            _orderService.SendTestMessage(message);
            return Ok();
        }

    }
}

(2) ServiceExtensions.cs

using Infrastructure.Config;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;

namespace RabbitMQ.WebApi.Order.Extensions
{
    public static class ServiceExtensions
    {
        public static void ConfigureCors(this IServiceCollection services)
        {
            services.AddCors(options=>
            {
                options.AddPolicy("AnyPolicy",
                    builder=>builder.AllowAnyOrigin()
                                    .AllowAnyMethod()
                                    .AllowAnyHeader());
            });
        }
        public static void ConfigureRabbitContext(this IServiceCollection services,IConfiguration config)
        {
            var section = config.GetSection("RabbitMQ");
            services.AddSingleton(
                  new RabbitConnection(section.Get<RabbitOption>())); 
        }
    }
}

(3) Startup.cs

using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using RabbitMQ.WebApi.Order.Extensions;
using Microsoft.OpenApi.Models;
using RabbitMQ.Services;
using Infrastructure.Producer;

namespace RabbitMQ.WebApi.Order
{
    public class Startup
    {
        public Startup(IConfiguration configuration)
        {
            Configuration = configuration;
        }

        public IConfiguration Configuration { get; }

        // This method gets called by the runtime. Use this method to add services to the container.
        public void ConfigureServices(IServiceCollection services)
        {
            services.AddControllers();
            services.AddSwaggerGen(c=>
            {
                c.SwaggerDoc("v1",new OpenApiInfo { Title="RabbitMQ.WebApi.Order",Version="v1"});
            });
            services.AddScoped<IOrderService, OrderService>();
            services.AddScoped<IRabbitProducer,RabbitProducer>();
            services.ConfigureCors();
            services.ConfigureRabbitContext(Configuration);
        }

        // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
        public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
        {
            if (env.IsDevelopment())
            {
                app.UseDeveloperExceptionPage();
                app.UseSwagger();
                app.UseSwaggerUI(c=>c.SwaggerEndpoint("/swagger/v1/swagger.json","RabbitMQ.WebApi.Order v1"));
            }

            app.UseRouting();

            app.UseAuthorization();

            app.UseEndpoints(endpoints =>
            {
                endpoints.MapControllers();
            });
        }
    }
}

(4) appsettings.json

{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft": "Warning",
      "Microsoft.Hosting.Lifetime": "Information"
    }
  },
  "RabbitMQ": {
    "HostName": "127.0.0.1",
    "Address": "",
    "Port": 5672,
    "UserName": "guest",
    "Password": "guest",
    "VirtualHost": "/"
  },

  "AllowedHosts": "*"
}

6.运行

(1)  powershell管理员模式运行RabbitMQ.Test项目

PS D:\DotNetProject\RabbitMQ.WebApiDemo\RabbitMQ.Test> dotnet run

 (2)  powershell管理员模式运行RabbitMQ.WebApi.Order项目 

PS D:\DotNetProject\RabbitMQ.WebApiDemo\RabbitMQ.WebApi.Order> dotnet run

 (3) 浏览器运行http://localhost:5000/swagger/index.html

 (4) 测试接口Test 

 (9)运行结果

 如需要项目源码，请留下邮箱，会将项目整个发送

IRabbitProducer