hjha-server/GlobalSever/MQ/GlobalMQ.cs

/*
 * 由SharpDevelop创建。
 * 用户： Administrator
 * 日期: 2018-10-30
 * 时间: 16:00
 *
 * 要改变这种模板请点击 工具|选项|代码编写|编辑标准头文件
 */

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using GameData;
using GameDAL;
using MrWu.Debug;
using MrWu.RabbitMQ;
using RabbitMQ.Client;
using Server.Data.Module;
using Server.Module;
using Server.MQ;
using Server.Pack;

namespace Server.MQ
{
    /*
     * 设计方案
     *
     * exit_game_1 : 提示玩家网络故障，踢出玩家所在的游戏
     *
     *
     *
     * */

    /// <summary>
    /// 消息管理 工厂
    /// </summary>
    public abstract class IGlobalFactory
    {
        /// <summary>
        /// 创建消息管理
        /// </summary>
        public abstract void CreateGlobalMQ();
    }

    /// <summary>
    /// rabbit消息投递管理类
    /// </summary>
    public class GlobalMQ : IGlobalMQ
    {
        /// <summary>
        /// 消息管理工厂
        /// </summary>
        public class Factory : IGlobalFactory
        {
            /// <summary>
            /// 创建消息管理类
            /// </summary>
            public override void CreateGlobalMQ()
            {
                if (instance == null)
                    instance = new GlobalMQ();
            }
        }

        /// <summary>
        /// rabbit消息投递管理类单例
        /// </summary>
        public static GlobalMQ instance { get; protected set; }

        /// <summary>
        /// 失败原因的 key 消息未发送成功(502) 消息未被路由(503) 消息被扔弃(504)
        /// </summary>
        public const string failCode = "failCode";

        /// <summary>
        /// 应该由谁来处理这个未被路由或被扔弃的包
        /// </summary>
        public const string doFailKey = "doFail";

        /// <summary>
        /// 应该由哪个节点来处理扔弃的包
        /// </summary>
        public const string doFailNodeNum = "doFailNodeNum";

        /// <summary>
        /// 处理失败的标记-有标记的才会处理,没有的会被扔弃
        /// </summary>
        public const string failTagKey = "failTag";

        /// <summary>
        /// 发送成功要做的事
        /// </summary>
        public const string successTagKey = "successTag";

        /// <summary>
        /// 构造
        /// </summary>
        protected GlobalMQ()
        {
        }

        /// <summary>
        /// 消息管理
        /// </summary>
        protected RabbitManager rabbit { get; private set; }

        /// <summary>
        /// rpc客户端
        /// </summary>
        public RpcClient rpcClient { get; private set; }

        /// <summary>
        /// 
        /// </summary>
        /// <returns></returns>
        public int GetNoAckCount()
        {
            if (rabbit == null)
                return 0;
            return rabbit.NoAckCount;
        }

        /// <summary>
        /// 发送完成事件
        /// </summary>
        /// <param name="msg"></param>
        public virtual bool SendOK(Message msg)
        {
            return true;
        }

        /// <summary>
        /// 发送失败
        /// </summary>
        /// <param name="msg"></param>
        public virtual void SendFail(Message msg)
        {
            return;
            if (msg.userData == null || !msg.userData.ContainsKey(failCode)
                                     || !msg.userData.ContainsKey(failTagKey) || !msg.userData.ContainsKey(doFailKey))
            {
                if (!msg.noAck && msg.deliveryTag > 0)
                    msg.Ack();
                return;
            }

            try
            {
                string dofailTag = msg.userData[failTagKey].ToString();

                //switch(dofailTag){ 

                //    };
                msg.Ack();
                Debug.Log("处理发送失败的消息:" + dofailTag + "," + msg);
                return;
            }
            catch (Exception e)
            {
                Debug.Error("处理发送失败消息时出现错误,msg:{0},exception:{1}", msg, e.ToString());
                msg.NAck();
                return;
            }
        }

        /// <summary>
        /// 链接阻塞
        /// </summary>
        public void ConnectionBlocked()
        {
            //if (rabbit == null)
            //    return;
            //rabbit.ConnectionBlocked();
        }

        /// <summary>
        /// 解除链接阻塞
        /// </summary>
        public void ConnectionUnBlocked()
        {
            //if (rabbit == null)
            //    return;
            //rabbit.ConnectionUnBlocked();
        }

        /// <summary>
        /// 发送消息
        /// </summary>
        /// <param name="msg">消息</param>
        /// <param name="isAutoFree">是否自动释放</param>
        public virtual void SendMessage(Message msg, bool isAutoFree)
        {
            //CheckFiltersAndPublishNoReturn(rabbit.BasicPublish, msg, isAutoFree, Recv2ReturnDescription.One_Void);//
            rabbit.BasicPublish(msg, isAutoFree);
        }

        /// <summary>
        /// 发送消息并要得到发送结果
        /// </summary>
        /// <param name="msg"></param>
        ///<param name="isAutoFree">是否自动释放</param>
        public virtual bool SendMessage_Result(Message msg, bool isAutoFree)
        {
            // return CheckFiltersAndPublish(rabbit.BasicPublish_Result, msg, isAutoFree, Recv2ReturnDescription.One_Bool);//
            return rabbit.BasicPublish_Result(msg, isAutoFree);
        }

        private ModuleUtile myUtile;

