十五、.net core（.NET 6）搭建RabbitMQ消息队列生产者和消费者的简单方法

public class MqConfigInfo
{
public string Host { get; set; }
public int Port { get; set; }
public string User { get; set; }
public string Password { get; set; }
public string ExchangeName { get; set; }
public bool Durable { get; set; }
}

public class ReadMqConfigHelper:IReadMqConfigHelper
{
private readonly ILogger<ReadMqConfigHelper> _logger;
public ReadMqConfigHelper(ILogger<ReadMqConfigHelper> logger)
{
_logger = logger;
}
public List<MqConfigInfo> ReadMqConfig()
{
try
{
List<MqConfigInfo> config = AppHelper.ReadAppSettings<MqConfigInfo>(new string[] { "MQ" }); // 读取MQ配置信息
if (config.Any())
{
return config;
}
_logger.LogError($"获取MQ配置信息失败：没有可用数据集");
return null;
}
catch (Exception ex)
{
_logger.LogError($"获取MQ配置信息失败：{ex.Message}");
return null;
}
}
}

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]; } /*
备注：使用数组的部分，是给消费端用的。目前生产者只设置了一个，消费者可能存在多个。
当然，有条件的还可以上RabbitMQ集群进行处理，会更好玩一点。
*/
private static IConnectionFactory _connectionSendFactory; //RabbitMQ工厂 发送端
private static IConnectionFactory[] _connectionReceiveFactory; //RabbitMQ工厂 接收端 private static IConnection _connectionSend; //连接 发送端
private static IConnection[] _connectionReceive; //连接 消费端 public static List<MqConfigInfo> _mqConfig; // 配置信息 private static IModel _modelSend; //通道 发送端
private static IModel[] _modelReceive; //通道 消费端 private static EventingBasicConsumer[] _basicConsumer; // 事件 /* 设置两个routingKey 和 队列名称，用来做测试使用*/
public static int _costomerCount = 2;
public static string[] _routingKey = new string[] {"WeskyNet001","WeskyNet002" };
public static string[] _queueName = new string[] { "Queue001", "Queue002" }; /// <summary>
/// 生产者初始化连接配置
/// </summary>
public void SendFactoryConnectionInit()
{
_connectionSendFactory = new ConnectionFactory
{
HostName = _mqConfig.FirstOrDefault().Host,
Port = _mqConfig.FirstOrDefault().Port,
UserName = _mqConfig.FirstOrDefault().User,
Password = _mqConfig.FirstOrDefault().Password
};
} /// <summary>
/// 生产者连接
/// </summary>
public void SendFactoryConnection()
{ if (null != _connectionSend && _connectionSend.IsOpen)
{
return; // 已有连接
}
_connectionSend = _connectionSendFactory.CreateConnection(); // 创建生产者连接 if (null != _modelSend && _modelSend.IsOpen)
{
return; // 已有通道
}
_modelSend = _connectionSend.CreateModel(); // 创建生产者通道 _modelSend.ExchangeDeclare(_mqConfig.FirstOrDefault().ExchangeName, ExchangeType.Direct); // 定义交换机名称和类型(direct) } /// <summary>
/// 消费者初始化连接配置
/// </summary>
public void ReceiveFactoryConnectionInit()
{
var factories = new ConnectionFactory
{
HostName = _mqConfig.FirstOrDefault().Host,
Port = _mqConfig.FirstOrDefault().Port,
UserName = _mqConfig.FirstOrDefault().User,
Password = _mqConfig.FirstOrDefault().Password
}; for (int i = 0; i < _costomerCount; i++)
{
_connectionReceiveFactory[i] = factories; // 给每个消费者绑定一个连接工厂
}
} /// <summary>
/// 消费者连接
/// </summary>
/// <param name="consumeIndex"></param>
/// <param name="exchangeName"></param>
/// <param name="routeKey"></param>
/// <param name="queueName"></param>
public void ConnectionReceive(int consumeIndex, string exchangeName, string routeKey, string queueName)
{
_logger.LogInformation($"开始连接RabbitMQ消费者：{routeKey}"); if (null != _connectionReceive[consumeIndex] && _connectionReceive[consumeIndex].IsOpen)
{
return;
}
_connectionReceive[consumeIndex] = _connectionReceiveFactory[consumeIndex].CreateConnection(); // 创建消费者连接 if (null != _modelReceive[consumeIndex] && _modelReceive[consumeIndex].IsOpen)
{
return;
}
_modelReceive[consumeIndex] = _connectionReceive[consumeIndex].CreateModel(); // 创建消费者通道 _basicConsumer[consumeIndex] = new EventingBasicConsumer(_modelReceive[consumeIndex]); _modelReceive[consumeIndex].ExchangeDeclare(exchangeName, ExchangeType.Direct); // 定义交换机名称和类型 与生产者保持一致 _modelReceive[consumeIndex].QueueDeclare(
queue: queueName, //消息队列名称
durable: _mqConfig.FirstOrDefault().Durable, // 是否可持久化，此处配置在文件中，默认全局持久化(true)，也可以自定义更改
exclusive: false,
autoDelete: false,
arguments: null
); // 定义消费者队列 _modelReceive[consumeIndex].QueueBind(queueName, exchangeName, routeKey); // 队列绑定给指定的交换机 _modelReceive[consumeIndex].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[consumeIndex].BasicAck(ea.DeliveryTag, true);
}, queueName, consumeIndex); } /// <summary>
/// 消费者接收消息的确认机制
/// </summary>
/// <param name="basicDeliverEventArgs"></param>
/// <param name="queueName"></param>
/// <param name="consumeIndex"></param>
private static void StartListener(EventHandler<BasicDeliverEventArgs> basicDeliverEventArgs, string queueName, int consumeIndex)
{
_basicConsumer[consumeIndex].Received += basicDeliverEventArgs;
_modelReceive[consumeIndex].BasicConsume(queue: queueName, autoAck: false, consumer: _basicConsumer[consumeIndex]); // 设置手动确认。 } /// <summary>
/// 消息发布
/// </summary>
/// <param name="message"></param>
/// <param name="exchangeName"></param>
/// <param name="routingKey"></param>
public static void PublishExchange(string message, string exchangeName, string routingKey = "")
{
byte[] body = Encoding.UTF8.GetBytes(message);
_modelSend.BasicPublish(exchangeName, routingKey, null, body);
} }