WSO2 CEP内存

Question

我们在CEP上看到内存错误。线程转储显示大约有32000个线程在监视器上休眠。此外，尽管CEP选项指定在外部内存上生成HeapDump，但我们没有看到任何生成的堆转储。请指点。(CEP -Xms256m -Xmx1536m )

1)卡桑德拉号在这一CEP上被禁用

2) CEP版本为2.1.0

3) CEP由WSO2 ESB (使用)负责。

4)除了向CEP发送实际有效负载数据外，ESB还向CEP发送周期性心跳(每15个ec)。

5)我们还在ESB上配置了JMX代理，它每15分钟监视一次CEP (cpu/内存线程)。

6)即使指定了-XX:HeapDumpPath=参数，也没有找到堆转储

• 在此OOM之前，CEP连续运行了7天。重新启动后，我们观察到线程数量正在以每天大约4000-5000线程的速度稳步增长。

CEP日志。

[2013-06-10 05:31:49,040] ERROR -  Thread Thread[ActiveMQ InactivityMonitor  WriteCheckTimer,5,main] died {org.apache.zookeeper.server.NIOServerCnxn}
java.lang.OutOfMemoryError: unable to create new native thread
at java.lang.Thread.start0(Native Method)
at java.lang.Thread.start(Thread.java:640)
at  java.util.concurrent.ThreadPoolExecutor.addIfUnderMaximumPoolSize(ThreadPoolExecutor.java:727)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:657)
at org.apache.activemq.transport.AbstractInactivityMonitor.writeCheck(AbstractInactivityMonitor.java:153)
at org.apache.activemq.transport.AbstractInactivityMonitor$2.run(AbstractInactivityMonitor.java:117)
at org.apache.activemq.thread.SchedulerTimerTask.run(SchedulerTimerTask.java:33)
at java.util.TimerThread.mainLoop(Timer.java:512)
at java.util.TimerThread.run(Timer.java:462)
[2013-06-10 05:31:49,040] ERROR -  Thread Thread[ActiveMQ InactivityMonitor WriteCheckTimer,5,main] died {org.apache.zookeeper.server.NIOServerCnxn}
java.lang.OutOfMemoryError: unable to create new native thread
at java.lang.Thread.start0(Native Method)
at java.lang.Thread.start(Thread.java:640)
at java.util.concurrent.ThreadPoolExecutor.addIfUnderMaximumPoolSize(ThreadPoolExecutor.java:727)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:657)
at org.apache.activemq.transport.AbstractInactivityMonitor.writeCheck(AbstractInactivityMonitor.java:153)
at org.apache.activemq.transport.AbstractInactivityMonitor$2.run(AbstractInactivityMonitor.java:117)
at org.apache.activemq.thread.SchedulerTimerTask.run(SchedulerTimerTask.java:33)
at java.util.TimerThread.mainLoop(Timer.java:512)
at java.util.TimerThread.run(Timer.java:462)

在CEP中配置的一些查询

<cep:query name="xxxBuildUpQuery">
<cep:expression><![CDATA[from  xxxCEPIntgDataStream[interfaceInformationInterfaceName=='xxx-xxx' or interfaceInformationInterfaceName=='xxx-xxx'or 
                               interfaceInformationInterfaceName=='xxx-xxx' or interfaceInformationInterfaceName=='xxx-xxx' or 
                               interfaceInformationInterfaceName=='xxx-xxx' or interfaceInformationInterfaceName=='xxxx-xxx' or 
                               interfaceInformationInterfaceName=='xxx-xxx' or 
                               interfaceInformationInterfaceName=='xxx-xxx' ]#window.time(60000)
insert into buildUpStream interfaceInformationInterfaceName, count(interfaceInformationxxxId) as noOfInflowMsgs group by interfaceInformationInterfaceName]]></cep:expression>
<cep:output brokerName="activemqJmsBroker" topic="xxxBuildUpInfoTopic">
  <cep:xmlMapping>
    <xxxAnalytics>
      <buildUpInfo>
        <interfaceName>{interfaceInformationInterfaceName}</interfaceName>
        <buildUpPerMin>{noOfInflowMsgs}</buildUpPerMin>
      </buildUpInfo>
    </xxxAnalytics>
  </cep:xmlMapping>
</cep:output>
</cep:query>
<cep:query name="xxxQueueDepthQuery">
<cep:expression><![CDATA[from xxxIntgrQueueDepthData_v1
insert into xxxIntgrQueueDepthStream flowName,appName, queueDepth]]>  </cep:expression>
<cep:output brokerName="activemqJmsBroker" topic="xxxIntgrQueueDepthTopic">
  <cep:xmlMapping>
    <xxxAnalytics>
      <queueDepthInfo>
        <flowName>{flowName}</flowName> 
        <appName>{appName}</appName>
        <depth>{queueDepth}</depth>
      </queueDepthInfo>
    </xxxAnalytics>
  </cep:xmlMapping>
</cep:output>
</cep:query>
<cep:query name="xxxClockDataQuery">
  <cep:expression><![CDATA[from testStream
insert into testOutClockDataStream AEDateTime]]></cep:expression>
  <cep:output brokerName="activemqJmsBroker" topic="xxxClockDataTopic">
    <cep:xmlMapping>
      <xxxClockFeed>
        <data>
          <XXDateTime>{XXDateTime}</XXDateTime>
        </data>
      </xxxClockFeed>
    </cep:xmlMapping>
  </cep:output>
 </cep:query>
 <cep:query name="xxxSimltrPaymntAvgQuery_1">
  <cep:expression><![CDATA[from xxxCEPIntgDataStream#window.time(15000)
    insert into xxxSimltrPymntAvgData avg(amount) as avgAmount, currency group by currency]]></cep:expression>
  <cep:output brokerName="activemqJmsBroker" topic="xxxAvgPaymntDetails">
    <cep:xmlMapping>
      <xxxAnalytics>
        <avgPaymentData>
          <avgAmount>{avgAmount}</avgAmount>
          <currency>{currency}</currency>
        </avgPaymentData>
      </xxxAnalytics>
    </cep:xmlMapping>
  </cep:output>

​

谢谢拉吉夫·帕蒂尔

Answer 1

看起来timeBatch窗口是问题所在，我们用它进行了几次查询.一旦拆除，螺纹坡道就被逮捕了。timeBatch窗口特性可能需要更多的测试。

Answer 2

我发现Siddhi启动了一个以Integer.MAX_VALUE为核心池大小的计划线程池。这意味着每个请求都将创建一个没有超时策略的新线程。(参考文献：ThreadPoolExecutor)

在WSO2修复此问题之前，您可以更改此线程池的大小。在org.wso2.siddhi.core.SiddhiManager类中，更改行：

this.siddhiContext.setScheduledExecutorService(Executors.newScheduledThreadPool(Integer.MAX_VALUE))；

( SiddhiManager ver1.1.0wso2v1中的第77行)

对此：

this.siddhiContext.setScheduledExecutorService(Executors.newScheduledThreadPool(100))；

此更改将创建一个核心池大小为100，最大池大小为Integer.MAX_VALUE，空闲线程(超过核心池大小)将在它们完成后立即被移除。