src/Library/NetPro.RabbitMQ/QueueService.cs

这里只显示每个文件的开头 60 行。登录后可以解锁完整代码。

微信扫码关注公众号，代码整理好后第一时间通知你。

using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using MQMiddleware.Configuration;
using Newtonsoft.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Text;
using System.Threading.Tasks;

namespace MQMiddleware
{
    /// <summary>
    /// Implementation of the custom RabbitMQ queue service.
    /// </summary>
    internal class QueueService : IQueueService, IDisposable
    {
        /// <summary>
        /// RabbitMQ connection.
        /// </summary>
        public IConnection Connection => _connection;

        /// <summary>
        /// RabbitMQ channel.
        /// </summary>
        public IModel Channel => _channel;

        EventHandler<BasicDeliverEventArgs> _receivedMessage;
        bool _consumingStarted;

        readonly IDictionary<string, IList<IMessageHandler>> _messageHandlers;
        readonly IDictionary<string, IList<IAsyncMessageHandler>> _asyncMessageHandlers;
        readonly IDictionary<string, IList<INonCyclicMessageHandler>> _nonCyclicHandlers;
        readonly IDictionary<string, IList<IAsyncNonCyclicMessageHandler>> _asyncNonCyclicHandlers;
        readonly IDictionary<Type, List<string>> _routingKeys;
        readonly IEnumerable<RabbitMqExchange> _exchanges;
        readonly ILogger<QueueService> _logger;
        readonly IConnection _connection;
        readonly IModel _channel;
        readonly EventingBasicConsumer _consumer;
        readonly object _lock = new object();

        const int ResendTimeout = 60;
        const int QueueExpirationTime = 60000;

        public QueueService(
            IEnumerable<IMessageHandler> messageHandlers,
            IEnumerable<IAsyncMessageHandler> asyncMessageHandlers,
            IEnumerable<INonCyclicMessageHandler> nonCyclicHandlers,
            IEnumerable<IAsyncNonCyclicMessageHandler> asyncnonCyclicHandlers,
            IEnumerable<RabbitMqExchange> exchanges,
            IEnumerable<MessageHandlerRouter> routers,
            ILoggerFactory loggerFactory,
            IOptions<RabbitMqClientOptions> options)
        {
            if (options is null)
            {