        /// <summary>
        /// rabbit消息管理启动
        /// </summary>
        /// <param name="config">rabbit配置</param>
        /// <param name="utile">模块</param>
        /// <param name="DelExists">是否删除已存在的路由,大升级要使用到,路由规则已改</param>
        /// <returns>启动成功还是失败</returns>
        public bool StartModule(RabbitConfig config, ModuleUtile utile, bool DelExists = false)
        {
            if (rabbit != null)
                return false;

            myUtile = utile;
            RountingManager.instance.Start((ModuleType)myUtile.id, myUtile.nodeNum);
            AddConfig(config, utile, DelExists);
            rabbit = new RabbitManager();

            rabbit.SendResult += SendResult;
            rabbit.notRoutingOk += NotRoutingOk;
            rabbit.Start(config);

            rpcClient = rabbit.CreateRpcClient();

            CreateConsumer();
            return true;
        }

        /// <summary>
        /// 添加配置
        /// </summary>
        /// <param name="config">配置</param>
        /// <param name="utile">当前模块</param>
        /// <param name="delExists">是否声明前先删除模块</param>
        public virtual void AddConfig(RabbitConfig config, ModuleUtile utile, bool delExists)
        {
            string exbak = "Expired"; //过期的交换机 过期的队列

            ExchangeInfo[] exchangeInfos = new ExchangeInfo[3];
            ExchangeInfo info = new ExchangeInfo();
            info.name = RountingManager.instance.clubsterRouteName;
            info.type = "topic";
            info.durable = true;
            info.autodelete = false;
            info.delExists = delExists;
            // info.alternate_exchange = exbak;

            exchangeInfos[0] = info;

            info = new ExchangeInfo();
            info.name = RountingManager.instance.clusterAllRouteName;
            info.type = "fanout";
            info.durable = true;
            info.autodelete = false;
            info.delExists = delExists;
            //   info.alternate_exchange = exbak;

            exchangeInfos[1] = info;

            info = new ExchangeInfo();
            info.name = RountingManager.instance.clusterNodeOne;
            info.type = "direct";
            info.durable = true;
            info.autodelete = false;
            info.delExists = delExists;
            //    info.alternate_exchange = exbak;

            exchangeInfos[2] = info;

            config.exchanges = exchangeInfos;

            QueueInfo[] queueinfos = new QueueInfo[2];
            QueueInfo queue = new QueueInfo();
            queue.name = RountingManager.instance.queueNode;
            queue.durable = true;
            queue.exclusive = false;
            queue.startClear = false;
            queue.delExists = delExists;
            queue.autodelete = false;
            queue.x_message_ttl = 0;
            //      queue.x_dead_letter_exchange = exbak;
            //      queue.x_dead_routing_key = "expire";
            queueinfos[0] = queue;

            queue = new QueueInfo();
            queue.name = RountingManager.instance.queueEveryOne;
            queue.durable = true;
            queue.exclusive = false;
            queue.startClear = false;
            queue.delExists = delExists;
            queue.autodelete = false;
            queue.x_message_ttl = 0;
            //      queue.x_dead_letter_exchange = exbak;
            //      queue.x_dead_routing_key = "expire";
            queueinfos[1] = queue;

            config.queues = queueinfos;

            BindInfo[] bindInfos = new BindInfo[6];
            BindInfo bindInfo = new BindInfo();
            bindInfo.destination = RountingManager.instance.queueNode;
            bindInfo.source = RountingManager.instance.clusterAllRouteName;
            bindInfo.routingkey = RountingManager.instance.routeallkey;
            bindInfo.bindType = "queue";

            Debug.Info("绑定信息,目标:" + bindInfo.destination + ",源:" + bindInfo.source + ",路由键:" + bindInfo.routingkey);

            bindInfos[0] = bindInfo;

            bindInfo = new BindInfo();
            bindInfo.destination = RountingManager.instance.queueNode;
            bindInfo.source = RountingManager.instance.clusterNodeOne;
            bindInfo.routingkey = RountingManager.instance.nodeRouteKey;
            bindInfo.bindType = "queue";

            Debug.Info("绑定信息,目标:" + bindInfo.destination + ",源:" + bindInfo.source + ",路由键:" + bindInfo.routingkey);

            bindInfos[1] = bindInfo;

            bindInfo = new BindInfo();
            bindInfo.destination = RountingManager.instance.queueEveryOne;
            bindInfo.source = RountingManager.instance.clubsterRouteName;
            bindInfo.routingkey = RountingManager.instance.routeEveryOneKey;
            bindInfo.bindType = "queue";

            Debug.Info("绑定信息,目标:" + bindInfo.destination + ",源:" + bindInfo.source + ",路由键:" + bindInfo.routingkey);

            bindInfos[2] = bindInfo;

            bindInfo = new BindInfo();
            bindInfo.destination = RountingManager.instance.clubsterRouteName;
            bindInfo.source = RountingManager.AllServerRouteName;
            bindInfo.routingkey = RountingManager.instance.routeallkey;
            bindInfo.bindType = "exchange";

            Debug.Info("绑定信息,目标:" + bindInfo.destination + ",源:" + bindInfo.source + ",路由键:" + bindInfo.routingkey);

            bindInfos[3] = bindInfo;

            bindInfo = new BindInfo();
            bindInfo.destination = RountingManager.instance.clusterAllRouteName;
            bindInfo.source = RountingManager.instance.clubsterRouteName;
            bindInfo.routingkey = RountingManager.instance.routeallkey;
            bindInfo.bindType = "exchange";

            Debug.Info("绑定信息,目标:" + bindInfo.destination + ",源:" + bindInfo.source + ",路由键:" + bindInfo.routingkey);

            bindInfos[4] = bindInfo;

            bindInfo = new BindInfo();

            bindInfo.destination = RountingManager.instance.clusterNodeOne;
            bindInfo.source = RountingManager.instance.clubsterRouteName;
            bindInfo.routingkey = RountingManager.instance.nodeRouteMatch;
            bindInfo.bindType = "exchange";

            Debug.Info("绑定信息,目标:" + bindInfo.destination + ",源:" + bindInfo.source + ",路由键:" + bindInfo.routingkey);

            bindInfos[5] = bindInfo;

            config.binds = bindInfos;
        }

