（七）.NET6.0部署RabbitMQ

public class MqConnectionHelper:IMqConnectionHelper
{
    private readonly ILogger<MqConnectionHelper> _logger;

    public MqConnectionHelper(ILogger<MqConnectionHelper> logger)
    {
        _logger = logger;
        _connectionReceiveFactory = new IConnectionFactory[_costomerCount];
        _connectionReceive = new IConnection[_costomerCount];
        _modelReceive = new IModel[_costomerCount];
        _basicConsumer = new EventingBasicConsumer[_costomerCount];
    }



    //使用数组的部分，是给消费端用的，目前生产者只设置了一个，消费者可能存在多个。

    /// <summary>
    /// RabbitMQ工厂发送端
    /// </summary>
    private static IConnectionFactory _connectionSendFactory;
    /// <summary>
    /// RabbitMQ工厂接收端
    /// </summary>
    private static IConnectionFactory[] _connectionReceiveFactory;

    /// <summary>
    /// 连接 发送端
    /// </summary>
    private static IConnection _connectionSend;
    /// <summary>
    /// 连接 消费端
    /// </summary>
    private static IConnection[] _connectionReceive;

    /// <summary>
    /// 配置信息
    /// </summary>
    public static List<MqConfigInfo> _mqConfigs;

    /// <summary>
    /// 通道 发送端
    /// </summary>
    private static IModel _modelSend;
    /// <summary>
    /// 通道 消费端
    /// </summary>
    private static IModel[] _modelReceive;

    private static EventingBasicConsumer[] _basicConsumer;//事件

    //设置两个routingKey和队列名称，用来做测试使用
    public static int _costomerCount = 2;
    public static string[] _routingKey = new string[] { "YsqNet001", "YsqNet002" };
    public static string[] _queueName = new string[] { "Queue001", "Queue002" };

    /// <summary>
    /// 生产者初始化连接配置
    /// </summary>
    public void SendFactoryConnectionInit()
    {
        _connectionSendFactory = new ConnectionFactory
        {
            HostName = _mqConfigs.FirstOrDefault().Host,
            Port = _mqConfigs.FirstOrDefault().Port,
            UserName = _mqConfigs.FirstOrDefault().User,
            Password = _mqConfigs.FirstOrDefault().Password
        };
    }

    /// <summary>
    /// 生产者连接
    /// </summary>
    public void SendFactoryConnection()
    {
        if (_connectionSend != null && _connectionSend.IsOpen)
        {
            return;//已有连接
        }

        _connectionSend = _connectionSendFactory.CreateConnection();//创建生产者连接

        if (_modelSend != null && _modelSend.IsOpen)
        {
            return;//已有通道
        }

        _modelSend = _connectionSend.CreateModel();//创建生产者通道

        _modelSend.ExchangeDeclare(_mqConfigs.FirstOrDefault().ExchangeName, ExchangeType.Direct);
    }

    /// <summary>
    /// 消费者初始化连接配置
    /// </summary>
    public void ReceiveFactoryConnectionInit()
    {
        var factories = new ConnectionFactory
        {
            HostName = _mqConfigs.FirstOrDefault().Host,
            Port = _mqConfigs.FirstOrDefault().Port,
            UserName = _mqConfigs.FirstOrDefault().User,
            Password = _mqConfigs.FirstOrDefault().Password
        };

        for (int i = 0; i < _costomerCount; i++)
        {
            _connectionReceiveFactory[i] = factories;//给每个消费者绑定一个连接工厂
        }
    }

    /// <summary>
    /// 消费者连接
    /// </summary>
    /// <param name="consumeId"></param>
    /// <param name="exchangeName"></param>
    /// <param name="routeKey"></param>
    /// <param name="queueName"></param>
    public void ConncetionReceive(int consumeId, string exchangeName, string routeKey, string queueName)
    {
        _logger.LogInformation($"开始连接RabbitMQ消费者：{routeKey}");

        if (_connectionReceive[consumeId] != null && _connectionReceive[consumeId].IsOpen)
        {
            return;
        }

        _connectionReceive[consumeId] = _connectionReceiveFactory[consumeId].CreateConnection();//创建消费者连接

        _modelReceive[consumeId] = _connectionReceive[consumeId].CreateModel();//创建消费者通道

        _basicConsumer[consumeId] = new EventingBasicConsumer(_modelReceive[consumeId]);

        _modelReceive[consumeId].ExchangeDeclare(exchangeName, ExchangeType.Direct);//定义交换机名称和类型 与生产者保持一致

        _modelReceive[consumeId].QueueDeclare(
            queue:queueName,//消息队列名称
            durable:_mqConfigs.FirstOrDefault().Durable,//是否可持久化，此处配置在文件中，默认全局持久化（true）
            exclusive:false,
            autoDelete:false,
            arguments:null
            );//定义消费者队列

        _modelReceive[consumeId].QueueBind(queueName,exchangeName,routeKey);//队列绑定给指定的交换机

        _modelReceive[consumeId].BasicQos(0, 1, false);//设置消费者每次只接收一条消息

        StartListener((model, ea) =>
        {
            byte[] message = ea.Body.ToArray();//接收到的消息
            string msg = Encoding.UTF8.GetString(message);

            _logger.LogInformation($"队列{queueName}接收到消息：{msg}");
            Thread.Sleep(2000);
            _modelReceive[consumeId].BasicAck(ea.DeliveryTag, true);
        }, queueName, consumeId);
    }

    /// <summary>
    /// 消费者接收消息的确认机制
    /// </summary>
    /// <param name="basicDeliverEventArgs"></param>
    /// <param name="queueName"></param>
    /// <param name="consumeId"></param>
    private static void StartListener(EventHandler<BasicDeliverEventArgs> basicDeliverEventArgs,string queueName,int consumeId)
    {
        _basicConsumer[consumeId].Received += basicDeliverEventArgs;
        _modelReceive[consumeId].BasicConsume(queue: queueName, autoAck: false, consumer: _basicConsumer[consumeId]);//设置手动确认
    }

    /// <summary>
    /// 消息发布
    /// </summary>
    /// <param name="message"></param>
    /// <param name="exchangeName"></param>
    /// <param name="routeKey"></param>
    public static void PublishExchange(string message,string exchangeName,string routeKey="")
    {
        byte[] body = Encoding.UTF8.GetBytes(message);
        _modelSend.BasicPublish(exchangeName,routeKey,null,body);
    }
}