仿微信 即时聊天工具 - SignalR （一）

using microsoft.aspnet.signalr;
using microsoft.aspnet.signalr.hubs;
using system;
using system.collections;
using system.collections.generic;
using system.linq;
using system.threading;
using system.threading.tasks;
using system.web;

namespace signalr.hubs
{
    [hubname("messagehub")]
    public class messagehub : hub
        {
            private readonly chatticker ticker;
            public messagehub()
            {
                ticker = chatticker.instance;
            }

            public void register(string username, string group = "default")
            {
                var list = (list<siginalrmodel>)httpruntime.cache.get("msg_hs");
                if (list == null)
                {
                    list = new list<siginalrmodel>();
                }
               

                if (list.any(x => x.connectionid == context.connectionid))
                {
                    clients.client(context.connectionid).broadcastmessage("已经注册，无需再次注册");
                }
            else if (list.any(x => x.name == username))
            {
                var model = list.where(x => x.name == username && x.group == group).firstordefault();
                if (model != null)
                {
                    //注册到全局  
                    ticker.globalcontext.groups.add(context.connectionid, group);
                    clients.client(model.connectionid).exit();
                    ticker.globalcontext.groups.remove(model.connectionid, group);
                    list.remove(model);
                    model.connectionid = context.connectionid;
                    list.add(model);
                    clients.group(group).removeuserlist(model.connectionid);
                    thread.sleep(200);
                    var gourplist = list.where(x => x.group == group).tolist();
                    clients.group(group).appenduserlist(context.connectionid, gourplist);
                    httpruntime.cache.insert("msg_hs", list);
                    // clients.client(model.connectionid).broadcastmessage("名称重复，只能注册一个");
                }
                //clients.client(context.connectionid).broadcastmessage("名称重复，只能注册一个");
            }
            else
                {
                    list.add(new siginalrmodel() { name = username, group = group, connectionid = context.connectionid });

                    //注册到全局  
                    ticker.globalcontext.groups.add(context.connectionid, group);
                    thread.sleep(200);

                    var gourplist = list.where(x => x.group == group).tolist();
                    clients.group(group).appenduserlist(context.connectionid, gourplist);
                    httpruntime.cache.insert("msg_hs", list);
                }

            }

            public void say(string msg)
            {
                var list = (list<siginalrmodel>)httpruntime.cache.get("msg_hs");
                if (list == null)
                {
                    list = new list<siginalrmodel>();
                }
                var usermodel = list.where(x => x.connectionid == context.connectionid).firstordefault();
                if (usermodel != null )
                {
                    clients.group(usermodel.group).say(usermodel.name, msg);
                }
            }

        public void exit()
        {
            ondisconnected(true);
        }

        public override task ondisconnected(bool s)
                {
                    var list = (list<siginalrmodel>)httpruntime.cache.get("msg_hs");
                    if (list == null)
                    {
                        list = new list<siginalrmodel>();
                    }
                    var closemodel = list.where(x => x.connectionid == context.connectionid).firstordefault();

                    if (closemodel != null)
                    {
                        list.remove(closemodel);

                        clients.group(closemodel.group).removeuserlist(context.connectionid);

                     }
                    httpruntime.cache.insert("msg_hs", list);
                
                    return base.ondisconnected(s);
                }
            }
        

    public class chatticker
        {
            #region 实现一个单例

            private static readonly chatticker _instance =
                new chatticker(globalhost.connectionmanager.gethubcontext<messagehub>());

            private readonly ihubcontext m_context;

            private chatticker(ihubcontext context)
            {

                m_context = context;
                //这里不能直接调用sender，因为sender是一个不退出的“死循环”，否则这个构造函数将不会退出。  
                //其他的流程也将不会再执行下去了。所以要采用异步的方式。  
                //task.run(() => sender());
            }

            public ihubcontext globalcontext
            {
                get { return m_context; }
            }

            public static chatticker instance
            {
                get { return _instance; }
            }

