IIS下实现Comet HTTP长连接

时间：2011-01-01 来源：明天.Net

这个东西前段时间无聊写的，还有不基于IIS的实现（HttpListener和Socket），以后有空可能也会写出来。自己到这儿下代码看也行。

所谓的长连接就是服务端长时间挂起请求，具体怎么做就是我们要说的了。

至于客户端通常有Iframe的Chunked方式，和普通的XMLHTTPRequest。Iframe好像就google在使用，因为很难解决浏览器加载进度问题。普通的XMLHTTPRequest就是ajax操作。

长连接很长么，其实也不长，虽然基于TCP的Http完全可以保持不断开，但是浏览器有超时机制，长时间没有返回会断开，因此一般都设置为20s，具体为什么我也想知道。

下面就说具体怎么做了，Asp.net的实现每个请求都是定位到一个Handler处理，但如果要保持长连接，怎么做呢？我就见过很多人Thread.Sleep()循环等待处理，但是有没想过，.Net 都是走线程池的，如果一个连接一个线程的话，线程池很快就满了，Asp.Net默认为每个核心分配25个线程（为什么是25个呢，MSDN上有篇Preformance文章，我就不找了）。当然这个线程数的最优值应该是根据不同的应用二不同。但是如果有1000人同时在线的话，你难道就设成1000个线程，而且每个线程都是在不停的轮询，线程上下文切换和大量轮询操作将会是这个应用的瓶颈。

在Asp.net下有个很简单的实现方式，那就是IHttpAsyncHandle：显示两个方法：

public IAsyncResult BeginProcessRequest(HttpContext context, AsyncCallback cb, object extraData)

public void EndProcessRequest(IAsyncResult result)

BeginProcessRequest就是开始执行的意思：context是上下文，cb其实就是回调FinishRequest （其中调用 EndProcessRequest），通知IIS结束请求，extraData是啥米东西呢？？我也很纳闷，看看源码才知道其实就是context，切，咱们不理他····

原理：在这里最重要的是提供cb参数，也就是Asp.net内部在这个请求执行BeginProcession之后将回调cb给了你，接下来他就不需要额外的线程等待你的请求完成了，可以转向其他请求。但是我们总的需要线程来处理我们的请求和通知IIS我们结束了吧。那么我们就需要自己的工作线程，在BeginProcessRequest将cb放入我们的线程待处理队列，不管是1个还是1000个都是批量处理。而且只占一个线程。为了利用cpu资源，默认设置每个核心一个线程就行了，太多反而因为锁机制影响性能。

还是看代码：

1 将请求加入队列

代码

public IAsyncResult BeginProcessRequest(HttpContext context, AsyncCallback cb, object extraData)
        {
            ChatRequest request = new ChatRequest();
            request.Identity = 1;
            int lastId =0;
            if (!string.IsNullOrEmpty(context.Request.QueryString["lastid"]))
            {
                int.TryParse(context.Request.QueryString["lastid"],out lastId);
            }
            request.lastId = lastId;
            //异步请求将请求放入自定义线程池内,由消息抽发或者定时轮询处理
            //Chunked很难处理浏览器未加载完问题，暂时不做，有消息即返回
            CometAsyncResult result = new CometAsyncResult(request,context, cb, extraData);
            result.HandleCometRequest();
            return result;
        }

        public void EndProcessRequest(IAsyncResult result)
        {
            //在有消息或者是超时内部线程池调用Callback，Asp.Net将调用这里处理结束
            CometAsyncResult cometResult = result as CometAsyncResult;
            if (cometResult.Response.IsTimeOut)
            {
                cometResult.Context.Response.Write("{\"d\":\"failure\",\"message\":\"time out!\"}");
            }
            else
            {
                cometResult.Context.Response.Write("{\"d\":\"success!\",\"message\":\receive:" + cometResult.Response.Message.Count().ToString() + "messages;\"}");
            }
        }

2 线程池

代码 using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using Ncuhome.Chat.Model;

namespace Ncuhome.Chat.SimpleThreadPool
{
    public static class CometThreadPool
    {
        #region Thread
        //最大工作线程
        public const int MaxThreadCount = 10;
        //默认工作线程，处理程序为CPU密集操作，默认与cpu核心数相同即可
        public const int DefaultThreadCount = 2;

