xiaoou
e9616125ce
6 major server modules (PdkFriendServer/GlobalSever/ServerCore/GameModule/GameNetModule) + game logic (GameFix/GameDAL/ServerData) + network layer (NetWorkMessage) + data layer (ObjectModel) + utilities (MrWu/Core/Config/CloudAPI/dll) + adapters (zyxAdapter/base) .NET 8.0 C# solution, 16 projects, 958 source files
807 lines
30 KiB
C#
807 lines
30 KiB
C#
/********************************
|
||
*
|
||
* 作者:吴隆健
|
||
* 创建时间: 2019/6/24 14:22:26
|
||
*
|
||
********************************/
|
||
#if RABBITMQ
|
||
using System;
|
||
using System.Collections.Concurrent;
|
||
using System.Collections.Generic;
|
||
using System.Text;
|
||
using System.Threading;
|
||
using RabbitMQ.Client;
|
||
using RabbitMQ.Client.Events;
|
||
|
||
namespace MrWu.RabbitMQ {
|
||
/// <summary>
|
||
/// rabbit管理
|
||
/// </summary>
|
||
public class RabbitManager {
|
||
|
||
/// <summary>
|
||
/// 配置
|
||
/// </summary>
|
||
public RabbitConfig config {
|
||
get;
|
||
private set;
|
||
}
|
||
|
||
/// <summary>
|
||
/// 发送失败
|
||
/// </summary>
|
||
public event MessageHandle SendResult;
|
||
|
||
/// <summary>
|
||
/// 未路由成功事件
|
||
/// </summary>
|
||
public event MessageHandle notRoutingOk;
|
||
|
||
/// <summary>
|
||
/// 构造
|
||
/// </summary>
|
||
public RabbitManager() {
|
||
}
|
||
|
||
/// <summary>
|
||
/// 未确认的数量
|
||
/// </summary>
|
||
public int NoAckCount {
|
||
get {
|
||
return waitAck.Count;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 启动rabbit管理
|
||
/// </summary>
|
||
/// <param name="config"></param>
|
||
public void Start(RabbitConfig config = null) {
|
||
if (connection != null) {
|
||
Debug.Debug.Error("已启动过!");
|
||
return;
|
||
}
|
||
if (config != null)
|
||
this.config = config;
|
||
if (this.config == null)
|
||
throw new ArgumentNullException("config is null");
|
||
|
||
Connection();
|
||
}
|
||
|
||
/// <summary>
|
||
/// 链接
|
||
/// </summary>
|
||
public IConnection connection {
|
||
get;
|
||
private set;
|
||
}
|
||
|
||
/// <summary>
|
||
/// 生产者信道
|
||
/// </summary>
|
||
private IModel producer {
|
||
get;
|
||
set;
|
||
}
|
||
|
||
/// <summary>
|
||
/// 获取信息的通道
|
||
/// </summary>
|
||
private IModel infoChannel {
|
||
get;
|
||
set;
|
||
}
|
||
|
||
/// <summary>
|
||
/// 等待发送确认的消息
|
||
/// </summary>
|
||
private ConcurrentDictionary<ulong, Message> waitAck { get; } = new ConcurrentDictionary<ulong, Message>();
|
||
|
||
/// <summary>
|
||
/// 生产者锁
|
||
/// </summary>
|
||
private readonly object lockobj = new object();
|
||
|
||
/// <summary>
|
||
/// 链接rabbit
|
||
/// </summary>
|
||
private void Connection() {
|
||
ConnectionFactory factory = new ConnectionFactory();
|
||
factory.UserName = config.username;
|
||
factory.Password = config.password;
|
||
factory.VirtualHost = config.vhost;
|
||
factory.HostName = config.host;
|
||
factory.Port = config.port;
|
||
factory.AutomaticRecoveryEnabled = true;
|
||
|
||
if (connection != null) {
|
||
connection.ConnectionBlocked -= ConnectionBlockedEventHandler;
|
||
connection.ConnectionUnblocked -= ConnectionUnblockedEventHandler;
|
||
}
|
||
|
||
|
||
connection = factory.CreateConnection();
|
||
connection.ConnectionBlocked += ConnectionBlockedEventHandler;
|
||
connection.ConnectionUnblocked += ConnectionUnblockedEventHandler;
|
||
|
||
Debug.Debug.Log("链接创建生产者!");
|
||
CreateProducer();
|
||
DeclareExchange(config.exchanges);
|
||
DeclareQueue(config.queues);
|
||
Bind(config.binds);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 创建生产者
|
||
/// </summary>
|
||
private void CreateProducer() {
|
||
try {
|
||
|
||
Debug.Debug.Log("创建生产者:" + Thread.CurrentThread.ManagedThreadId);
|
||
|
||
producer = connection.CreateModel();
|
||
producer.ConfirmSelect();
|
||
producer.ModelShutdown += ModelShutdown;
|
||
producer.BasicReturn += MessageReturn;
|
||
producer.BasicAcks += MessageAck;
|
||
producer.BasicNacks += MessageNAck;
|
||
} catch (Exception e) {
|
||
Debug.Debug.Error("创建生产者出错!" + e);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 消息阻塞!!!
|
||
/// </summary>
|
||
/// <param name="sender"></param>
|
||
/// <param name="args"></param>
|
||
private void ConnectionBlockedEventHandler(IConnection sender, ConnectionBlockedEventArgs args) {
|
||
Debug.Debug.Fatal("RabbitMQ 消息阻塞,链接被禁用!" + args.Reason);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 消息阻塞被解除!
|
||
/// </summary>
|
||
/// <param name="sender"></param>
|
||
private void ConnectionUnblockedEventHandler(IConnection sender) {
|
||
Debug.Debug.Fatal("RabbitMQ 消息解除阻塞!");
|
||
}
|
||
|
||
/// <summary>
|
||
/// 链接禁用!
|
||
/// </summary>
|
||
public void ConnectionBlocked() {
|
||
if (connection == null)
|
||
return;
|
||
|
||
connection.HandleConnectionBlocked("auto blocked");
|
||
}
|
||
|
||
/// <summary>
|
||
/// 解除链接禁用
|
||
/// </summary>
|
||
public void ConnectionUnBlocked() {
|
||
if (connection == null)
|
||
return;
|
||
connection.HandleConnectionUnblocked();
|
||
}
|
||
|
||
/// <summary>
|
||
/// 创建信息通道
|
||
/// </summary>
|
||
private void CreateInfoChannel() {
|
||
try {
|
||
infoChannel = connection.CreateModel();
|
||
infoChannel.ModelShutdown += ModelShutdown;
|
||
} catch (Exception e) {
|
||
Debug.Debug.Error("创建信息获取通道出错!" + e.ToString());
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 消息发送成功
|
||
/// </summary>
|
||
/// <param name="model"></param>
|
||
/// <param name="args"></param>
|
||
private void MessageAck(IModel model, BasicAckEventArgs args) {
|
||
DoAck(model, args.DeliveryTag, args.Multiple, true);
|
||
}
|
||
|
||
/// <summary>
|
||
///
|
||
/// </summary>
|
||
/// <param name="model"></param>
|
||
/// <param name="deliverytag"></param>
|
||
/// <param name="multiple"></param>
|
||
/// <param name="success"></param>
|
||
private void DoAck(IModel model, ulong deliverytag, bool multiple, bool success) {
|
||
|
||
void DoAck(Message _msg, bool _success) {
|
||
_msg.success = success;
|
||
if (_msg.SendResult != null)
|
||
_msg.SendResult(_msg);
|
||
else
|
||
SendResult?.Invoke(_msg);
|
||
|
||
if (_msg.isAutoFree)
|
||
MessagePool.PutMessage(ref _msg);
|
||
}
|
||
|
||
Message msg;
|
||
|
||
//同步确认的消息也会到这里来!!!
|
||
if (!multiple) {
|
||
if (waitAck.TryRemove(deliverytag, out msg)) {//拿到
|
||
DoAck(msg, success);
|
||
}
|
||
} else {
|
||
List<ulong> keys = new List<ulong>();
|
||
foreach (var item in waitAck) {
|
||
if (item.Value.deliveryTag <= deliverytag)
|
||
keys.Add(item.Value.deliveryTag);
|
||
}
|
||
foreach (var item in keys) {
|
||
if (waitAck.TryRemove(item, out msg)) {
|
||
DoAck(msg, success);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
|
||
|
||
/// <summary>
|
||
/// 消息未被发送
|
||
/// </summary>
|
||
/// <param name="model"></param>
|
||
/// <param name="args"></param>
|
||
private void MessageNAck(IModel model, BasicNackEventArgs args) {
|
||
DoAck(model, args.DeliveryTag, args.Multiple, false);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 信道断开
|
||
/// </summary>
|
||
/// <param name="model"></param>
|
||
/// <param name="reason"></param>
|
||
private void ModelShutdown(IModel model, ShutdownEventArgs reason) {
|
||
|
||
Debug.Debug.Warning("信道断开事件:" + Thread.CurrentThread.ManagedThreadId);
|
||
|
||
if (producer != null && producer.IsClosed) {
|
||
Debug.Debug.Info("剩余确认数量:" + waitAck.Count);
|
||
// producer.BasicNack
|
||
producer.ModelShutdown -= ModelShutdown;
|
||
producer.BasicReturn -= MessageReturn;
|
||
producer.BasicAcks -= MessageAck;
|
||
producer.BasicNacks -= MessageNAck;
|
||
|
||
waitAck.Clear();
|
||
|
||
producer = null;
|
||
} else if (infoChannel != null && infoChannel.IsClosed)
|
||
infoChannel = null;
|
||
Debug.Debug.Error("信道断开:" + reason);
|
||
}
|
||
|
||
|
||
/// <summary>
|
||
/// 消息被返回
|
||
/// </summary>
|
||
/// <param name="model"></param>
|
||
/// <param name="args"></param>
|
||
private void MessageReturn(IModel model, BasicReturnEventArgs args) {
|
||
|
||
if (args.BasicProperties != null) {
|
||
string msgid = args.BasicProperties.MessageId;
|
||
Message msg = MessagePool.GetMessage(msgid);// new Message(msgid);
|
||
msg.noAck = true;
|
||
msg.body = args.Body;
|
||
msg.channel = model;
|
||
msg.exchange = args.Exchange;
|
||
msg.routingkey = args.RoutingKey;
|
||
msg.userData = args.BasicProperties.Headers;
|
||
|
||
notRoutingOk?.Invoke(msg);
|
||
} else {
|
||
Debug.Debug.Error("怎么可能BasicProperties 不存在");
|
||
}
|
||
|
||
Debug.Debug.Error("消息被撤回,ReplyText:" + args.Exchange + "," + args.RoutingKey +
|
||
"," + args.ReplyText + ",ReplyCode:" + args.ReplyCode + ",body:" + Encoding.UTF8.GetString(args.Body));
|
||
}
|
||
|
||
/// <summary>
|
||
/// 声明路由
|
||
/// </summary>
|
||
public void DeclareExchange(ExchangeInfo[] infos) {
|
||
if (infos != null) {
|
||
int len = infos.Length;
|
||
for (int i = 0; i < len; i++) {
|
||
ExchangeInfo exchange = infos[i];
|
||
IDictionary<string, object> properties = new Dictionary<string, object>();
|
||
if (!string.IsNullOrEmpty(exchange.alternate_exchange)) {
|
||
properties.Add("alternate-exchange", exchange.alternate_exchange);
|
||
}
|
||
|
||
if (exchange.delExists)
|
||
DelExchange(exchange.name);
|
||
|
||
//Debug.Debug.Log("进入锁44444"+Thread.CurrentThread.ManagedThreadId);
|
||
lock (lockobj) {
|
||
producer.ExchangeDeclare(exchange.name, exchange.type, exchange.durable, exchange.autodelete, properties);
|
||
}
|
||
//Debug.Debug.Log("退出锁4444");
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 声明队列
|
||
/// </summary>
|
||
public QueueDeclareOk DeclareQueue() {
|
||
QueueDeclareOk result = null;
|
||
//Debug.Debug.Log("进入锁3333"+Thread.CurrentThread.ManagedThreadId);
|
||
lock (lockobj) {
|
||
try {
|
||
result = producer.QueueDeclare();
|
||
} catch {
|
||
return null;
|
||
}
|
||
}
|
||
//Debug.Debug.Log("退出锁3333");
|
||
return result;
|
||
}
|
||
|
||
/// <summary>
|
||
///声明队列
|
||
/// </summary>
|
||
public void DeclareQueue(QueueInfo[] infos) {
|
||
if (infos != null) {
|
||
int len = infos.Length;
|
||
for (int i = 0; i < len; i++) {
|
||
QueueInfo queue = infos[i];
|
||
IDictionary<string, object> properties = new Dictionary<string, object>();
|
||
if (!string.IsNullOrEmpty(queue.x_dead_letter_exchange)) {
|
||
properties.Add("x-dead-letter-exchange", queue.x_dead_letter_exchange);
|
||
}
|
||
if (!string.IsNullOrEmpty(queue.x_dead_routing_key)) {
|
||
properties.Add("x-dead-routing-key", queue.x_dead_routing_key);
|
||
}
|
||
if (queue.x_expires > 0) {
|
||
properties.Add("x-expires", queue.x_expires);
|
||
}
|
||
if (queue.x_max_length > 0) {
|
||
properties.Add("x-max-length", queue.x_max_length);
|
||
}
|
||
if (queue.x_max_length_bytes > 0) {
|
||
properties.Add("x-max-length-bytes", queue.x_max_length_bytes);
|
||
}
|
||
if (queue.x_max_priority > 0) {
|
||
properties.Add("x-max-priority", queue.x_max_priority);
|
||
}
|
||
if (queue.x_message_ttl > 0) {
|
||
properties.Add("x-message-ttl", queue.x_message_ttl);
|
||
}
|
||
if (!string.IsNullOrEmpty(queue.x_overflow)) {
|
||
properties.Add("x-overflow", queue.x_overflow);
|
||
}
|
||
if (!string.IsNullOrEmpty(queue.x_queue_mode)) {
|
||
properties.Add("x-queue-mode", queue.x_queue_mode);
|
||
}
|
||
if (queue.delExists)
|
||
DelQueue(queue.name);
|
||
|
||
//Debug.Debug.Log("进入锁7454354354"+Thread.CurrentThread.ManagedThreadId);
|
||
lock (lockobj) {
|
||
producer.QueueDeclare(queue.name, queue.durable, queue.exclusive, queue.autodelete, properties);
|
||
if (queue.startClear)
|
||
producer.QueuePurge(queue.name);
|
||
}
|
||
//Debug.Debug.Log("退出锁7454354354"+Thread.CurrentThread.ManagedThreadId);
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 绑定
|
||
/// </summary>
|
||
public void Bind(BindInfo[] infos) {
|
||
if (infos != null) {
|
||
int len = infos.Length;
|
||
for (int i = 0; i < len; i++) {
|
||
BindInfo bind = infos[i];
|
||
switch (bind.bindType) {
|
||
case "queue":
|
||
//Debug.Debug.Log("进入锁asdasfasf"+Thread.CurrentThread.ManagedThreadId);
|
||
lock (lockobj) {
|
||
producer.QueueBind(bind.destination, bind.source, bind.routingkey);
|
||
}
|
||
//Debug.Debug.Log("退出锁asdasfasf"+Thread.CurrentThread.ManagedThreadId);
|
||
break;
|
||
case "exchange":
|
||
//Debug.Debug.Log("进入锁asdsafasfasd"+Thread.CurrentThread.ManagedThreadId);
|
||
lock (lockobj) {
|
||
producer.ExchangeBind(bind.destination, bind.source, bind.routingkey);
|
||
}
|
||
//Debug.Debug.Log("退出锁asdsafasfasd"+Thread.CurrentThread.ManagedThreadId);
|
||
break;
|
||
default:
|
||
throw new ArgumentException("BindType value is not queue/exchange");
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 获取队列信息
|
||
/// </summary>
|
||
/// <param name="name">名称</param>
|
||
/// <returns>队列信息</returns>
|
||
public QueueDeclareOk GetQueueInfo(string name) {
|
||
try {
|
||
IModel channel = connection.CreateModel();
|
||
var result = channel.QueueDeclarePassive(name);
|
||
return result;
|
||
} catch {
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 交换机存在不存在
|
||
/// </summary>
|
||
/// <param name="name">交换机名称</param>
|
||
/// <returns>true 表示存在 false 可能是网络故障，或者不存在</returns>
|
||
public bool ExchangeExists(string name) {
|
||
try {
|
||
IModel channel = connection.CreateModel();
|
||
channel.ExchangeDeclarePassive(name);
|
||
return true;
|
||
} catch {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
private readonly ConcurrentBag<RpcClient> rpcs = new ConcurrentBag<RpcClient>();
|
||
|
||
/// <summary>
|
||
/// 创建RPC客户端
|
||
/// </summary>
|
||
/// <returns></returns>
|
||
public RpcClient CreateRpcClient(ushort prefetchCount = 3) {
|
||
var rpc = new RpcClient(config, prefetchCount);
|
||
rpcs.Add(rpc);
|
||
return rpc;
|
||
}
|
||
|
||
/// <summary>
|
||
/// 创建信道
|
||
/// </summary>
|
||
/// <returns></returns>
|
||
internal IModel CreateIModel() {
|
||
return connection.CreateModel();
|
||
}
|
||
|
||
/// <summary>
|
||
/// 获取消息属性
|
||
/// </summary>
|
||
/// <returns></returns>
|
||
public IBasicProperties GetBasicProperties() {
|
||
try {
|
||
|
||
IBasicProperties result = null;
|
||
//Debug.Debug.Log("进入锁6666" + Thread.CurrentThread.ManagedThreadId);
|
||
lock (lockobj) {
|
||
//Debug.Debug.Log("生产者:" + (producer == null));
|
||
if (producer == null)
|
||
CreateProducer();
|
||
if (producer == null)
|
||
result = null;
|
||
else
|
||
result = producer.CreateBasicProperties();
|
||
}
|
||
return result;
|
||
} catch (Exception e) {
|
||
Debug.Debug.Error("获取消息属性出错:" + e.ToString());
|
||
return null;
|
||
} finally {
|
||
//Debug.Debug.Log("退出锁6666");
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 无返回值的发送,按配置来决定是异步还是同步确认
|
||
/// </summary>
|
||
public void BasicPublish(Message msg, bool isAutoFree) {
|
||
BasicPublish(msg, config.AsynConfirm, isAutoFree);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 有返回值的发送_必须是同步，马上需要知道结果
|
||
/// </summary>
|
||
/// <returns></returns>
|
||
public bool BasicPublish_Result(Message msg, bool isAutoFree) {
|
||
return BasicPublish(msg, false, isAutoFree);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 发送消息
|
||
/// </summary>
|
||
/// <param name="msg">消息</param>
|
||
/// <param name="isAsyn">isAsyn 是否异步</param>
|
||
/// <param name="isAutoFree">是否自动释放</param>
|
||
private bool BasicPublish(Message msg, bool isAsyn, bool isAutoFree) {
|
||
if (msg == null || msg.body == null || msg.body.Length == 0) {
|
||
Debug.Debug.Error("发送数据为空!" + new System.Diagnostics.StackTrace().ToString());
|
||
return false;
|
||
}
|
||
|
||
msg.isAutoFree = isAutoFree;
|
||
bool isNotice = false;
|
||
bool result = false;
|
||
try {
|
||
//Debug.Debug.Log("进入锁1111" + Thread.CurrentThread.ManagedThreadId);
|
||
lock (lockobj) {
|
||
//Debug.Debug.Log("处理发送!" + (producer == null));
|
||
if (producer == null) {
|
||
CreateProducer();
|
||
}
|
||
//Debug.Debug.Log("111");
|
||
if (producer == null) {
|
||
isNotice = true;
|
||
msg.success = false;
|
||
} else {
|
||
msg.channel = producer;
|
||
|
||
msg.noAck = true;
|
||
msg.deliveryTag = producer.NextPublishSeqNo;
|
||
if (msg.propertis == null)
|
||
msg.propertis = producer.CreateBasicProperties();
|
||
msg.propertis.MessageId = msg.id;
|
||
msg.propertis.Headers = msg.userData;
|
||
//持久化
|
||
msg.propertis.SetPersistent(msg.durable);
|
||
//Debug.Debug.Info($"读取到过期时间:{msg.propertis.Expiration}");
|
||
if (!msg.durable)
|
||
{
|
||
msg.propertis.Expiration = "3000";
|
||
}
|
||
|
||
//Debug.Debug.Log("发送消息:" + msg.bodystr + "," + msg.exchange + "," + msg.propertis + "," + msg.mandatory);
|
||
|
||
producer.BasicPublish(msg.exchange, msg.routingkey, msg.mandatory, false, msg.propertis, msg.body);
|
||
if (!isAsyn) {//同步
|
||
//Debug.Debug.Log("同步等待!");
|
||
msg.success = producer.WaitForConfirms();
|
||
result = msg.success;
|
||
if (!msg.success) {
|
||
Console.WriteLine(msg.success);
|
||
}
|
||
isNotice = true;
|
||
|
||
} else { //异步 //添加到列表
|
||
if (producer.IsClosed) {
|
||
isNotice = true;
|
||
msg.success = false;
|
||
} else {
|
||
//Debug.Debug.Log("异步发送:" + msg.deliveryTag);
|
||
|
||
//Debug.Debug.Log("得到数据:" + waitAck.Count);
|
||
waitAck.GetOrAdd(msg.deliveryTag, msg);
|
||
|
||
//Debug.Debug.Log("得到数据:" + waitAck.Count);
|
||
|
||
}
|
||
}
|
||
}
|
||
}
|
||
} catch (Exception e) {
|
||
Debug.Debug.Warning("发送消息失败:" + e.ToString() + Environment.NewLine + msg.ToString());
|
||
isNotice = true;
|
||
msg.success = false;
|
||
}
|
||
|
||
if (isNotice) {
|
||
if (msg.SendResult != null)
|
||
msg.SendResult(msg);
|
||
else
|
||
SendResult?.Invoke(msg);
|
||
|
||
if (msg.isAutoFree)
|
||
MessagePool.PutMessage(ref msg);
|
||
}
|
||
|
||
return result;
|
||
}
|
||
|
||
/// <summary>
|
||
/// 删除队列
|
||
/// </summary>
|
||
/// <param name="queue">队列名称</param>
|
||
/// <param name="ifUnused">是否没有用户才删除</param>
|
||
/// <param name="ifEmpty">是否是空队列才删除</param>
|
||
/// <returns>uint 0表示删除失败 大于等于1表示删除成功 队列中消息数量+1</returns>
|
||
public uint DelQueue(string queue, bool ifUnused = false, bool ifEmpty = false) {
|
||
IModel model = connection.CreateModel();
|
||
uint result = 0;
|
||
try {
|
||
result = model.QueueDelete(queue, ifUnused, ifEmpty);
|
||
result += 1;
|
||
} catch (Exception e) {
|
||
Debug.Debug.Warning("删除队列异常:" + e.ToString());
|
||
}
|
||
return result;
|
||
}
|
||
|
||
/// <summary>
|
||
/// 删除交换机
|
||
/// </summary>
|
||
/// <param name="exchange">交换机</param>
|
||
/// <param name="ifUnused">是否没用户才删除</param>
|
||
/// <returns>true 表示删除成功 false 队列不存在</returns>
|
||
public bool DelExchange(string exchange, bool ifUnused = false) {
|
||
IModel model = connection.CreateModel();
|
||
try {
|
||
model.ExchangeDelete(exchange, ifUnused);
|
||
} catch (Exception e) {
|
||
Debug.Debug.Warning("删除交换机异常:" + e.ToString());
|
||
return false;
|
||
}
|
||
return true;
|
||
}
|
||
|
||
/// <summary>
|
||
/// 获取消息
|
||
/// </summary>
|
||
/// <param name="queue">队列</param>
|
||
/// <param name="noAck">true 表示自动确认 false 表示手动确认</param>
|
||
/// <returns></returns>
|
||
public Message GetMessage(string queue, bool noAck = true) {
|
||
//Debug.Debug.Log("进入锁2222" + Thread.CurrentThread.ManagedThreadId);
|
||
lock (lockobj) {
|
||
if (infoChannel == null)
|
||
CreateInfoChannel();
|
||
if (infoChannel == null)
|
||
return null;
|
||
}
|
||
//Debug.Debug.Log("退出锁2222");
|
||
IModel key = infoChannel;
|
||
Message msg = null;
|
||
lock (key) {
|
||
try {
|
||
BasicGetResult bgr = key.BasicGet(queue, noAck);
|
||
|
||
if (bgr != null) {
|
||
string msgid = null;
|
||
IDictionary<string, object> userdata = null;
|
||
if (bgr.BasicProperties == null) {
|
||
Debug.Debug.Error("这个消息竟然没有属性!");
|
||
} else {
|
||
msgid = bgr.BasicProperties.MessageId;
|
||
userdata = bgr.BasicProperties.Headers;
|
||
}
|
||
msg = MessagePool.GetMessage(msgid);// new Message(msgid);
|
||
msg.noAck = noAck;
|
||
msg.exchange = bgr.Exchange;
|
||
msg.routingkey = bgr.RoutingKey;
|
||
msg.propertis = bgr.BasicProperties;
|
||
msg.body = bgr.Body;
|
||
msg.deliveryTag = bgr.DeliveryTag;
|
||
msg.channel = key;
|
||
msg.redelivered = bgr.Redelivered;
|
||
msg.messageCount = bgr.MessageCount;
|
||
msg.userData = userdata;
|
||
}
|
||
} catch (Exception e) {
|
||
Debug.Debug.Error("获取消息出错:" + e.ToString());
|
||
return null;
|
||
}
|
||
}
|
||
return msg;
|
||
}
|
||
|
||
/// <summary>
|
||
/// 添加消费者
|
||
/// </summary>
|
||
/// <param name="consumer">消费者</param>
|
||
public void AddConsumer(Consumer consumer) {
|
||
IModel model = connection.CreateModel();
|
||
consumer.channel = model;
|
||
if (string.IsNullOrEmpty(consumer.consumertag))
|
||
consumer.consumertag = Guid.NewGuid().ToString();
|
||
|
||
consumer.consumertag = model.BasicConsume(consumer.queue, consumer.noAck, consumer.consumertag,
|
||
consumer.noLoacl, consumer.exclusive, consumer.arguments, new BasicConsumer(consumer));
|
||
|
||
if (consumer.prefetchCount > 0) {
|
||
model.BasicQos(0, consumer.prefetchCount, true);
|
||
}
|
||
|
||
}
|
||
|
||
/// <summary>
|
||
/// 移出消费者
|
||
/// </summary>
|
||
/// <param name="consumer"></param>
|
||
public void RemoveConsumer(Consumer consumer) {
|
||
if (consumer == null || consumer.channel == null || consumer.channel.IsClosed)
|
||
return;
|
||
consumer.channel.BasicCancel(consumer.consumertag);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 关闭
|
||
/// </summary>
|
||
public void Close() {
|
||
if (connection != null) {
|
||
connection.Dispose();
|
||
connection = null;
|
||
}
|
||
foreach (var item in rpcs) {
|
||
item.Close();
|
||
}
|
||
}
|
||
|
||
|
||
/// <summary>
|
||
/// 消费者类
|
||
/// </summary>
|
||
private class BasicConsumer : DefaultBasicConsumer {
|
||
|
||
public Consumer consumer;
|
||
|
||
public BasicConsumer(Consumer consumer) {
|
||
this.consumer = consumer;
|
||
}
|
||
|
||
public override void HandleBasicCancel(string consumerTag) {
|
||
base.HandleBasicCancel(consumerTag);
|
||
}
|
||
|
||
public override void HandleBasicConsumeOk(string consumerTag) {
|
||
base.HandleBasicConsumeOk(consumerTag);
|
||
}
|
||
|
||
public override void HandleBasicCancelOk(string consumerTag) {
|
||
base.HandleBasicCancelOk(consumerTag);
|
||
}
|
||
|
||
public override void HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, IBasicProperties properties, byte[] body) {
|
||
string msgid = null;
|
||
IDictionary<string, object> userdata = null;
|
||
if (properties == null) {
|
||
Debug.Debug.Error("消费者接收消息异常,没有收到消息id");
|
||
} else {
|
||
msgid = properties.MessageId;
|
||
userdata = properties.Headers;
|
||
}
|
||
|
||
Message message = MessagePool.GetMessage(msgid); // new Message(msgid);
|
||
message.exchange = exchange;
|
||
message.routingkey = routingKey;
|
||
message.noAck = consumer.noAck;
|
||
message.propertis = properties;
|
||
message.redelivered = redelivered;
|
||
message.deliveryTag = deliveryTag;
|
||
message.channel = consumer.channel;
|
||
message.body = body;
|
||
message.userData = userdata;
|
||
|
||
//去处理消息
|
||
consumer.Receive?.Invoke(message);
|
||
//base.HandleBasicDeliver(consumerTag, deliveryTag, redelivered, exchange, routingKey, properties, body);
|
||
}
|
||
|
||
public override void HandleModelShutdown(IModel model, ShutdownEventArgs reason) {
|
||
base.HandleModelShutdown(model, reason);
|
||
}
|
||
}
|
||
|
||
|
||
}
|
||
|
||
|
||
}
|
||
#endif |