        /// <summary>
        /// 是否开启接收消息日志
        /// </summary>
        public static bool openMsgLog;

        /// <summary>
        /// 接收到消息
        /// </summary>
        /// <param name="msg"></param>
        public virtual void ReciveMessage(Message msg)
        {
            if (openMsgLog)
                Debug.Log("接收到消息:" + msg.ToString());

            if (msg.userData != null && msg.userData.ContainsKey(failCode))
            {
                //有错误信息的消息,处理错误
                SendFail(msg);
                MessagePool.PutMessage(ref msg);
            }
            else
            {
                //正常收包
                AddMessage(msg);
            }
        }

        /// <summary>
        /// 未被路由的消息 事件过来的
        /// </summary>
        /// <param name="msg">消息</param>
        protected void NotRoutingOk(Message msg)
        {
            return;
            DoFail(msg, 503);
        }

        /// <summary>
        /// 分发处理失败的消息
        /// </summary>
        /// <param name="msg"></param>
        /// <param name="FailCode"></param>
        protected void DoFail(Message msg, int FailCode)
        {
            try
            {
                if (!msg.userData.ContainsKey(failTagKey)) //没有标记不处理
                    return;
                if (msg.userData.ContainsKey(failCode))
                {
                    //已经有错误标记,说明已经进入了此流程-并投递给别人使用,当时又出了问题避免死循环不再处理，记录日志
                    Debug.Error("消息发送失败,并未被处理.failCode:{0},msg:{1}", msg.userData[failCode], msg);
                    return;
                }

                //先赋值失败原因 未发送成功
                msg.userData[failCode] = FailCode;

                bool isMyFail = false;

                if (msg.userData.ContainsKey(doFailKey))
                {
                    //没赋值处理失败的人就是自己处理
                    object tmp = msg.userData[doFailKey];
                    int id = (int)tmp;
                    if (id == myUtile.id)
                    {
                        //是本模块处理的

                        if (msg.userData.ContainsKey(doFailNodeNum))
                        {
                            //如果指定了处理的节点
                            int nodenum = (int)msg.userData[doFailNodeNum];
                            if (nodenum == myUtile.id)
                                isMyFail = true;
                        }
                        else
                            isMyFail = true;

                        if (isMyFail) //处理发送失败的错误
                            SendFail(msg);
                    }
                }

                if (!isMyFail)
                {
                    //投递给处理失败的模块队列
                    //待补待补

                    //处理错误模块的id
                    int nodeid = (int)msg.userData[doFailKey];

                    bool isNAck = false;

                    if (msg.userData.ContainsKey(doFailNodeNum))
                    {
                        //指定了节点-投递给指定节点
                        int nodenum = (int)msg.userData[doFailNodeNum];
                        msg.SendResult = null;
                        if (!DeliveryOne_Result(new ServerNode(nodeid, nodenum), msg, false))
                        {
                            //发不出去就拒收,下次再处理
                            isNAck = true;
                        }
                    }
                    else
                    {
                        //没指定节点,投递给任意节点
                        msg.SendResult = null;
                        if (!DeliveryEveryOne_Result(nodeid, msg, false))
                        {
                            isNAck = true;
                        }
                    }

                    if (!msg.noAck && msg.deliveryTag > 0)
                    {
                        if (isNAck)
                            msg.NAck();
                        else
                            msg.Ack();
                    }
                }


                Debug.Error("发送消息失败:" + msg.ToString());
            }
            catch (Exception e)
            {
                Debug.Error("分发处理任务错误时失败:" + e.ToString());
            }
        }

        /// <summary>
        /// 发送结果 事件过来的
        /// </summary>
        /// <param name="msg">消息</param>
        protected void SendResult(Message msg)
        {
            if (msg.success)
            {
                //发送成功
                if (msg.userData == null || !msg.userData.ContainsKey(successTagKey))
                {
                    //没有标记 不处理
                    return;
                }

                SendOK(msg);
            }
            else
            {
                //发送失败
                DoFail(msg, 502);
            }
        }

        /// <summary>
        /// 单独的消费者
        /// </summary>
        private Consumer exclusiveConsumer;

        /// <summary>
        /// 共享的消费者
        /// </summary>
        private Consumer sharedConsumer;

        /// <summary>
        /// 处理错误消息的消费者
        /// </summary>
        private Consumer doFailConsumer;

        /// <summary>
        /// 过期的消息队列
        /// </summary>
        private const string exipreQueue = "expired";

