正如标题所说，我们在使用 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 在运行时发布语句是否有任何限制？
谢谢！
亚娜