        public static int ThreadCount { get; set; }

        internal static CometThread[] CometThreads;

        /// <summary>
        /// 启动线程池，注册会话处理对象
        /// </summary>
        public static void Start(IChatSessionManager sessionManager)
        {
            Start(DefaultThreadCount, sessionManager);
        }

        /// <summary>
        /// 启动线程池，注册会话处理对象
        /// </summary>
        public static void Start(int threadCount, IChatSessionManager sessionManager)
        {
            if (threadCount < MaxThreadCount && threadCount > 0)
            {
                ThreadCount = threadCount;
            }
            else
            {
                ThreadCount = DefaultThreadCount;
            }

            CometThreads = new CometThread[ThreadCount];
            for (int i = 0; i < ThreadCount; i++)
            {
                CometThreads[i] = new CometThread(sessionManager);
            }
        }
        #endregion

        #region Handler
        private static object SyncRoot = new object();

        private static int AssignRequestThreadIndex = 0;

        /// <summary>
        /// 处理消息
        /// </summary>
        public static void HandleMessage(ChatMessageModel message)
        {
            //每个线程各自一份数据
            lock (SyncRoot)
            {
                for (int i = 0; i < ThreadCount; i++)
                {
                    CometThreads[i].HandeChatMessage(message);
                }
            }
        }

        /// <summary>
        /// 处理消息
        /// </summary>
        public static void HandleMessage(IEnumerable<ChatMessageModel> messages)
        {
            //每个线程各自一份数据
            lock (SyncRoot)
            {
                for (int i = 0; i < ThreadCount; i++)
                {
                    CometThreads[i].HandeChatMessage(messages);
                }
            }
        }

        /// <summary>
        /// 把长连接队列
        /// </summary>
        /// <param name="result"></param>
        public static void QueueCometRequest(ICometRequest result)
        {
            lock (SyncRoot)
            {
                if (AssignRequestThreadIndex == ThreadCount)
                {
                    AssignRequestThreadIndex = 0;
                }
                CometThreads[AssignRequestThreadIndex].EnQueueCometRequest(result);
                AssignRequestThreadIndex++;
            }
        }
        #endregion
    }
}

3线程：

代码

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using Ncuhome.Chat.Model;

namespace Ncuhome.Chat.SimpleThreadPool
{
    internal class CometThread
    {
        #region property
        //兼容浏览器，超时应为20s
        private const int RequestTimeOut = 5;
        //CurrentThread
        private Thread ChatThread;
        //Request Queue
        private LinkedList<ICometRequest> CometRequestList;

        private IChatSessionManager SessionManager;

        // 事件触发模式
        private SessionTriggerMode SessionRaisedMode;

        //聊天记录，每个线程保存一份记录，数据量不大，避免并发性能问题
        private List<ChatMessageModel> CometChatMessage;
        //并发锁
        private object RequestSyncRoot = new object();
        private object MessageSyncRoot = new object();

        //会话，事件驱动触发
        private AutoResetEvent SessionWaitHandle = new AutoResetEvent(false);

        //线程无请求是，等待信号
        private AutoResetEvent ThreadWaitHandle = new AutoResetEvent(false);
        #endregion

        /// <summary>
        /// 线程初始化
        /// </summary>
        public CometThread(IChatSessionManager sessionManager)
        {
            CometRequestList = new LinkedList<ICometRequest>();
            //使用事件驱动
            SessionRaisedMode = SessionTriggerMode.EventTrigger;
            CometChatMessage = new List<ChatMessageModel>();

            SessionManager = sessionManager;
            ChatThread = new Thread(new ThreadStart(CometThreadStart));
            ChatThread.IsBackground = false;
            ChatThread.Start();
        }

        public int RequestCount
        {
            get
            {
                return CometRequestList.Count*CometThreadPool.ThreadCount;
            }
        }