        /// <summary>
        /// 创建消费者
        /// </summary>
        public virtual void CreateConsumer()
        {
            //消费者有2 独有消息与共有消息
            Consumer consumer = new Consumer();
            consumer.noAck = false;
            consumer.noLoacl = false;
            consumer.queue = RountingManager.instance.queueNode;
            consumer.exclusive = true;
            consumer.Receive = ReciveMessage;
            exclusiveConsumer = consumer;
            rabbit.AddConsumer(consumer);

            consumer = new Consumer();
            consumer.noAck = false;
            consumer.noLoacl = false;
            consumer.queue = RountingManager.instance.queueEveryOne;
            consumer.exclusive = false;
            consumer.Receive = ReciveMessage;
            //consumer.prefetchCount = 3;
            sharedConsumer = consumer;
            rabbit.AddConsumer(sharedConsumer);
        }

        /// <summary>
        /// 添加错误消息的消费者
        /// </summary>
        public void AddFailConsumer()
        {
            return;
            Consumer consumer = new Consumer();
            consumer.noAck = false;
            consumer.noLoacl = false;
            consumer.queue = exipreQueue;
            consumer.exclusive = false;
            consumer.Receive = NotRoutingOk;
            doFailConsumer = consumer;
            rabbit.AddConsumer(doFailConsumer);
        }

        /// <summary>
        /// 移出处理失败消息的消费者
        /// </summary>
        public void RemoveFailConsumer()
        {
            if (doFailConsumer != null)
                rabbit.RemoveConsumer(doFailConsumer);
            doFailConsumer = null;
        }

        /// <summary>
        /// 关闭模块
        /// </summary>
        public void Close()
        {
            if (rabbit == null)
                return;
            rabbit.Close();
        }
        
        /// <summary>
        /// 等待处理的消息
        /// </summary>
        private readonly ConcurrentQueue<WaitBeMsg> _waitBeMessages = new ConcurrentQueue<WaitBeMsg>();

        public int GteMessageCnt()
        {
            return _waitBeMessages.Count;
        }

        /// <summary>
        /// 拿一条消息
        /// </summary>
        /// <returns></returns>
        public WaitBeMsg GetMessage()
        {
            if (_waitBeMessages.Count > 0 && _waitBeMessages.TryDequeue(out WaitBeMsg msg))
            {
                return msg;
            }

            return null;
        }

        /// <summary>
        /// 添加消息
        /// </summary>
        /// <param name="msg"></param>
        public void AddMessage(Message msg)
        {
            WaitBeMsg tmp = WaitBeMsgPool.GetWaiteBeMsg(msg);
            _waitBeMessages.Enqueue(tmp);
        }

        /// <summary>
        /// 添加消息
        /// </summary>
        /// <param name="msg"></param>
        public void AddMessage(WaitBeMsg msg)
        {
            _waitBeMessages.Enqueue(msg);
        }

        /// <summary>
        /// 创建消息
        /// </summary>
        /// <param name="jsonmessage">json整包</param>
        /// <param name="failTag">失败标记</param>
        /// <param name="doFail">谁来处理失败标记</param>
        /// <param name="doFailNode">处理失败标记的节点</param>
        /// <param name="successTag">成功标记</param>
        /// <param name="userData"></param>
        /// <returns></returns>
        public static Message CreateMessage(string jsonmessage, string failTag = null, int doFail = -1,
            int doFailNode = -1, string successTag = null, IDictionary<string, object> userData = null)
        {
            Message msg = MessagePool.GetMessage();
            msg.bodystr = jsonmessage;
            IDictionary<string, object> dic = new Dictionary<string, object>();
            if (doFail != -1)
                dic[doFailKey] = doFail;
            if (failTag != null)
                dic[failTagKey] = failTag;
            if (doFailNode != -1)
                dic[doFailNodeNum] = doFailNode;
            if (successTag != null)
                dic[successTagKey] = successTag;

            msg.userData = dic;
            msg.mandatory = true;
            msg.userData = userData;
            msg.durable = false;
            return msg;
        }

        /// <summary>
        /// 创建消息
        /// </summary>
        /// <param name="head">包头</param>
        /// <param name="otherstring">包体字符串</param>
        /// <param name="failTag">发送失败标记</param>
        /// <param name="doFail">处理失败的模块</param>
        /// <param name="doFailNode">处理失败的节点</param>
        /// <param name="successTag">处理成功的标记</param>
        /// <param name="userData"></param>
        /// <returns>消息</returns>
        public static Message CreateMessage(PackHead head, string otherstring, string failTag = null, int doFail = -1,
            int doFailNode = -1, string successTag = null, IDictionary<string, object> userData = null)
        {
            head.sendutil = GlobalMQ.instance.myUtile;
            string message = JsonPack.GetJsonByPack_head(head, otherstring);
            return CreateMessage(message, failTag, doFail, doFailNode, successTag, userData);
        }

        /// <summary>
        /// 创建消息
        /// </summary>
        /// <param name="head">包头</param>
        /// <param name="data">包数据</param>
        /// <param name="failTag">发送失败的标记</param>
        /// <param name="doFail">处理失败的模块</param>
        /// <param name="doFailNode">处理失败的节点</param>
        /// <param name="successTag">处理成功的标记</param>
        /// <param name="userData"></param>
        /// <returns></returns>
        public static Message CreateMessage(PackHead head, object data = null, string failTag = null, int doFail = -1,
            int doFailNode = -1, string successTag = null, IDictionary<string, object> userData = null)
        {
            head.sendutil = GlobalMQ.instance.myUtile;
            string message = JsonPack.GetJsonByPack(head, data);
            var msg = CreateMessage(message, failTag, doFail, doFailNode, successTag, userData);
            return msg;
        }

