BuYuHouDuan/Infrastructure/RabbitMQHelper/MQHostedService.cs

using EasyNetQ;
using EasyNetQ.Topology;
using RabbitMQHelper.Attributes;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using RabbitMQHelper.Models;
using MongoDB.Driver;

namespace RabbitMQHelper
{
    public class MQHostedService : BackgroundService
    {
        private readonly IServiceProvider _serviceProvider;
        private readonly IBus _bus;
        private readonly IRabbitConfig _mongoDBConfig;
        private readonly IMongoCollection<ReceiveRecord> _collection;
        private static readonly object lockHelper = new object();
        public MQHostedService(IServiceProvider serviceProvider, IBus bus, IRabbitConfig mongoDBConfig)
        {
            _serviceProvider = serviceProvider;
            _bus = bus;
            _mongoDBConfig = mongoDBConfig;
            var database = new MongoClient(_mongoDBConfig.MongoConnectionString).GetDatabase(_mongoDBConfig.MongoDatabaseName);
            _collection = database.GetCollection<ReceiveRecord>("mq_receive_record");
        }
        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            var executorDescriptorList = new List<ConsumerExecutorDescriptor>();
            foreach (Type ty in _mongoDBConfig.ListSubscribe)
            {
                executorDescriptorList.AddRange(GetTopicAttributesDescription(ty.GetTypeInfo()));
            }


            var exchange = await _bus.Advanced.ExchangeDeclareAsync(_mongoDBConfig.DefaultExchange, ExchangeType.Topic);
                        
            foreach (var item in executorDescriptorList)
            {
                using (var scope = _serviceProvider.GetRequiredService<IServiceScopeFactory>().CreateScope())
                {
                    var topic = item.Attribute.Topic;
                    var provider = scope.ServiceProvider;
                    var obj = ActivatorUtilities.GetServiceOrCreateInstance(provider, item.ImplTypeInfo.GetTypeInfo());

                    var subscriptId = $"{ item.ClassName.ToLower() }.{item.MethodInfo.Name.ToLower()}";

                    var queue = await _bus.Advanced.QueueDeclareAsync(subscriptId);

                    await _bus.Advanced.BindAsync(exchange, queue, topic, null);

                    _bus.Advanced.Consume<RabbitMessage>(queue, (msg, info) => Task.Factory.StartNew(async () =>
                    {
                        await ReceiveMessage(msg.Body, info.Queue, info.RoutingKey);
                        lock (lockHelper)
                        {
                            object[] parameters = new object[] { msg.Body.Data, msg.Body.Headers };
                            item.MethodInfo.Invoke(obj, parameters);
                        }
                    }));

                    //await _bus.PubSub.SubscribeAsync<RabbitMessage>(subscriptId, async msg =>
                    //{
                    //    await ReceiveMessage(msg, subscriptId, topic);
                    //    object[] parameters = new object[] { msg.Data, msg.Headers };
                    //    item.MethodInfo.Invoke(obj, parameters);
                        
                    //}, x => x.WithTopic(topic).w);
                }
            }
            await Task.Delay(Timeout.Infinite, stoppingToken);
        }
        public async Task ReceiveMessage(RabbitMessage msg,string subscriptId,string topic)
        {
            try {
                //var database = new MongoClient(_mongoDBConfig.MongoConnectionString).GetDatabase(_mongoDBConfig.MongoDatabaseName);
                //var collection = database.GetCollection<ReceiveRecord>("mq_receive_record");
                await _collection.InsertOneAsync(new ReceiveRecord(msg.RecordId, subscriptId, topic));
            }
            catch (Exception ex)
            {
                
            }
        }
        private IEnumerable<ConsumerExecutorDescriptor> GetTopicAttributesDescription(TypeInfo typeInfo)
        {
            foreach (var method in typeInfo.DeclaredMethods)
            {
                var topicAttr = method.GetCustomAttributes<RabbitMQAttribute>(true);
                var topicAttributes = topicAttr as IList<RabbitMQAttribute> ?? topicAttr.ToList();

                if (!topicAttributes.Any())
                {
                    continue;
                }

                foreach (var attr in topicAttributes)
                {
                    yield return InitDescriptor(attr, method, typeInfo, typeInfo.Name);
                }
            }
        }
        private ConsumerExecutorDescriptor InitDescriptor(RabbitMQAttribute attr, MethodInfo methodInfo, TypeInfo implType, string name)
        {
            var descriptor = new ConsumerExecutorDescriptor
            {
                Attribute = attr,
                MethodInfo = methodInfo,
                ImplTypeInfo = implType,
                ClassName = name
            };
            return descriptor;
        }
    }

}