public class FileUploadedMq : IActiveMq

    {

        private static string ConnString;

        public static ConnectionFactory Factory { get; private set; }

        //private static IMessageProducer MsgProducer;

        public static event Action<FileUploadedModel> Recved;

        static FileUploadedMq()

        {

            ConnString = dbo.GetDbConnString(MqTypeEnum.FileUpload.ToString());

            Factory = new ConnectionFactory(ConnString);

        }

        public void Send(FileUploadedModel Msg)

        {

            try

            {

                using (IConnection connection = Factory.CreateConnection())

                {

                    //通过连接创建Session会话

                    using (ISession session = connection.CreateSession())

                    {

                        //通过会话创建生产者，方法里面new出来的是MQ中的Queue

                        var MsgProducer = session.CreateProducer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("FileUploaded"));

                        //创建一个发送的消息对象

                        var TxtMsg = MsgProducer.CreateTextMessage();

                        TxtMsg.Properties.SetString("filter", "FileUploaded");

                        //给这个对象赋实际的消息

                        TxtMsg.Text = Msg.ToJson();

                        //设置消息对象的属性，这个很重要哦，是Queue的过滤条件，也是P2P消息的唯一指定属性

                        //生产者把消息发送出去，几个枚举参数MsgDeliveryMode是否长链，MsgPriority消息优先级别，发送最小单位，当然还有其他重载

                        MsgProducer.Send(TxtMsg, MsgDeliveryMode.NonPersistent, MsgPriority.Normal, TimeSpan.MinValue);

                    }

                }

            }

            catch (Exception ex)

            {

                InfoTypeEnum.Error.LogTo(ex.Message);

            }

        }

    }

    public class ActiveMq

    {

        public static FileUploadedMq FileUploaded = new FileUploadedMq();

        public static FileResolvedMq FileResolved = new FileResolvedMq();

        public static FileFastDfsSavedMq FileFastDfsSaved = new FileFastDfsSavedMq();

        // public static EmailUploadMq EmailUpload=new EmailUploadMq();

        public static bool RegisteFileUploaded(Action<FileUploadedModel> msgRecv, Action<FileUploadedModel, Exception> ErrorFunc = null)

        {

            return RegisteMq(MqTypeEnum.FileUpload, o =>

            {

                if (msgRecv == null) return;

                msgRecv(o.FromJson<FileUploadedModel>());

            }, (a, b) =>

            {

                if (ErrorFunc == null) return;

                ErrorFunc(a.FromJson<FileUploadedModel>(), b);

            });

        }

        public static bool RegisteFileResolved(Action<FileResolvedModel> msgRecv, Action<FileResolvedModel, Exception> ErrorFunc = null)

        {

            return RegisteMq(MqTypeEnum.FileResolved, o =>

            {

                if (msgRecv == null) return;

                msgRecv(o.FromJson<FileResolvedModel>());

            }, (a, b) =>

            {

                if (ErrorFunc == null) return;

                ErrorFunc(a.FromJson<FileResolvedModel>(), b);

            });

        }

        public static bool RegisteFileFastDfsSaved(Action<FileFastDfsSavedModel> msgRecv, Action<FileFastDfsSavedModel, Exception> ErrorFunc = null)

        {

            return RegisteMq(MqTypeEnum.FileFastDfsSaved, o =>

            {

                if (msgRecv == null) return;

                msgRecv(o.FromJson<FileFastDfsSavedModel>());

            }, (a, b) =>

            {

                if (ErrorFunc == null) return;

                ErrorFunc(a.FromJson<FileFastDfsSavedModel>(), b);

            });

        }

        private static bool RegisteMq(MqTypeEnum Name, Action<string> msgRecv, Action<string, Exception> ErrorFunc)

        {

            if (msgRecv == null) return false;

            var QueueName = Name.ToString();

            try

            {

                var factory = FileResolvedMq.Factory;

                //通过工厂构建连接

                IConnection connection = factory.CreateConnection();

                //这个是连接的客户端名称标识

                //connection.ClientId = Environment.MachineName + "FileResolvedMqListener";

                //启动连接，监听的话要主动启动连接

                connection.Start();

                //通过连接创建一个会话

                ISession session = connection.CreateSession();

                //通过会话创建一个消费者，这里就是Queue这种会话类型的监听参数设置

                IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue(QueueName));//, "filter='FileResolved'");

                session.DeleteDestination(QueueName);

                //注册监听事件

                consumer.Listener += message =>

                {

                    var txt = (ITextMessage)message;

                    if (txt == null || txt.Text.HasValue() == false) { return; }

                    var model = txt.Text;

                    try

                    {

                        msgRecv.Invoke(model);

                    }

                    catch (Exception e)

                    {

                        InfoTypeEnum.Error.LogTo(QueueName + ":在处理过程出现错误!" + connection.ClientId, model.ToJson(), e.Message);

                        if (ErrorFunc != null)

                        {

                            try

                            {

                                ErrorFunc(model, e);

                            }

                            catch { }

                        }

                    }

                };

                return true;

            }

            catch (Exception ex)

            {

                InfoTypeEnum.Error.LogTo(QueueName + ":" + ex.Message);

                return false;

            }

        }

    }