SignalR与ActiveMQ结合构建实时通信

一、概述

本教程主要阐释了如何利用SignalR与消息队列的结合，实现不同客户端的交互

• SignalR如何和消息队列交互（暂使用ActiveMQ消息队列）
• SignalR寄宿在web中和其他SignalR、控制台客户端交互。
• SignalR单独寄宿在控制台中和其他SignalR、控制台客户端交互。

下面屏幕截图展示了各个客户端通过ActiveMQ相互通信

　　1、SignalR寄宿在web：

　　2、SignalR寄宿在控制台中，web客户端调用SignalR，读者自行测试。

 工程目录：

一、创建项目

　　1、创建生产者项目，该项目要是通过控制台输入消息，发送到消息队列

　　　　创建控制台应用程序命名为ActiveMQNetProcucer，然后用包管理器安装ActiveMQ的.Net客户端

　　　　Install-Package Apache.NMS.ActiveMQ

　　　　主要代码如下：

using Apache.NMS;
using Apache.NMS.ActiveMQ;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace ActiveMQNet
{
    class Program
    {
        static IConnectionFactory _factory = null;
        static IConnection _connection = null;
        static ITextMessage _message = null;

        static void Main(string[] args)
        {
            //创建工厂
            _factory = new ConnectionFactory("tcp://127.0.0.1:61616/");

            try
            {
                //创建连接
                using (_connection = _factory.CreateConnection())
                {
                    //创建会话
                    using (ISession session = _connection.CreateSession())
                    {
                        //创建一个主题
                        IDestination destination = new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("topic");

                        //创建生产者
                        IMessageProducer producer = session.CreateProducer(destination);

                        Console.WriteLine("Please enter any key to continue! ");
                        Console.ReadKey();
                        Console.WriteLine("Sending: ");

                        //创建一个文本消息
                        _message = producer.CreateTextMessage("Hello AcitveMQ....");

                        //发送消息
                        producer.Send(_message, MsgDeliveryMode.NonPersistent, MsgPriority.Normal, TimeSpan.MinValue);
                        while (true)
                        {
                            var msg = Console.ReadLine();
                            _message = producer.CreateTextMessage(msg);
                            producer.Send(_message, MsgDeliveryMode.NonPersistent, MsgPriority.Normal, TimeSpan.MinValue);
                        }
                       
                    }
                }

            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.ToString());
            }

            Console.ReadLine();

        }
    }
}

 　　2、创建消费者项目，该项目主要是订阅消息队列中的消息　　

　　　　创建控制台应用程序命名为ActiveMQNetCustomer，然后用包管理器安装ActiveMQ的.Net客户端

　　　　Install-Package Apache.NMS.ActiveMQ

　　　　主要代码：

using Apache.NMS;
using Apache.NMS.ActiveMQ;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace ActiveMQNetCustomer
{
    class Program
    {
        static IConnectionFactory _factory = null;

        static void Main(string[] args)
        {
            try
            {
                //创建连接工厂
                _factory = new ConnectionFactory("tcp://127.0.0.1:61616/");
                //创建连接
                using (IConnection conn = _factory.CreateConnection())
                {
                    //设置客户端ID
                   // conn.ClientId = "Customer";
                    conn.Start();
                    //创建会话
                    using (ISession session = conn.CreateSession())
                    {
                        //创建主题
                        var topic = new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("topic");

                        //创建消费者
                        IMessageConsumer consumer = session.CreateDurableConsumer(topic, "Customer", null, false);

                        //注册监听事件
                        consumer.Listener += new MessageListener(consumer_Listener);

                        //这句代码非常重要，
                        //这里没有read方法，Session会话会被关闭，那么消费者将监听不到生产者的消息
                        Console.Read();
                    }

                    //关闭连接
                    conn.Stop();
                    conn.Close();
                }

            }
            catch (Exception ex)
            {
                Console.Write(ex.ToString());
            }

        }

        /// <summary>
        /// 消费监听事件
        /// </summary>
        /// <param name="message"></param>
        static void consumer_Listener(IMessage message)
        {
            ITextMessage msg = (ITextMessage)message;
            Console.WriteLine("Receive: " + msg.Text);
        }
    }
}

 　　3、创建包装ActiveMQ生产者和消费者项目，供SignalR.ActiveMQ.WebHost项目使用，来发布消息和订阅消息

　　　　创建类库项目Signalr.ActiveMQ，然后用包管理器安装ActiveMQ的.Net客户端

　　　　Install-Package Apache.NMS.ActiveMQ

　　　　主要代码；

　　　　生产者类：创建单实例生产者对象调用Send发放，发送消息到ActiveMQ消息队列　　　　

using Apache.NMS;
using Apache.NMS.ActiveMQ;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Signalr.ActiveMQ
{
  public  class Procucer
    {
        private IMessageProducer producer;
        private static Procucer instance=null;
        private Procucer(string customerId,string address)
        {
            instance = this;
            //创建工厂
            IConnectionFactory _factory = new ConnectionFactory("tcp://127.0.0.1:61616/");

            try
            {
                //创建连接
                IConnection _connection = _factory.CreateConnection();
                {
                    //创建会话
                    ISession session = _connection.CreateSession();
                    {
                        //创建一个主题
                        IDestination destination = new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("topic");

                        //创建生产者
                        producer = session.CreateProducer(destination);

                        Console.WriteLine("Please enter any key to continue! ");
                      //  Console.ReadKey();
                        Console.WriteLine("Sending: ");                      

                    }
                }

            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.ToString());
            }

            //Console.ReadLine();
        }

        public static Procucer GetInstance(string customerId="",string address= "tcp://127.0.0.1:61616/")
        {
            if (instance == null)
                instance = new Procucer(customerId, address);
            return instance;
        }

        public void Send(string msg)
        {
            //创建一个文本消息
            ITextMessage _message = producer.CreateTextMessage(msg);
            //发送消息
            producer.Send(_message, MsgDeliveryMode.NonPersistent, MsgPriority.Normal, TimeSpan.MinValue);
        }
    }
}

 　　　 消费者类：启用单独的线程监听消息队列中的消息，当监听到消息后 广播给所有的 SinglaR客户端，其中静态属性Clients保存了所有的SinglaR客户端，当SinglaR客户端连接或者断开的时候会更新Clients属性详细代码在SignalR.ActiveMQ.WebHost中的　MyHub文件中。为了阻止当前线程退出调用了 System.Threading.Thread.CurrentThread.Join();阻塞当前线程，避免当web中方法执行完毕后对象被回收，起不到监听消息队列的作用。

