Nesper在使用win窗口時CPU /內存使用率高：時間

Question

我使用Nesper作爲CEP引擎來處理我的應用程序中的事件。

我想下面的EPL聲明型號：

• 事件的Value場的平均值在有限窗口時間。
• 如果這些平均值中的任何一個匹配了1期望值，則會生成一個事件。

我此建模爲：

SELECT (
(AVG(ParameterEvent1.Value) = 1) OR 
(AVG(ParameterEvent2.Value) = 1) 
... 
(AVG(ParameterEvent50.Value) = 1) 
) AS BooleanValue 
FROM 
ParameterEvent(Id = 1).win:time(3 sec) AS ParameterEvent1, 
ParameterEvent(Id = 2).win:time(3 sec) AS ParameterEvent2 
... 
ParameterEvent(Id = 50).win:time(3 sec) AS ParameterEvent50 

而在一個單獨的線程，我供給發動機與50事件/秒的穩定速率的參數值。

此設置會導致巨大的CPU和RAM使用量，直到發生異常com.espertech.esper.client.EPException: ReaderWriterLock timeout expired。

我想知道是什麼導致了這個問題。 有趣的是，當我將窗口從win:time(3 sec)更改爲std:lastevent()時，可以解決此問題。但是，我需要窗口來確保參數的值是1到生成事件之前的3秒。

完整的演示代碼如下。你只需要安裝Nesper包來運行它：

using System; 
using System.Collections.Generic; 
using System.Linq; 
using System.Text; 
using System.Threading.Tasks; 
using com.espertech.esper.client; 

namespace NesperTester 
{ 
    class Program 
    { 
     public class ParameterEvent 
     { 
      public int Id { get; set; } 
      public int Value { get; set; } 
     } 

     static EPServiceProvider engine; 

     const int NParameters = 50; 

     static void Main(string[] args) 
     { 
      engine = EPServiceProviderManager.GetDefaultProvider(); 

      engine.EPAdministrator.Configuration.AddEventType("ParameterEvent", typeof(ParameterEvent)); 

      CreateEPLStatement(); 

      Task.Factory.StartNew(() => SimulateParameterChange(), TaskCreationOptions.LongRunning); 

      System.Threading.Thread.Sleep(int.MaxValue); 
     } 

     static void CreateEPLStatement() 
     { 
      const int windowLength = 3; 

      var sb = new StringBuilder(); 

      for (int i = 1; i <= NParameters; i++) 
      { 
       sb.AppendFormat("(AVG(ParameterEvent{0}.Value) = 1) OR ", i); 
      } 

      var rule = sb.ToString(); 
      rule = rule.Remove(rule.Length - 4, 4); 
      sb.Clear(); 

      for (int paramId = 1; paramId <= NParameters; paramId++) 
      { 
       sb.AppendFormat("ParameterEvent(Id = {0}).win:time({1} sec) AS ParameterEvent{0}, ", paramId, windowLength); 
       //sb.AppendFormat("ParameterEvent(Id = {0}).std:lastevent() AS ParameterEvent{0}, ", paramId, windowLength); 
      } 

      var selectSources = sb.ToString(); 
      selectSources = selectSources.Remove(selectSources.Length - 2, 2); 

      var eplStr = string.Format("SELECT ({0}) AS BooleanValue FROM {1}", rule, selectSources); 

      var statementName = string.Format("statement1"); 

      var statement = engine.EPAdministrator.CreateEPL(eplStr, statementName, null); 
      statement.Start(); 
      statement.Events += Statement_Events; 
     } 


     static bool PrevState = false; 
     static void Statement_Events(object sender, UpdateEventArgs e) 
     { 
      var underlying = e.NewEvents[0].Underlying as Dictionary<string, object>; 

      if (underlying.Last().Value == null) 
      { 
       Console.WriteLine("Statement value is unknown"); 
      } 
      else 
      { 
       var statementValue = (bool)underlying.First().Value; 

       if (statementValue != PrevState) 
       { 
        PrevState = statementValue; 
        Console.WriteLine("Statement is {0}", statementValue); 
       } 
      } 
     } 

     static void SimulateParameterChange() 
     { 
      int counter = 0; 

      while (true) 
      { 
       for (int i = 1; i <= NParameters; i++) 
       { 
        var evt = new ParameterEvent(); 

        evt.Id = i; 

        if (i == 1 && (counter/4) % 2 == 0) 
        { 
         evt.Value = 1; 
        } 
        else 
        { 
         evt.Value = 0; 
        } 

        engine.EPRuntime.SendEvent(evt); 

        System.Threading.Thread.Sleep(1000/NParameters); 
       } 

       counter++; 
      } 
     } 
    } 
} 

Answer 1

見http://espertech.com/esper/solution_patterns.php#expiry-3 「具有」條款觸發。