        #region 投递消息 message

        /// <summary>
        /// 投递消息给一个模块 并需要知道投递结果-只能知道是否投递成功,无法知道是否到达队列
        /// </summary>
        /// <param name="target">目标</param>
        /// <param name="msg">消息</param>
        /// <param name="isAutoFree">是否自动释放</param>
        /// <param name="durable">持久化</param>
        /// <returns>true 表示在线并投递成功 false 表示投递失败或者目标不在线</returns>
        public bool DeliveryOne_Result(ServerNode target, Message msg, bool isAutoFree, bool durable = false)
        {
            if (msg == null)
                return false;

            if (target == null)
            {
                Debug.Warning("target is null msg:{0}", msg);
                if (isAutoFree)
                    MessagePool.PutMessage(ref msg);
                return false;
            }

            if (myUtile.Equals(target))
            {
                //发给自己的
                GlobalMQ.instance.AddMessage(msg);
                return true;
            }

            msg.exchange = RountingManager.instance.GetClusterRoute(target.id);
            msg.routingkey = RountingManager.instance.GetNodeRouteKey(target.id, target.nodeNum);
            msg.durable = durable;

            if (string.IsNullOrEmpty(msg.exchange) || string.IsNullOrEmpty(msg.routingkey))
            {
                Debug.Error("消息的交换机或路由键为空6666:" + msg.bodystr);
                return false;
            }


            //return CheckFiltersAndPublish(rabbit.BasicPublish_Result, msg, isAutoFree, Recv2ReturnDescription.One_Bool);//
            return rabbit.BasicPublish_Result(msg, isAutoFree);
        }

        /// <summary>
        /// 投递消息给一个模块 不管结果
        /// </summary>
        /// <param name="target"></param>
        /// <param name="msg"></param>
        /// <param name="isAutoFree">是否自动释放</param>
        /// <param name="durable">持久化</param>
        public virtual void DeliveryOne(ServerNode target, Message msg, bool isAutoFree, bool durable = false)
        {
            if (target == null)
            {
                Debug.Warning("target is null msg:{0}", msg);
                return;
            }

            if (myUtile.Equals(target))
            {
                //发给自己的
                GlobalMQ.instance.AddMessage(msg);
                return;
            }

            msg.exchange = RountingManager.instance.GetClusterRoute(target.id);
            msg.routingkey = RountingManager.instance.GetNodeRouteKey(target.id, target.nodeNum);
            msg.durable = durable;

            if (string.IsNullOrEmpty(msg.exchange) || string.IsNullOrEmpty(msg.routingkey))
            {
                Debug.Error("消息的交换机或路由键为空5555:" + msg.bodystr);
                return;
            }

            //CheckFiltersAndPublishNoReturn(rabbit.BasicPublish, msg, isAutoFree, Recv2ReturnDescription.One_Void);
            rabbit.BasicPublish(msg, isAutoFree);
        }


        /// <summary>
        /// 群发给任意模块信息
        /// </summary>
        /// <param name="moduleId"></param>
        /// <param name="msg"></param>
        /// <param name="isAutoFree">是否自动释放</param>
        /// <param name="durable">持久化</param>
        /// <returns></returns>
        public bool DeliveryEveryOne_Result(int moduleId, Message msg, bool isAutoFree, bool durable = false)
        {
            msg.exchange = RountingManager.instance.GetClusterRoute(moduleId);
            msg.routingkey = RountingManager.instance.GetEveryOneRouteKey(moduleId);
            msg.durable = durable;

            if (string.IsNullOrEmpty(msg.exchange) || string.IsNullOrEmpty(msg.routingkey))
            {
                Debug.Error("消息的交换机或路由键为空444:" + moduleId + "," + msg.bodystr);
                return false;
            }


            //return CheckFiltersAndPublish(rabbit.BasicPublish_Result, msg, isAutoFree, Recv2ReturnDescription.EveryOne_Bool);
            return rabbit.BasicPublish_Result(msg, isAutoFree);
        }

        /// <summary>
        /// 群发给任意模块信息
        /// </summary>
        /// <param name="moduleId">目标模块id</param>
        /// <param name="msg">消息内容</param>
        /// <returns>没有绑定该模块则为false 绑定则为true</returns>
        /// <param name="isAutoFree">是否自动释放</param>
        /// <param name="durable">持久化</param>
        public void DeliveryEveryOne(int moduleId, Message msg, bool isAutoFree, bool durable = false)
        {
            msg.exchange = RountingManager.instance.GetClusterRoute(moduleId);
            msg.routingkey = RountingManager.instance.GetEveryOneRouteKey(moduleId);
            msg.durable = durable;

            if (string.IsNullOrEmpty(msg.exchange) || string.IsNullOrEmpty(msg.routingkey))
            {
                Debug.Error("消息的交换机或路由键为空333:" + msg.bodystr);
                return;
            }

            //CheckFiltersAndPublishNoReturn(rabbit.BasicPublish, msg, isAutoFree, Recv2ReturnDescription.EveryOne_Void);
            rabbit.BasicPublish(msg, isAutoFree);
        }