using Apache.NMS;
using Apache.NMS.ActiveMQ;
using Microsoft.AspNet.SignalR.Hubs;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Web;

namespace SignalR.ActiveMQ
{
    public class Customer
    {
        private static object lockObj = new object();
        private static IHubCallerConnectionContext<dynamic> _clients;
        public static IHubCallerConnectionContext<dynamic> Clients
        {
            get { return _clients; }
            set
            {
                lock (lockObj)
                {
                    _clients = value;
                }
            }
        }
        public static void Run(string cutomerId="",string address= "tcp://127.0.0.1:61616/")
        {
            System.Threading.Thread t = new System.Threading.Thread(() =>
            {
                try
                {
                    //创建连接工厂
                    IConnectionFactory _factory = new ConnectionFactory(address);
                    //创建连接
                    using (IConnection conn = _factory.CreateConnection())
                    {
                        //设置客户端ID
                        conn.ClientId = cutomerId;
                        conn.Start();
                        //创建会话
                        using (ISession session = conn.CreateSession())
                        {
                            //创建主题
                            var topic = new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("topic");

                            //创建消费者
                            IMessageConsumer consumer = session.CreateDurableConsumer(topic, "Customer", null, false);

                            //注册监听事件
                            consumer.Listener += new MessageListener(consumer_Listener);

                            //阻塞当前线程，监听消息
                            System.Threading.Thread.CurrentThread.Join();
                        }
                        //关闭连接
                        conn.Stop();
                        conn.Close();
                    }

                }
                catch (Exception ex)
                {
                    Debug.WriteLine(ex.ToString());
                    Console.WriteLine(ex.ToString());
                }
               
            });

            t.Start();
        }
        static void consumer_Listener(IMessage message)
        {
            ITextMessage msg = (ITextMessage)message;
            if (Clients != null)
            {
                Clients.All.broadcastMessage(msg.Text);
            }
            Debug.WriteLine("Receive: " + msg.Text);
            Console.WriteLine("Receive: " + msg.Text);
        }
    }
}

 　　4、创建web自宿主的SignalR项目，该项目既发布消息，也订阅消息

　　　　创建MVC项目SignalR.ActiveMQ.WebHost，然后用包管理器安装ActiveMQ的.Net客户端

　　　　Install-Package Apache.NMS.ActiveMQ

　　　　创建SignalR的hub：当有客户端连接或者断开的时候更新Customer.Clients 静态属性，保存所有的SignalR客户端。

　　　　web端通过调用代理的Send方法发送消息到消息队列。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Web;
using Microsoft.AspNet.SignalR;
using Signalr.ActiveMQ;
using System.Threading.Tasks;

namespace SignalR.ActiveMQ.Sample.Signal.Class
{   
    public class chatHub : Hub
    {
        public void Send(string clientName, string message)
        {
            Procucer.GetInstance().Send(message);            
        }
        public override Task OnConnected()
        {
            Customer.Clients = this.Clients;
            return base.OnConnected();
        }

        public override Task OnDisconnected(bool stopCalled)
        {
            Customer.Clients = this.Clients;
            return base.OnDisconnected(stopCalled);
        }
    }
}

　　　　Startup类中启动消费者监听线程，调用的项目Signalr.ActiveMQ中的Customer.Run()方法：

using Microsoft.AspNet.SignalR;
using Microsoft.Owin;
using Owin;
using SignalR.ActiveMQ;

[assembly: OwinStartupAttribute(typeof(SignalR.ActiveMQ.Sample.Startup))]
namespace SignalR.ActiveMQ.Sample
{
    public partial class Startup
    {
        public void Configuration(IAppBuilder app)
        {          
            app.MapSignalR();
         
            Customer.Run();//启动消费者监听线程
        }
    }
}

 二、启动顺序：

1、启动ActiveMQ程序 可参考  http://www.cnblogs.com/xwdreamer/archive/2012/02/21/2360818.html

2、启动ActiveMQNetProcucer项目

3、ActiveMQNetCustomer项目

4、启动SignalR.ActiveMQ.WebHost，开多个浏览器窗口，模拟多个SignalR客户端 

 三、SignalR宿主和web客户端分离两个项目 

Signalr.ActiveMQ.SelfHost 用控制台寄宿SignalR提供的服务供Signalr.ActiveMQ.Web使用

Signalr.ActiveMQ.Web 通过chart.html调用Signalr.ActiveMQ.SelfHost的服务 

Signalr.ActiveMQ.SelfHost 和SignalR.ActiveMQ.WebHost不能同时启动，现在两个项目绑定到了同一个端口。

四、测试

　　在生产者窗口中输入消息回车，观察其他客户端的变化

     在Singlar的web客户端发送消息，观察其他客户端的变化

源代码：https://github.com/zhaoyingju/SignalrActiveMQ.git