            #endregion
        }

    public class siginalrmodel {
        public string connectionid { get; set; }

        public string group { get; set; }
        public string name { get; set; }
    }
}

我把类和方法都写到一块了，大家最好是分开！

using microsoft.aspnet.signalr;
using microsoft.aspnet.signalr.client;
using signalr.hubs;
using signalr.viewmodels;
using system;
using system.collections;
using system.collections.generic;
using system.io;
using system.linq;
using system.web;
using system.web.mvc;
using newtonsoft.json;
using system.diagnostics;
using system.text.regularexpressions;

namespace signalr.controllers
{
    public class homecontroller : controller
    {
        public actionresult index()
        {
    
            return view();
        }


        public actionresult getv(string v)
        {
            if (!string.isnullorempty(v))
            {
                string url = redishelper.get(v)?.tostring();
                if (!string.isnullorempty(url))
                {
                    return json(new { isok = true, m = url }, jsonrequestbehavior.allowget);
                }
                return json(new { isok = false}, jsonrequestbehavior.allowget);
            }
            return json(new { isok = false }, jsonrequestbehavior.allowget);
        }

        public actionresult getkey(string url)
        {
            if (!string.isnullorempty(url))
            {
                var s = "v" + util.getrandomletterandnumberstring(new random(), 5).tolower();
                var dt = convert.todatetime(datetime.now.adddays(1).tostring("yyyy-mm-dd 04:00:00"));
                int min = convert.toint16((dt - datetime.now).totalminutes);
                redishelper.set(s, url, min);
                return json(new { isok = true, m = s }, jsonrequestbehavior.allowget);
            }
            return json(new { isok = false }, jsonrequestbehavior.allowget);
        }

        public actionresult upfile()
        {
            try
            {
                if (request.files.count > 0)
                {
                    var file = request.files[0];
                    if (file != null)
                    {
                        var imglist = new list<string>() { ".gif", ".jpg", ".bmp", ".png" };
                        var videolist = new list<string>() { ".mp4" };
                        filemodel fmodel = new filemodel();

                        string name = guid.newguid().tostring();
                        string fileext = path.getextension(file.filename).tolower();//上传文件扩展名
                        string path = server.mappath("~/files/") + name + fileext;
                        file.saveas(path);

                        string extension = new fileinfo(path).extension;

                        if (extension == ".mp4")
                        {
                            fmodel.t = 2;
                        }
                        else if (imglist.contains(extension))
                        {
                            fmodel.t = 1;
                        }
                        else
                        {
                            fmodel.t = 0;
                        }
                        string url = guid.newguid().tostring();
                        fmodel.url = "http://" + request.url.host;
                        if (request.url.port != 80)
                        {
                            fmodel.url += ":" + request.url.port;
                        }
                        fmodel.url += "/files/" + name + fileext;
                        getimagethumb(server.mappath("~") + "files\\" + name + fileext, name);
                        return json(new { isok = true, m = "file:" + jsonconvert.serializeobject(fmodel) }, jsonrequestbehavior.allowget);
                    }
                }
            }
            catch(exception ex)
            {
                log.info(ex);
            }
           
           
            return content("");
        }

        public string getimagethumb(string localvideo,string name)
        {
            string path = appdomain.currentdomain.basedirectory;
            string ffmpegpath = path + "/ffmpeg.exe";
            string orivideopath = localvideo;
            int frameindex = 5;
            int _thubwidth;
            int _thubheight;
            getmovwidthandheight(localvideo, out _thubwidth, out _thubheight);
            int thubwidth = 200;
            int thubheight = _thubwidth == 0 ? 200 : (thubwidth * _thubheight / _thubwidth );  
            
            string thubimagepath = path +  "files\\" + name + ".jpg";
            string command = string.format("\"{0}\" -i \"{1}\" -ss {2} -vframes 1 -r 1 -ac 1 -ab 2 -s {3}*{4} -f image2 \"{5}\"", ffmpegpath, orivideopath, frameindex, thubwidth, thubheight, thubimagepath);
            cmd.runcmd(command);
            return name;
        }