        /// <summary>
        /// 投递这个模块的所有节点
        /// </summary>
        /// <param name="moduleId"></param>
        /// <param name="msg"></param>
        /// <param name="isAutoFree">是否自动释放</param>
        /// <param name="durable">持久化</param>
        public bool DeliveryAll_Result(int moduleId, Message msg, bool isAutoFree, bool durable = false)
        {
            msg.exchange = RountingManager.instance.GetClusterRoute(moduleId);
            msg.routingkey = RountingManager.instance.routeallkey;
            msg.durable = durable;

            if (string.IsNullOrEmpty(msg.exchange) || string.IsNullOrEmpty(msg.routingkey))
            {
                Debug.Error("消息的交换机与路由键为空!111" + msg.bodystr);
                return false;
            }

            return CheckFiltersAndPublish(rabbit.BasicPublish_Result, msg, isAutoFree,
                Recv2ReturnDescription.All_Bool); //rabbit.BasicPublish_Result(msg, isAutoFree);
        }

        /// <summary>
        /// 投递给这类所有模块
        /// </summary>
        /// <param name="moduleId">目标类模块id</param>
        /// <param name="msg">消息内容</param>
        /// <param name="isAutoFree">是否自动释放</param>
        /// <param name="durable">持久化</param>
        public void DeliveryAll(int moduleId, Message msg, bool isAutoFree, bool durable = false)
        {
            msg.exchange = RountingManager.instance.GetClusterRoute(moduleId);
            msg.routingkey = RountingManager.instance.routeallkey;
            msg.durable = durable;

            if (string.IsNullOrEmpty(msg.exchange) || string.IsNullOrEmpty(msg.routingkey))
            {
                Debug.Error("消息的交换机与路由键为空!222:" + msg.bodystr);
                return;
            }

            //CheckFiltersAndPublishNoReturn(rabbit.BasicPublish, msg, isAutoFree, Recv2ReturnDescription.All_Void);
            rabbit.BasicPublish(msg, isAutoFree);
        }

        /// <summary>
        /// 投递给所有模块
        /// </summary>
        /// <param name="msg"></param>
        /// <param name="isAutoFree">是否自动释放</param>
        /// <param name="durable">持久化处理</param>
        public bool DeliverAll_Result(Message msg, bool isAutoFree, bool durable = false)
        {
            msg.exchange = RountingManager.AllServerRouteName;
            msg.routingkey = RountingManager.instance.routeallkey;
            msg.durable = durable;

            if (string.IsNullOrEmpty(msg.exchange) || string.IsNullOrEmpty(msg.routingkey))
            {
                // Debug.Error("消息的交换机或路由键为空!7777" + msg.bodystr);
                return false;
            }

            return CheckFiltersAndPublish(rabbit.BasicPublish_Result, msg, isAutoFree,
                Recv2ReturnDescription.All_Bool); //rabbit.BasicPublish_Result(msg, isAutoFree);
        }

        /// <summary>
        /// 投递给所有模块
        /// </summary>
        /// <param name="msg"></param>
        /// <param name="isAutoFree">是否自动释放</param>
        /// <param name="durable">持久化</param>
        public void DeliveryAll(Message msg, bool isAutoFree, bool durable = false)
        {
            msg.exchange = RountingManager.AllServerRouteName;
            msg.routingkey = RountingManager.instance.routeallkey;
            msg.durable = durable;

            if (string.IsNullOrEmpty(msg.exchange) || string.IsNullOrEmpty(msg.routingkey))
            {
                Debug.Error("消息的交换机或路由键为空!8888:" + msg.bodystr);
                return;
            }

            //CheckFiltersAndPublishNoReturn(rabbit.BasicPublish, msg, true, Recv2ReturnDescription.All_Void);
            rabbit.BasicPublish(msg, isAutoFree);
        }


        /// <summary>
        /// RPC Call 方法
        /// </summary>
        /// <param name="target">目标</param>
        /// <param name="message">消息</param>
        /// <param name="waitTime">等待多久</param>
        /// <returns></returns>
        public string Call(ServerNode target, string message, int waitTime = 3000)
        {
            if (target == null)
            {
                Debug.Warning("target is null msg:{0}", message);
                return null;
            }

            string exchange = RountingManager.instance.GetClusterRoute(target.id);
            string routingkey = RountingManager.instance.GetNodeRouteKey(target.id, target.nodeNum);

            if (string.IsNullOrEmpty(exchange) || string.IsNullOrEmpty(routingkey))
            {
                Debug.Error("消息的交换机或路由键为 call 5555:" + message);
                return null;
            }

            return CheckFiltersAndCall(() => rpcClient.Call(exchange, routingkey, message, waitTime), message,
                Recv2ReturnDescription.Call_One_String);
        }

