EventStore .NET API Client在使用线程池线程同步写入Event致使EventStore链接中断的问题研究

    最近，在使用EventStore的.NET Client API采用大量线程池线程同步写入Event时（用于模拟ASP.NET服务端大并发写入Event的状况），发现EventStore的链接会随机中断，而且在服务端日志中显示客户端链接Heartbeat超时（若是不了解EventStore，请点击传送门）。因为系统中全局共享一个EventStore链接，当链接中断时，会致使全部的写入操做被Block，而EventStore链接的重连速度比较慢（测试机器上重连须要耗费20秒到1分钟），这样会致使比较严重的性能问题、甚至致使客户端请求超时。

示例代码以下：

for( int index=0;index<100;index++)
{
    Task.Run( ()=> { connection. AppendToStreamAsync(...).Wait(); } );　　
}

   也许有人会问，为何不为每个请求单独创建一个EventStore链接？好吧，绕开官方文档推荐使用单链接这个问题不谈，彷佛为每个请求创建一个链接是一个好的解决方案，可是实际测试发现这个方案比单连接方案更不靠谱。通过测试发现（使用EventStore版本3.0.5.0测试），为每个请求单独创建链接会有如下几个方面的问题：

   1）每次请求都须要从新链接EventStore Server，耗费的时间比较长，性能比单链接差。

   2）重复的创建和断开链接会致使EventStore服务端的TCP端口失去响应（大概30K次链接以后），在这种状况下除非重启EventStore Server，不然客户端再也没法链接上EventStore Server。

   3）若是一个链接非正常中断，则全部的链接都会断开重连。

   因为多链接存在上面的问题（中途也尝试过链接池，发现也一样存在上述问题），咱们只能在单链接这个方向上继续前行，我尝试着将上面的代码修改成以下形式，居然发现问题神奇的消失了。

for( int index=0;index<100;index++)
{
    var tempThread = new Thread(()=>{connection. AppendToStreamAsync(...).Wait();});  
    tempThread.IsBackground = true;
    tempThread.Start();
}

　　这两段代码有什么区别呢？有问题的代码是使用线程池的线程来同步写入Event，而没有问题的代码则使用线程同步写入Event，貌似没有什么区别。会是什么问题呢？我又回过头将第一段代码修改成以下形式（将同步写入修改成异步写入，注意：将Wait方法调用去掉），居然也发现没有问题。

for( int index=0;index<100;index++)
{
    Task.Run( ()=> { connection. AppendToStreamAsync(...); } );　　
}

　　那么问题出在什么地方呢？会不会是EventStore自己的问题呢？因为使用HTTP接口来写入Event很是的稳定，能够基本排除是Event Store Server的问题，那么EentStore .Net Client API有问题的可能性就比较大了。

      没有办法，从GIT上下载EventStore的源代码开始调试，最终发现了问题多是下面的代码引发的：

public void EnqueueMessage(Message message)
        {
            Ensure.NotNull(message, "message");

            _messageQueue.Enqueue(message);
            if (Interlocked.CompareExchange(ref _isProcessing, 1, 0) == 0)
                ThreadPool.QueueUserWorkItem(ProcessQueue);
        }

        private void ProcessQueue(object state)
        {
            do
            {
                Message message;

                while (_messageQueue.TryDequeue(out message))
                {
                    Action<Message> handler;
                    if (!_handlers.TryGetValue(message.GetType(), out handler))
                        throw new Exception(string.Format("No handler registered for message {0}", message.GetType().Name));
                    handler(message);
                }

                Interlocked.Exchange(ref _isProcessing, 0);
            } while (_messageQueue.Count > 0 && Interlocked.CompareExchange(ref _isProcessing, 1, 0) == 0);
        }

　　这段代码的目的是将客户端写入的自定义事件或者系统产生的事件(不管客户端写入的自定义事件仍是系统事件都会被存储到一个惟一的Queue中)，顺序的发送到服务端。说到这里就必须提一下EventStore TCP链接的心跳机制，当客户端与EventStore服务端创建链接以后，客户端会按期发送一个Heartbeat事件到服务端，通知服务端客户端还活着，若是在必定的时间内，服务端收不到来自客户端的Heartbeat事件，那么服务端会主动关闭链接，而且在日志中记录一条客户端Heartbeat超时日志。

     那么这段代码与链接中断有什么关系呢？注意，客户端的Heartbeat事件也是经过这段代码发送到服务端的。若是，我说若是，因为什么缘由致使Heartbeat事件不能及时的发送到服务端，会不会致使链接中断呢？答案是确定的。

     那么这段怎么看都没有问题的代码为何在使用线程池线程并发写入的状况下会致使Heartbeat事件发送不及时呢？是否是这句话有问题？难道在大并发的状况下QueueUserWorkItem会致使ProcessQueue的调用会被延迟？

ThreadPool.QueueUserWorkItem(ProcessQueue);

　　为了验证个人想法，我建立了一个System.Threading.Thread.Timer，定时100毫秒，在Callback中输出两次Callback之间的时间差，当大量的使用ThreadPool的线程时，发现两次Callback之间的时间差慢慢的从200毫秒，越变越大，直到到好几秒。问题到这里就已经很清楚了，当ThreadPool的线程被大量占用时，经过QueueUserWorkItem注册的回调方法会有必定的延迟，具体的延迟时间与ThreadPool的线程被占用的时间和数量有关系，对于实时性要求比较高的任务，好比EventStore的ProcessQueue来讲，是不适合使用QueueUserWorkItem来注册回调的。

     最后，我将EventStore代码中的QueueWorkItem调用替换为线程调用，发现问题解决。

     后续，为了说明这个问题，我在EventStore的Group中发布了一个帖子说明这个问题，EventStore的做者认为Task.Run不能真实的模拟ASP.NET的状况，建议我到真实环境中测试。为此我建立了一个简单的ASP.NET MVC项目和一个简单的客户端，模拟大压力下两种实现的差异。经过测试发现，使用原版的API，随着并发线程数量的增加，Event的写入速度愈来愈慢，而使用修改后的API，则发现随着并发线程数量的增加，Event的写入速度变化不明显，基本上没有太大的差异，理论上来讲在某一个并发下应该会致使Heartbeat超时。因为IIS默认并发访问数量的限制，又懒得去调整服务器，而且因为IIS自己又管理了一套链接池，想压出Heartbeat超时比较困难，就没有作Heartbeat超时的压力测试。