        /// <summary>
        /// 获取视频的帧宽度和帧高度
        /// </summary>
        /// <param name="videofilepath">mov文件的路径</param>
        /// <returns>null表示获取宽度或高度失败</returns>
        public static void getmovwidthandheight(string videofilepath, out int width, out int height)
        {
            try
            {

                //执行命令获取该文件的一些信息 
                string ffmpegpath = appdomain.currentdomain.basedirectory +  "/ffmpeg.exe";
                string output;
                string error;
                executecommand("\"" + ffmpegpath + "\"" + " -i " + "\"" + videofilepath + "\"", out output, out error);
                if (string.isnullorempty(error))
                {
                    width = 0;
                    height = 0;
                }

                //通过正则表达式获取信息里面的宽度信息
                regex regex = new regex("(\\d{2,4})x(\\d{2,4})", regexoptions.compiled);
                match m = regex.match(error);
                if (m.success)
                {
                    width = int.parse(m.groups[1].value);
                    height = int.parse(m.groups[2].value);
                }
                else
                {
                    width = 0;
                    height = 0;
                }
            }
            catch (exception)
            {
                width = 0;
                height = 0;
            }
        }

        public static void executecommand(string command, out string output, out string error)
        {
            try
            {
                //创建一个进程
                process pc = new process();
                pc.startinfo.filename = command;
                pc.startinfo.useshellexecute = false;
                pc.startinfo.redirectstandardoutput = true;
                pc.startinfo.redirectstandarderror = true;
                pc.startinfo.createnowindow = true;

                //启动进程
                pc.start();

                //准备读出输出流和错误流
                string outputdata = string.empty;
                string errordata = string.empty;
                pc.beginoutputreadline();
                pc.beginerrorreadline();

                pc.outputdatareceived += (ss, ee) =>
                {
                    outputdata += ee.data;
                };

                pc.errordatareceived += (ss, ee) =>
                {
                    errordata += ee.data;
                };

                //等待退出
                pc.waitforexit();

                //关闭进程
                pc.close();

                //返回流结果
                output = outputdata;
                error = errordata;
            }
            catch (exception)
            {
                output = null;
                error = null;
            }
        }

    }

    public class util
    {
        public static string getrandomletterandnumberstring(random random, int length)
        {
            if (length < 0)
            {
                throw new argumentoutofrangeexception("length");
            }
            char[] pattern = new char[] { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9',
        'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o', 'p',
        'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z' };
            string result = "";
            int n = pattern.length;
            for (int i = 0; i < length; i++)
            {
                int rnd = random.next(0, n);
                result += pattern[rnd];
            }
            return result;
        }
    }

    class cmd
    {
        private static string cmdpath = @"c:\windows\system32\cmd.exe";
        /// <summary>
        /// 执行cmd命令 返回cmd窗口显示的信息
        /// 多命令请使用批处理命令连接符：
        /// <![cdata[
        /// &:同时执行两个命令
        /// |:将上一个命令的输出,作为下一个命令的输入
        /// &&：当&&前的命令成功时,才执行&&后的命令
        /// ||：当||前的命令失败时,才执行||后的命令]]>
        /// </summary>
        /// <param name="cmd">执行的命令</param>
        public static string runcmd(string cmd)
        {
            cmd = cmd.trim().trimend('&') + "&exit";//说明：不管命令是否成功均执行exit命令，否则当调用readtoend()方法时，会处于假死状态
            using (process p = new process())
            {
                p.startinfo.filename = cmdpath;
                p.startinfo.useshellexecute = false;        //是否使用操作系统shell启动
                p.startinfo.redirectstandardinput = true;   //接受来自调用程序的输入信息
                p.startinfo.redirectstandardoutput = true;  //由调用程序获取输出信息
                p.startinfo.redirectstandarderror = true;   //重定向标准错误输出
                p.startinfo.createnowindow = true;          //不显示程序窗口
                p.start();//启动程序

                //向cmd窗口写入命令
                p.standardinput.writeline(cmd);
                p.standardinput.autoflush = true;

                //获取cmd窗口的输出信息
                string output = p.standardoutput.readtoend();
                p.waitforexit();//等待程序执行完退出进程
                p.close();

                return output;
            }
        }
    }
}