        /// <summary>
        /// RPC Call 任意一个模块
        /// </summary>
        /// <param name="moduleId">目标id</param>
        /// <param name="message">消息</param>
        /// <param name="waitTime">等待多久</param>
        /// <returns></returns>
        public string Call_EveryOne(int moduleId, string message, int waitTime = 3000)
        {
            string exchange = RountingManager.instance.GetClusterRoute(moduleId);
            string routingkey = RountingManager.instance.GetEveryOneRouteKey(moduleId);

            if (string.IsNullOrEmpty(exchange) || string.IsNullOrEmpty(routingkey))
            {
                Debug.Error("消息交换或路由键为 call 6666:" + moduleId + "," + message);
                return null;
            }

            var ret = CheckFiltersAndCall(() => rpcClient.Call(exchange, routingkey, message, waitTime), message,
                Recv2ReturnDescription.Call_EveryOne_String);
            return ret;
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="target"></param>
        /// <param name="head"></param>
        /// <param name="data"></param>
        /// <param name="waitTime"></param>
        /// <returns></returns>
        public string Call(ServerNode target, PackHead head, object data = null, int waitTime = 3000)
        {
            string message = JsonPack.GetJsonByPack(head, data);
            return Call(target, message, waitTime);
        }

        /// <summary>
        /// Rpc call
        /// </summary>
        /// <param name="target"></param>
        /// <param name="head"></param>
        /// <param name="jsonmessage"></param>
        /// <param name="waitTime"></param>
        /// <returns></returns>
        public string Call(ServerNode target, PackHead head, string jsonmessage, int waitTime = 3000)
        {
            string message = JsonPack.GetJsonByPack_head(head, jsonmessage);
            return Call(target, message, waitTime);
        }

        /// <summary>
        /// call 方法
        /// </summary>
        /// <param name="moduleId"></param>
        /// <param name="head"></param>
        /// <param name="data"></param>
        /// <param name="waitTime"></param>
        /// <returns></returns>
        public string Call_EveryOne(int moduleId, PackHead head, object data = null, int waitTime = 3000)
        {
            string message = JsonPack.GetJsonByPack(head, data);
            return Call_EveryOne(moduleId, message, waitTime);
        }

        /*
        /// <summary>
        ///
        /// </summary>
        /// <param name="moduleId"></param>
        /// <param name="head"></param>
        /// <param name="data"></param>
        /// <param name="waitTime"></param>
        /// <returns></returns>
        public System.Threading.Tasks.Task<string> AsyncCall_EveryOne(int moduleId, ServerHead head, object data = null, int waitTime = 3000) {
            string message = JsonPack.GetJsonByPack(head, data);
            return AsyncCall_EveryOne(moduleId, message, waitTime);
        }
        */


        /*
        /// <summary>
        ///
        /// </summary>
        /// <param name="moduleId"></param>
        /// <param name="message"></param>
        /// <param name="waitTime"></param>
        /// <returns></returns>
        public System.Threading.Tasks.Task<string> AsyncCall_EveryOne(int moduleId, string message, int waitTime = 3000) {
            string exchange = RountingManager.instance.GetClusterRoute(moduleId);
            string routingkey = RountingManager.instance.GetEveryOneRouteKey(moduleId);

            if(string.IsNullOrEmpty(exchange) || string.IsNullOrEmpty(routingkey)) {
                Debug.Error("消息交换或路由键为 call 6666:" + message);
                return null;
            }
            var ret = rpcClient.AsyncCall(exchange, routingkey, message, waitTime);
            return ret;
        }
        */

        /// <summary>
        /// call 方法
        /// </summary>
        /// <param name="moduleId"></param>
        /// <param name="head"></param>
        /// <param name="jsonmessage"></param>
        /// <param name="waitTime"></param>
        /// <returns></returns>
        public string Call_EveryOne_str(int moduleId, PackHead head, string jsonmessage = null, int waitTime = 3000)
        {
            string message = JsonPack.GetJsonByPack_head(head, jsonmessage);
            return Call_EveryOne(moduleId, message, waitTime);
        }

        /// <summary>
        /// RPC回复
        /// </summary>
        /// <param name="ibp"></param>
        /// <param name="message"></param>
        /// <param name="userData"></param>
        public void CallBack(IBasicProperties ibp, object message, IDictionary<string, object> userData)
        {
            Message msg = MessagePool.GetMessage();
            msg.exchange = string.Empty;
            msg.routingkey = ibp.ReplyTo;
            msg.propertis = rabbit.GetBasicProperties();
            msg.propertis.CorrelationId = ibp.CorrelationId;
            msg.bodystr = JsonPack.GetJson(message);
            msg.userData = userData;
            Debug.Log("RPC回复:" + msg.propertis.CorrelationId);

            CheckFiltersCallbackAndPublishNoReturn(rabbit.BasicPublish, msg, true, Recv2ReturnDescription.CallBack);
            //rabbit.BasicPublish(msg, true);
        }

        #endregion


        /// <summary>
        /// AddFilter
        /// </summary>
        /// <param name="filter"></param>
        public static void AddFilter(IMQFilter filter)
        {
            Filters.Add(filter);
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="filter"></param>
        public static void RemoveFilter(IMQFilter filter)
        {
            if (Filters.Contains(filter))
            {
                Filters.Remove(filter);
            }
        }

        /// <summary>
        /// 
        /// </summary>
        public static ICollection<IMQFilter> Filters { get; } =
            new System.Collections.ObjectModel.Collection<IMQFilter>();

        /// <summary>
        /// 
        /// </summary>
        /// <param name="defaultPublisher"></param>
        /// <param name="msg"></param>
        /// <param name="isAutoFree"></param>
        /// <param name="msgDesc"></param>
        protected static void CheckFiltersAndPublishNoReturn(Action<Message, bool> defaultPublisher, Message msg,
            bool isAutoFree, Recv2ReturnDescription msgDesc)
        {
            var filer = Filters.FirstOrDefault(p => p.Filter(msg, msgDesc));
            if (filer != null)
            {
                filer.HandleMessage(msg, msgDesc);
                MessagePool.PutMessage(ref msg);
            }
            else
            {
                defaultPublisher(msg, isAutoFree);
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="defaultPublisher"></param>
        /// <param name="msg"></param>
        /// <param name="isAutoFree"></param>
        /// <param name="msgDesc"></param>
        protected static void CheckFiltersCallbackAndPublishNoReturn(Action<Message, bool> defaultPublisher,
            Message msg, bool isAutoFree, Recv2ReturnDescription msgDesc)
        {
            var filer = Filters.FirstOrDefault(p => p.FilterCallback(msg));
            if (filer != null)
            {
                filer.HandleCallbackMessage(msg);
                MessagePool.PutMessage(ref msg);
            }
            else
            {
                defaultPublisher(msg, isAutoFree);
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="defaultPublisher"></param>
        /// <param name="msg"></param>
        /// <param name="isAutoFree"></param>
        /// <param name="msgDesc"></param>
        /// <returns></returns>
        protected static T CheckFiltersAndPublish<T>(Func<Message, bool, T> defaultPublisher, Message msg,
            bool isAutoFree, Recv2ReturnDescription msgDesc)
        {
            var filer = Filters.FirstOrDefault(p => p.Filter(msg, msgDesc));
            if (filer != null)
            {
                var ret = filer.HandleMessage(msg, msgDesc);
                if (ret is T value)
                {
                    MessagePool.PutMessage(ref msg);
                    return value;
                }

                throw new Exception(
                    $"消息过滤器: {filer.Name} 返回值类型无效, 期待类型:{typeof(T)}, 当前类型: {ret?.GetType()}, returnResult: {msgDesc}");
            }
            else
            {
                return defaultPublisher(msg, isAutoFree);
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="defaultPublisher"></param>
        /// <param name="msg"></param>
        /// <param name="msgDesc"></param>
        /// <returns></returns>
        protected static T CheckFiltersAndCall<T>(Func<T> defaultPublisher, string msg, Recv2ReturnDescription msgDesc)
        {
            var filer = Filters.FirstOrDefault(p => p.CallFilter(msg, msgDesc));
            if (filer != null)
            {
                var ret = filer.HandleCallMessage(msg, msgDesc);
                if (ret is T value)
                {
                    return value;
                }

                throw new Exception(
                    $"消息过滤器: {filer.Name} 返回值类型无效, 期待类型:{typeof(T)}, 当前类型: {ret?.GetType()}, returnResult: {msgDesc}");
            }
            else
            {
                return defaultPublisher();
            }
        }
    }

    /// <summary>
    /// 外发到rabbit消息的过滤器
    /// </summary>
    public interface IMQFilter
    {
        /// <summary>
        /// 过滤器名称
        /// </summary>
        string Name { get; }

        /// <summary>
        /// 检查此消息是否符合过滤条件
        /// </summary>
        /// <param name="message"></param>
        /// <param name="msgDesc"></param>
        /// <returns></returns>
        bool Filter(Message message, Recv2ReturnDescription msgDesc);

        /// <summary>
        /// FilterCallback
        /// </summary>
        /// <param name="message"></param>
        /// <returns></returns>
        bool FilterCallback(Message message);

        /// <summary>
        /// 
        /// </summary>
        /// <param name="message"></param>
        /// <returns></returns>
        void HandleCallbackMessage(Message message);

        /// <summary>
        /// 处理此消息
        /// </summary>
        /// <param name="message"></param>
        /// <param name="msgDesc"></param>
        /// <returns></returns>
        object HandleMessage(Message message, Recv2ReturnDescription msgDesc);

        /// <summary>
        /// 检查RPC Call 消息是否符合过滤条件
        /// </summary>
        /// <param name="msg"></param>
        /// <param name="msgDesc"></param>
        /// <returns></returns>
        bool CallFilter(string msg, Recv2ReturnDescription msgDesc);

        /// <summary>
        /// 处理 RPC Call 消息
        /// </summary>
        /// <param name="msg"></param>
        /// <param name="msgDesc"></param>
        /// <returns></returns>
        object HandleCallMessage(string msg, Recv2ReturnDescription msgDesc);
    }

    /// <summary>
    /// ReturnResult
    /// </summary>
    public enum Recv2ReturnDescription : byte
    {
        /// <summary>
        /// EveryOne, 无返回 
        /// </summary>
        EveryOne_Void = 1,

        /// <summary>
        /// EveryOne, 返回 bool
        /// </summary>
        EveryOne_Bool = 2,

        /// <summary>
        /// EveryOne, 返回 string
        /// </summary>
        EveryOne_String = 3,

        /// <summary>
        /// CallBack
        /// </summary>
        CallBack = 10,

        /// <summary>
        /// All, 无返回 
        /// </summary>
        All_Void = 20,

        /// <summary>
        /// All, 返回 bool
        /// </summary>
        All_Bool = 21,

        /// <summary>
        /// One 无返回 
        /// </summary>
        One_Void = 30,

        /// <summary>
        /// One, 返回 bool
        /// </summary>
        One_Bool = 31,

        /// <summary>
        /// Call EveryOne, 返回 string
        /// </summary>
        Call_EveryOne_String = 40,

        /// <summary>
        /// Call One, 返回 string
        /// </summary>
        Call_One_String = 41,
    }
}