0

正如标题所说,我们在使用 Esper 和 RabbitMQ 构建项目时遇到了问题。因此我们需要使用 Esperio(4.10 版)的 amqp 适配器。

我们从手册中的基础开始,但由于某种原因,我们的查询没有得到应有的评估。我们已经发现需要在 amqp 适配器的声明中切换“EventBusSink”和 EventBusSource(构建错误,例如“未定义输出流”)

我们使用 Spring AMQP 在 Rabbitmq 中设置了 2 个队列,如下所示:

   <rabbit:connection-factory id="connectionFactory" />

   <rabbit:template id="amqpTemplate" connection-factory="connectionFactory" />

   <rabbit:admin connection-factory="connectionFactory" />

   <rabbit:queue name="myQueue" durable="false">
         <rabbit:queue-arguments value-type="java.lang.Long">
                <entry key="x-message-ttl" value="50000" />
         </rabbit:queue-arguments>
   </rabbit:queue>

   <rabbit:queue name="myOutputQueue" durable="false">
         <rabbit:queue-arguments value-type="java.lang.Long">
                <entry key="x-message-ttl" value="50000" />
         </rabbit:queue-arguments>
   </rabbit:queue>

   <rabbit:listener-container
         connection-factory="connectionFactory">
         <rabbit:listener ref="foo" method="listen"
                queue-names="myOutputQueue" />
   </rabbit:listener-container>

   <bean id="foo" class=" messagebroker.Foo" />

并像这样配置 Esper 和 amqp 适配器:

         Configuration config = new Configuration();
         config.addEventType("InputEvent", Event.class);
         config.addEventType("OutputEvent", OutputEvent.class);
         config.getEngineDefaults().getLogging().setEnableTimerDebug(false);
         config.getEngineDefaults().getLogging().setEnableExecutionDebug(true);
         config.getEngineDefaults().getLogging().setEnableQueryPlan(true);
         epService = EPServiceProviderManager.getDefaultProvider(config);
         EPRuntime cepRT = epService.getEPRuntime();
         EPAdministrator cepAdm = epService.getEPAdministrator();
         cepAdm.getConfiguration().addImport(AMQPSource.class.getPackage().getName() + ".*");

         String epl = "Create Dataflow AMQPIncomingDataFlow \n"
                       // + "Create schema test as (id string), \n"
                       + "AMQPSource -> instream<InputEvent> \n"
                       + "{host: 'localhost'," + "port: 5672," + "username:'guest',"
                       + "password:'guest'," + "queueName: 'myQueue',"
                       + "declareDurable: false," + "declareExclusive: false,"
                       + "declareAutoDelete: false," + "prefetchCount : 100,"
                       + "waitMSecNextMsg: 0," + "consumeAutoAck: true,"
                       + "collector: {class: 'AMQPToObjectCollectorSerializable'},"
                       + "logMessages: true } \n" + "EventBusSink(instream){}";

         cepAdm.createEPL(epl);
         EPDataFlowInstance instance = cepRT.getDataFlowRuntime().instantiate(
                       "AMQPIncomingDataFlow");
         instance.start();

         cepAdm.getConfiguration().addImport(
                       AMQPSink.class.getPackage().getName() + ".*");
         String epl2 = "Create Dataflow AMQPOutgoingDataFlow \n"
                       + "EventBusSource -> outstream<OutputEvent>{} \n"
                       + "AMQPSink(outstream)"
                       + "{host: 'localhost',"
                       + "port: 5672,"
                       + "username:'guest',"
                       + "password:'guest',"
                       + "queueName: 'myOutputQueue',"
                       + "declareDurable: false,"
                       + "declareExclusive: false,"
                       + "declareAutoDelete: false,"
                       + "waitMSecNextMsg: 0,"
                       + "collector: {class: 'ObjectToAMQPCollectorSerializable'}, logMessages: true }";

         cepAdm.createEPL(epl2);
         EPDataFlowInstance instance2 = cepRT.getDataFlowRuntime().instantiate(
                       "AMQPOutgoingDataFlow");
         instance2.start();

当我们在一个方法(构造函数或初始化方法)中设置配置和发布多个语句时,一切正常。但是,当我们第一次使用自定义配置(使用适配器)启动 esper 并在稍后阶段发布语句(并随后发送事件以匹配这些语句)时,没有任何内容与我们的语句匹配。尽管如此,我们使用相同的 esperProvider(我们代码中的 epService)。当我们在发布语句的方法中复制配置部分时,我们会得到一个异常,指出适配器已经存在。

一个查询的例子是:

           EPStatement statement = epService.getEPAdministrator().createEPL("insert into OutputEvent select e.id as id, e.id as id, e.sourceRef as sourceRef, 'Start' as eventType from pattern[every e = InputEvent((sourceRef = '28853a2b6a88477197d88ee9f89d2ab7' or sourceRef =      '39b648697f654b2891e79077d3a18fe6' or sourceRef = 'ea0db5e30d244d4fa6e922bf21c89e6f') and standardEvent = 'create')]");

发送的事件:

 epService.getEPRuntime().sendEvent(new InputEvent (50, 51, "28853a2b6a88477197d88ee9f89d2ab7", "item", 50, false, null,  "create"));
 epService.getEPRuntime().sendEvent(new InputEvent (50, 52, "f77537e468c84adfa117d6aab48f929c", " item ", 50, false, null,  "create"));
 epService.getEPRuntime().sendEvent(new InputEvent (50, 53, "1ea4c8af03354a1da134e33783b22b24", " item ",50, false, null,  "create"));
 epService.getEPRuntime().sendEvent(new InputEvent (50, 50, "ab1ead380b6b48bdb3872de4277195a0", " item ",-1, false, null,  "create"));

Esper 在运行时发布语句是否有任何限制?

谢谢!

亚娜

4

2 回答 2

0

我们在 tomcat 服务器上运行,但经过一些尝试和错误后,我们在不知道到底出了什么问题的情况下解决了问题。无论哪种方式,确实没有限制..

于 2014-01-08T13:24:16.667 回答
0

在运行时发布 EPL 没有任何限制。您是否在 J2EE 服务器中运行?

于 2013-11-13T11:54:38.123 回答