我还是都写到一块了，大家记得分开！

using system;
using system.collections.generic;
using system.linq;
using system.web;
using system.web.mvc;

namespace signalr.controllers
{
    public class scontroller : controller
    {
        // get: s
        public actionresult index()
        {
            return view();
        }
    }
}

using system;
using system.collections.generic;
using system.linq;
using system.web;

namespace signalr.viewmodels
{
    public class filemodel
    {
        /// <summary>
        /// 1 : 图片  2：视频
        /// </summary>
        public int t { get; set; }

        public string url { get; set; }
    }
}　

using microsoft.aspnet.signalr.messaging;
using stackexchange.redis;
using system;
using system.collections.generic;
using system.io;
using system.linq;
using system.net;
using system.runtime.serialization.formatters.binary;
using system.threading.tasks;
using system.web;

namespace signalr
{
    public class redishelper
    {
        private static string constr = "xxxx.cn:6379";

        private static object _locker = new object();
        private static connectionmultiplexer _instance = null;

        /// <summary>
        /// 使用一个静态属性来返回已连接的实例，如下列中所示。这样，一旦 connectionmultiplexer 断开连接，便可以初始化新的连接实例。
        /// </summary>
        public static connectionmultiplexer instance
        {
            get
            {
                if (constr.length == 0)
                {
                    throw new exception("连接字符串未设置！");
                }
                if (_instance == null)
                {
                    lock (_locker)
                    {
                        if (_instance == null || !_instance.isconnected)
                        {
                            _instance = connectionmultiplexer.connect(constr);
                        }
                    }
                }
                //注册如下事件
                _instance.connectionfailed += muxerconnectionfailed;
                _instance.connectionrestored += muxerconnectionrestored;
                _instance.errormessage += muxererrormessage;
                _instance.configurationchanged += muxerconfigurationchanged;
                _instance.hashslotmoved += muxerhashslotmoved;
                _instance.internalerror += muxerinternalerror;
                return _instance;
            }
        }

        static redishelper()
        {
        }


        /// <summary>
        /// 
        /// </summary>
        /// <returns></returns>
        public static idatabase getdatabase()
        {
            return instance.getdatabase();
        }

        /// <summary>
        /// 这里的 mergekey 用来拼接 key 的前缀，具体不同的业务模块使用不同的前缀。
        /// </summary>
        /// <param name="key"></param>
        /// <returns></returns>
        private static string mergekey(string key)
        {
            return "signalr:"+ key;
            //return basesysteminfo.systemcode + key;
        }

        /// <summary>
        /// 根据key获取缓存对象
        /// </summary>
        /// <typeparam name="t"></typeparam>
        /// <param name="key"></param>
        /// <returns></returns>
        public static t get<t>(string key)
        {
            key = mergekey(key);
            return deserialize<t>(getdatabase().stringget(key));
        }

        /// <summary>
        /// 根据key获取缓存对象
        /// </summary>
        /// <param name="key"></param>
        /// <returns></returns>
        public static object get(string key)
        {
            key = mergekey(key);
            return deserialize<object>(getdatabase().stringget(key));
        }

        /// <summary>
        /// 设置缓存
        /// </summary>
        /// <param name="key"></param>
        /// <param name="value"></param>
        /// <param name="expireminutes"></param>
        public static void set(string key, object value, int expireminutes = 0)
        {
            key = mergekey(key);
            if (expireminutes > 0)
            {
                getdatabase().stringset(key, serialize(value), timespan.fromminutes(expireminutes));
            }
            else
            {
                getdatabase().stringset(key, serialize(value));
            }

        }


