using Microsoft.Extensions.Logging; using Polly; using Polly.Retry; using RabbitMQ.Client; using RabbitMQ.Client.Events; using RabbitMQ.Client.Exceptions; using System; using System.IO; using System.Net.Sockets; namespace Admin.Core.EventBus { /// /// RabbitMQ持久连接 /// public class RabbitMQPersistentConnection : IRabbitMQPersistentConnection { private readonly IConnectionFactory _connectionFactory; private readonly ILogger _logger; private readonly int _retryCount; IConnection _connection; bool _disposed; object sync_root = new object(); public RabbitMQPersistentConnection(IConnectionFactory connectionFactory, ILogger logger, int retryCount = 5) { _connectionFactory = connectionFactory ?? throw new ArgumentNullException(nameof(connectionFactory)); _logger = logger ?? throw new ArgumentNullException(nameof(logger)); _retryCount = retryCount; } /// /// 是否已连接 /// public bool IsConnected { get { return _connection != null && _connection.IsOpen && !_disposed; } } /// /// 创建Model /// /// public IModel CreateModel() { if (!IsConnected) { throw new InvalidOperationException("No RabbitMQ connections are available to perform this action"); } return _connection.CreateModel(); } /// /// 释放 /// public void Dispose() { if (_disposed) return; _disposed = true; try { _connection.Dispose(); } catch (IOException ex) { _logger.LogCritical(ex.ToString()); } } /// /// 连接 /// /// public bool TryConnect() { _logger.LogInformation("RabbitMQ Client is trying to connect"); lock (sync_root) { var policy = RetryPolicy.Handle() .Or() .WaitAndRetry(_retryCount, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)), (ex, time) => { _logger.LogWarning(ex, "RabbitMQ Client could not connect after {TimeOut}s ({ExceptionMessage})", $"{time.TotalSeconds:n1}", ex.Message); } ); policy.Execute(() => { _connection = _connectionFactory .CreateConnection(); }); if (IsConnected) { _connection.ConnectionShutdown += OnConnectionShutdown; _connection.CallbackException += OnCallbackException; _connection.ConnectionBlocked += OnConnectionBlocked; _logger.LogInformation("RabbitMQ Client acquired a persistent connection to '{HostName}' and is subscribed to failure events", _connection.Endpoint.HostName); return true; } else { _logger.LogCritical("FATAL ERROR: RabbitMQ connections could not be created and opened"); return false; } } } /// /// 连接被阻断 /// /// /// private void OnConnectionBlocked(object sender, ConnectionBlockedEventArgs e) { if (_disposed) return; _logger.LogWarning("A RabbitMQ connection is shutdown. Trying to re-connect..."); TryConnect(); } /// /// 连接出现异常 /// /// /// void OnCallbackException(object sender, CallbackExceptionEventArgs e) { if (_disposed) return; _logger.LogWarning("A RabbitMQ connection throw exception. Trying to re-connect..."); TryConnect(); } /// /// 连接被关闭 /// /// /// void OnConnectionShutdown(object sender, ShutdownEventArgs reason) { if (_disposed) return; _logger.LogWarning("A RabbitMQ connection is on shutdown. Trying to re-connect..."); TryConnect(); } } }