RabbitMQ 客户端毗连、发送、吸收处置惩罚消息 - IT评测·应用市场-qidao123.com

private ConnectionFactory factory;
private IChannel channel;
private IConnection connection;
private NetWorkState state;
private Action<string, byte[]> receivedCallBack;
private const int TimeOut = 10; //连接超时 10 秒
private bool dispose = false;
public RabbitMQConnect(RabbitMQConnectData connectData)
{
this.connectData = connectData;
State = NetWorkState.Disconnected;
dispose = false;
}
public string Queue
{
get { return connectData.queue; }
}
public NetWorkState State
{
get { return state; }
private set { state = value; }
}
public IChannel Channel
{
get { return channel; }
}
/// <summary>
/// 网络是否连接中
/// </summary>
public bool IsConnect
{
get
{
if (null == channel || null == connection)
{
return false;
}
return channel.IsOpen && connection.IsOpen;
}
}
public async Task StartConnect()
{
if (State == NetWorkState.Connecting)
{
await Task.Delay(TimeOut * 1000);
}
if (State == NetWorkState.Connected)
{
return;
}
// 创建连接工厂
// 如果初始化失败，不会启动恢复连接
//factory = new ConnectionFactory()
//{
// HostName = hostName, // 替换为你的 RabbitMQ 服务器地址
// UserName = userName, // 替换为用户名
// Password = password // 替换为密码
//};
string url = $"amqp://{connectData.userName}:{connectData.password}@{connectData.hostName}:{connectData.port}"; //string.Format("amqp://unity:unity@139.9.137.14:5672");
factory = new ConnectionFactory()
{
Uri = new Uri(url)
};
// 自动恢复连接
factory.AutomaticRecoveryEnabled = true;
// 如果由于异常导致恢复失败（例如RabbitMQ节点仍然不可达），它将在固定的时间间隔（默认为5秒）后重试。间隔时间可配置如下
// Connection.CloseAsync 关闭的连接不会启动自动恢复连接
factory.NetworkRecoveryInterval = TimeSpan.FromSeconds(10);
factory.TopologyRecoveryEnabled = true;
while (State != NetWorkState.Connected)
{
if (!dispose)
{
await Connect();
}
}
await Task.Delay(1);
if (!string.IsNullOrEmpty(connectData.receiveQueeu))
{
await BasicConsumer();
}
}
private async Task Connect()
{
try
{
State = NetWorkState.Connecting;
// 异步创建连接
connection = await factory.CreateConnectionAsync();
channel = await connection.CreateChannelAsync();
// 声明队列
QueueDeclareOk queueDeclareOk = await channel.QueueDeclareAsync(
queue: connectData.queue,
durable: false,
exclusive: false,
autoDelete: false,
arguments: null);
/*
autoDelete = true：没有消费者时队列自动删除，通常用于临时或一次性的队列。
autoDelete = false：队列不会自动删除，通常用于需要长期存在的队列。
选择是否设置 autoDelete = true 取决于你是否希望队列在没有消费者时自动删除。如果你的队列是临时的、一次性的，那么使用 autoDelete = true 会更适合；如果队列是长期需要使用的，则设置为 autoDelete = false 会更为合适
*/
State = NetWorkState.Connected;
// 设置消费者的预取计数为10，允许同时处理10条消息
await channel.BasicQosAsync(
prefetchSize: 0,
prefetchCount: 10,
global: false);
Debug.Log("RabbitMQ Connect Success");
GameNotifycation.GetInstance().Notify<NetWorkState>(ENUM_MSG_TYPE.MSG_NETWORK_STATE_CHANGE, State);
}
catch (BrokerUnreachableException e)
{
await Task.Delay(5000);
State = NetWorkState.ConnectFailed;
Debug.LogError("ConnectError:" + e.ToString());
// apply retry logic
}
await Task.Delay(1);
}
/// <summary>
/// 发送消息
/// exchange: 要发布消息的交换机名称。
/// routingKey: 路由键，决定消息应该路由到哪个队列。
/// mandatory: 如果设置为 true，RabbitMQ 会确保消息至少被投递到一个队列。如果没有队列接收该消息，RabbitMQ 会触发 basic.return。
/// immediate: 如果设置为 true，RabbitMQ 会在消息无法立即被消费时丢弃消息。
/// basicProperties: 消息的属性，类型为 IBasicProperties。这些属性可以设置消息的优先级、持久性等。
/// body: 消息体的字节数组。
///
/// BasicPublishAsync 方法没有返回消息投递的结果。它仅仅表示“请求已经被成功发送到 RabbitMQ 的交换机”。如果发布操作成功，Task 会正常完成，不会抛出异常。你可以通过异常处理来捕获潜在的错误。
/// </summary>
/// <param name="msg"></param>
public async Task SendAsync(string message)
{
if (!IsConnect)
{
UnityEngine.Debug.Log("Send not IsConnect");
await StartConnect();
}
try
{
IChannel channel = Channel;
var body = Encoding.UTF8.GetBytes(message);
var props = new BasicProperties();
props.ContentType = "text/plain";
props.DeliveryMode = DeliveryModes.Transient;
await channel.BasicPublishAsync(
exchange: "",
routingKey: Queue,
mandatory: false,
basicProperties: props,
body: body).ConfigureAwait(false);
//Debug.Log($"[x] Sent: Complete");
}
catch (Exception ex)
{
UnityEngine.Debug.LogError($"Error publishing message: {ex.Message}");
}
}
/// <summary>
/// 设置接收消息回调
/// </summary>
/// <param name="receivedCallBack"></param>
public void SetReceive(Action<string, byte[]> receivedCallBack)
{
this.receivedCallBack = receivedCallBack;
}
/// <summary>
/// 创建异步消费者
/// </summary>
/// <returns></returns>
public async Task<string> BasicConsumer()
{
if (!IsConnect)
{
await StartConnect();
}
var consumer = new AsyncEventingBasicConsumer(Channel);
// 处理消息的异步回调逻辑
consumer.ReceivedAsync += ReceivedAsync;
// 开始消费
string result = await Channel.BasicConsumeAsync(
queue: connectData.receiveQueeu, // 指定消费者要监听的队列名称
autoAck: false, // 决定是否自动确认消息。如果 true，消息在交付时会自动确认。如果 false，则需要手动调用 BasicAck 确认消息
consumer: consumer); // 指定消息的处理方式，通过实现 IBasicConsumer 接口来定义如何处理从队列中接收到的消息
/*
autoAck = true：消息一旦传递给消费者，RabbitMQ 就认为该消息被成功处理，无需再确认。
autoAck = false：消费者需要显式地调用 channel.BasicAck 来确认消息的处理，通常用于消息处理失败时能够重试消息。
*/
return result;
}
/// <summary>
/// 异步接收消息
/// 如果 Channel.BasicConsumeAsync 方法中 autoAck 设置为 true，那么 channel.BasicAckAsync 调用是不允许的
/// 想在 Channel.BasicConsumeAsync 消费消息收到消息时调用 channel.BasicAckAsync，必须将 Channel.BasicConsumeAsync 方法中 autoAck 设置为 false
/// </summary>
/// <param name="sender"></param>
/// <param name="eventArgs"></param>
/// <returns></returns>
private async Task ReceivedAsync(object sender, BasicDeliverEventArgs eventArgs)
{
try
{
//Debug.Log("ReceivedAsync");
AsyncEventingBasicConsumer consumer = sender as AsyncEventingBasicConsumer;
string queue = consumer.Channel.CurrentQueue;
var body = eventArgs.Body.ToArray();
receivedCallBack?.Invoke(queue, body);
// 模拟异步任务处理（比如访问数据库或调用其他服务）
await channel.BasicAckAsync(eventArgs.DeliveryTag, false);
}
catch (Exception ex)
{
Debug.LogError($"Error processing message: {ex.Message}");
// 如果处理失败，可以拒绝并重新入队（可选）
//await Channel.BasicNackAsync(eventArgs.DeliveryTag, multiple: false, requeue: true);
}
await Task.Delay(1);
}
/// <summary>
/// 关闭连接
/// </summary>
public async void Dispose()
{
dispose = true;
// 先关闭通道、再关闭连接
if (channel != null) // 通道关闭
{
await channel.CloseAsync();
channel = null;
}
if (connection != null) // 连接关闭
{
UnityEngine.Debug.Log("ConnectDispose");
await connection.CloseAsync();
connection = null;
}
await Task.Delay(1);
}
}

复制代码