        /// <summary>
        /// 判断在缓存中是否存在该key的缓存数据
        /// </summary>
        /// <param name="key"></param>
        /// <returns></returns>
        public static bool exists(string key)
        {
            key = mergekey(key);
            return getdatabase().keyexists(key); //可直接调用
        }

        /// <summary>
        /// 移除指定key的缓存
        /// </summary>
        /// <param name="key"></param>
        /// <returns></returns>
        public static bool remove(string key)
        {
            key = mergekey(key);
            return getdatabase().keydelete(key);
        }

        /// <summary>
        /// 异步设置
        /// </summary>
        /// <param name="key"></param>
        /// <param name="value"></param>
        public static async task setasync(string key, object value)
        {
            key = mergekey(key);
            await getdatabase().stringsetasync(key, serialize(value));
        }

        /// <summary>
        /// 根据key获取缓存对象
        /// </summary>
        /// <param name="key"></param>
        /// <returns></returns>
        public static async task<object> getasync(string key)
        {
            key = mergekey(key);
            object value = await getdatabase().stringgetasync(key);
            return value;
        }

        /// <summary>
        /// 实现递增
        /// </summary>
        /// <param name="key"></param>
        /// <returns></returns>
        public static long increment(string key)
        {
            key = mergekey(key);
            //三种命令模式
            //sync,同步模式会直接阻塞调用者，但是显然不会阻塞其他线程。
            //async,异步模式直接走的是task模型。
            //fire - and - forget,就是发送命令，然后完全不关心最终什么时候完成命令操作。
            //即发即弃：通过配置 commandflags 来实现即发即弃功能，在该实例中该方法会立即返回，如果是string则返回null 如果是int则返回0.这个操作将会继续在后台运行，一个典型的用法页面计数器的实现：
            return getdatabase().stringincrement(key, flags: commandflags.fireandforget);
        }

        /// <summary>
        /// 实现递减
        /// </summary>
        /// <param name="key"></param>
        /// <param name="value"></param>
        /// <returns></returns>
        public static long decrement(string key, string value)
        {
            key = mergekey(key);
            return getdatabase().hashdecrement(key, value, flags: commandflags.fireandforget);
        }

        /// <summary>
        /// 序列化对象
        /// </summary>
        /// <param name="o"></param>
        /// <returns></returns>
        private static byte[] serialize(object o)
        {
            if (o == null)
            {
                return null;
            }
            binaryformatter binaryformatter = new binaryformatter();
            using (memorystream memorystream = new memorystream())
            {
                binaryformatter.serialize(memorystream, o);
                byte[] objectdataasstream = memorystream.toarray();
                return objectdataasstream;
            }
        }

        /// <summary>
        /// 反序列化对象
        /// </summary>
        /// <typeparam name="t"></typeparam>
        /// <param name="stream"></param>
        /// <returns></returns>
        private static t deserialize<t>(byte[] stream)
        {
            if (stream == null)
            {
                return default(t);
            }
            binaryformatter binaryformatter = new binaryformatter();
            using (memorystream memorystream = new memorystream(stream))
            {
                t result = (t)binaryformatter.deserialize(memorystream);
                return result;
            }
        }

        /// <summary>
        /// 配置更改时
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private static void muxerconfigurationchanged(object sender, endpointeventargs e)
        {
            //loghelper.safelogmessage("configuration changed: " + e.endpoint);
        }

        /// <summary>
        /// 发生错误时
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private static void muxererrormessage(object sender, rediserroreventargs e)
        {
            //loghelper.safelogmessage("errormessage: " + e.message);
        }

        /// <summary>
        /// 重新建立连接之前的错误
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private static void muxerconnectionrestored(object sender, connectionfailedeventargs e)
        {
            //loghelper.safelogmessage("connectionrestored: " + e.endpoint);
        }

