今年最后一天了，蛋疼得很，写点东西吧
这个前段时间无聊写的，还有不基于IIS的实现，HttpListener和Socket（Socket暂时没写）。自己到这儿下代码看也行。
所谓的长连接就是服务端长时间挂起请求， 不立即返回，等必要时在返回。
至于客户端通常有Iframe的Chunked方式，和普通的XMLHTTPRequest。Iframe好像就google在使用，因为很难解决浏览器加载进度问题。 普通的XMLHTTPRequest就是ajax操作。
长连接很长么，其实也不长， 虽然基于TCP的Http完全可以保持不断开，但是浏览器有超时机制，长时间没有返回会断开，看普遍都设置为20s，大概是考虑浏览器超时。
Asp.net的实现每个请求都是定位到一个Handler处理，但如果要保持长连接，怎么做呢？我就见过很多人Thread.Sleep()循环等待处理，但是有没想过，.Net 都是走线程池的，如果一个连接一个线程的话，线程池很快就满了，Asp.Net默认为每个核心分配25个线程（为什么是25个呢，在普遍的IO/CPU 比例上这个值大体比较合理，当然根据业务各异，可以自己调节，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资源，我们默认设置每个核心一个线程，多了也没用，因为没有IO等待了。
还是看代码：
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;
//异步请求将请求放入自定义线程池内,由消息抽发或者定时轮询处理
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;\"}");
}
}
{
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;
//异步请求将请求放入自定义线程池内,由消息抽发或者定时轮询处理
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
}
}
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);
}
}
}
}
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实现没有接着写。
已经过了0点，早点睡了。