        #region 处理
        private void CometThreadStart()
        {
            while (true)
            {
                //转成数组再处理，避免长时间对CometRequestList对象 lock
                ICometRequest[] processRequest;
                lock (RequestSyncRoot)
                {
                    processRequest = CometRequestList.ToArray();
                }

                if (processRequest.Count() == 0)
                {
                    ThreadWaitHandle.WaitOne();
                }

                //处理请求
                if (SessionRaisedMode == SessionTriggerMode.EventTrigger)
                {
                    HandleEventTriggerMode(processRequest);
                }
                else
                {
                    HandlePollingMode(processRequest);
                }
            }
        }

        /// <summary>
        /// 以新消息触发队列请求处理
        /// </summary>
        void HandleEventTriggerMode(ICometRequest[] requests)
        {
            //1s超时进入轮询
            SessionWaitHandle.WaitOne(1000);
            ChatMessageModel[] chatMessages;
            lock (MessageSyncRoot)
            {
                chatMessages = CometChatMessage.ToArray();
                //内存中值保留前20条记录，避免查询耗时
                CometChatMessage = CometChatMessage.Take(20).ToList();
            }

            foreach (var request in requests)
            {

                SessionManager.DoChatSession(request, chatMessages, FinishCometRequest);
            }
        }

        /// <summary>
        /// 单轮询模式处理，定时检查消息队列
        /// </summary>
        void HandlePollingMode(ICometRequest[] requests)
        {
            ChatMessageModel[] chatMessages;
            lock (MessageSyncRoot)
            {
                chatMessages = CometChatMessage.ToArray();
                //内存中值保留前20条记录，避免查询耗时
                CometChatMessage = CometChatMessage.Take(20).ToList();
            }
            foreach (var request in requests)
            {
                SessionManager.DoChatSession(request, chatMessages, FinishCometRequest);
            }

            //定时扫描
            Thread.Sleep(200);
        }

        /// <summary>
        /// 立即处理请求(返回时候得到处理)
        /// </summary>
        void HandleCurrentRequest(ICometRequest request)
        {
            lock (MessageSyncRoot)
            {
                //处理一个请求，不对MessageList copy了
                SessionManager.DoChatSession(request,CometChatMessage ,null );
              if(request.IsCompeled)
              {
                    request.FinishCometRequest();
                }
            }
        }
        #endregion

        #region Messages
        /// <summary>
        /// 添加新消息
        /// </summary>
        public void HandeChatMessage(ChatMessageModel message)
        {
            lock (MessageSyncRoot)
            {
                CometChatMessage.Add(message);
            }
            //新消息信号
            SessionWaitHandle.Set();
        }

        /// <summary>
        /// 添加新消息
        /// </summary>
        public void HandeChatMessage(IEnumerable<ChatMessageModel> messages)
        {
            lock (MessageSyncRoot)
            {
                CometChatMessage.AddRange(messages);
            }
            //新消息信号
            SessionWaitHandle.Set();
        }
        #endregion

        /// <summary>
        /// 完成长连接处理
        /// </summary>
        public void FinishCometRequest(ICometRequest request)
        {
            if (request.IsCompeled||(DateTime.Now - request.BeginTime).TotalSeconds >= RequestTimeOut)
            {
                DeQueueCometRequest(request);

                request.FinishCometRequest();
            }
        }

        /// <summary>
        /// 将请求加入线程处理队列
        /// </summary>
        public void EnQueueCometRequest(ICometRequest request)
        {
            request.CometConcurrentCount = RequestCount;

            //需要立即处理请求，如果有数据及立即返回,无数据才加入队列
            HandleCurrentRequest(request);

            if (request.IsCompeled)
            {
                return;
            }

            //将请求加入队列处理
            lock (RequestSyncRoot)
            {
                CometRequestList.AddFirst(request);
                //通知线程开始工作
                ThreadWaitHandle.Set();
            }
        }

        /// <summary>
        /// 完成请求删除节点
        /// </summary>
        public void DeQueueCometRequest(ICometRequest request)
        {
            lock (RequestSyncRoot)
            {
                CometRequestList.Remove(request);
            }
        }
    }
}

详细代码参加：https://ncuhome.googlecode.com/svn/trunk/ncuhome

经过本机测试在3000个长连接是完成正常工作，代码很不完善，仅供参考，项目里面还有HttpListener的实现，Socket实现没有接着写。