        /// <summary>
        /// 连接失败 ， 如果重新连接成功你将不会收到这个通知
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private static void muxerconnectionfailed(object sender, connectionfailedeventargs e)
        {
            //loghelper.safelogmessage("重新连接：endpoint failed: " + e.endpoint + ", " + e.failuretype +(e.exception == null ? "" : (", " + e.exception.message)));
        }

        /// <summary>
        /// 更改集群
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private static void muxerhashslotmoved(object sender, hashslotmovedeventargs e)
        {
            //loghelper.safelogmessage("hashslotmoved:newendpoint" + e.newendpoint + ", oldendpoint" + e.oldendpoint);
        }

        /// <summary>
        /// redis类库错误
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private static void muxerinternalerror(object sender, internalerroreventargs e)
        {
            //loghelper.safelogmessage("internalerror:message" + e.exception.message);
        }

        //场景不一样，选择的模式便会不一样，大家可以按照自己系统架构情况合理选择长连接还是lazy。
        //建立连接后，通过调用connectionmultiplexer.getdatabase 方法返回对 redis cache 数据库的引用。从 getdatabase 方法返回的对象是一个轻量级直通对象，不需要进行存储。

        /// <summary>
        /// 使用的是lazy，在真正需要连接时创建连接。
        /// 延迟加载技术
        /// 微软azure中的配置 连接模板
        /// </summary>
        //private static lazy<connectionmultiplexer> lazyconnection = new lazy<connectionmultiplexer>(() =>
        //{
        //    //var options = configurationoptions.parse(constr);
        //    ////options.clientname = getappname(); // only known at runtime
        //    //options.allowadmin = true;
        //    //return connectionmultiplexer.connect(options);
        //    connectionmultiplexer muxer = connectionmultiplexer.connect(coonstr);
        //    muxer.connectionfailed += muxerconnectionfailed;
        //    muxer.connectionrestored += muxerconnectionrestored;
        //    muxer.errormessage += muxererrormessage;
        //    muxer.configurationchanged += muxerconfigurationchanged;
        //    muxer.hashslotmoved += muxerhashslotmoved;
        //    muxer.internalerror += muxerinternalerror;
        //    return muxer;
        //});


        #region  当作消息代理中间件使用 一般使用更专业的消息队列来处理这种业务场景

        /// <summary>
        /// 当作消息代理中间件使用
        /// 消息组建中,重要的概念便是生产者,消费者,消息中间件。
        /// </summary>
        /// <param name="channel"></param>
        /// <param name="message"></param>
        /// <returns></returns>
        public static long publish(string channel, string message)
        {
            stackexchange.redis.isubscriber sub = instance.getsubscriber();
            //return sub.publish("messages", "hello");
            return sub.publish(channel, message);
        }

        /// <summary>
        /// 在消费者端得到该消息并输出
        /// </summary>
        /// <param name="channelfrom"></param>
        /// <returns></returns>
        public static void subscribe(string channelfrom)
        {
            stackexchange.redis.isubscriber sub = instance.getsubscriber();
            sub.subscribe(channelfrom, (channel, message) =>
            {
                console.writeline((string)message);
            });
        }

        #endregion

        /// <summary>
        /// getserver方法会接收一个endpoint类或者一个唯一标识一台服务器的键值对
        /// 有时候需要为单个服务器指定特定的命令
        /// 使用iserver可以使用所有的shell命令，比如：
        /// datetime lastsave = server.lastsave();
        /// clientinfo[] clients = server.clientlist();
        /// 如果报错在连接字符串后加 ,allowadmin=true;
        /// </summary>
        /// <returns></returns>
        public static iserver getserver(string host, int port)
        {
            iserver server = instance.getserver(host, port);
            return server;
        }

        /// <summary>
        /// 获取全部终结点
        /// </summary>
        /// <returns></returns>
        public static endpoint[] getendpoints()
        {
            endpoint[] endpoints = instance.getendpoints();
            return endpoints;
        }
    }
}