java - 通过 Esper 事件进行单独进程的通信

Question

我试图让多个 java 进程使用 Esper 交换事件。一个进程应该发送事件，另一个进程准备查询并根据报告的事件使用react。

当这两个操作在同一个 java 进程中完成时，一切正常。但是当我使用两个不同的进程时，它们只是看不到对方。 我想知道这次沟通的关键是什么？我对提供者使用了相同的名称。到目前为止我能做的就是这些。

制作人:

String aType = espertest.dummy.A.class.getName();
Configuration cepConfig = new Configuration();
cepConfig.addEventType("A",aType);
EPServiceProvider epService = EPServiceProviderManager.getProvider("DummyProvider", cepConfig);
Object o = new A();
epService.getEPRuntime().sendEvent(o);

消费者:

String aType = A.class.getName();
String expression = "select count(*) from "+aType + "";
System.out.println("Our Query: " + expression);

Configuration cepConfig = new Configuration();
cepConfig.addEventType("A",aType);

EPServiceProvider epService = EPServiceProviderManager.getProvider("DummyProvider", cepConfig);

EPStatement statement = epService.getEPAdministrator().createEPL(expression);

DummyListener listener = new DummyListener();
statement.addListener(listener);
System.out.println("Anything");


try{
    A a = new A();
    epService.getEPRuntime().sendEvent(a);
    Thread.sleep(60000);
}catch(Exception E)
{
    System.out.println("Exception ");
}

消费者尝试对 A 类型的事件进行计数。它还发送 A 的实例作为测试，这工作得很好。监听器按预期被调用。上面的代码只是摘录。

Answer 1

您需要配置中间件(消息队列、分布式缓存、网络文件系统、套接字连接等......)以将事件从生产者 JVM 获取到消费者 JVM。如果您可以将生产者和消费者部署到支持 Apache Camel 的容器(例如 ServiceMix)，那么建立一个使用 ActiveMQ 将对象传输到 Esper 的原型(prototype)应该很简单，因为 Camel 对这两种产品都支持。

JVM 1

• 来自数据源
• 至 CEP 引擎 1
• 到消息队列

JVM 2(也可以托管 MQ Broker)

• 来自消息队列
• 至 CEP 引擎 2
• 目的地

更新:

如果生产者和消费者可以是同一 JVM 中的线程，那么问题可能出在消费者身上。我看不到消费者在哪里对生产者的事件执行任何操作。请尝试类似的方法(向生产者/消费者提供 esper 引用，并使用更新方法重新处理消费者以处理 select 语句的结果)。

测试驱动程序:

public Driver() {
    String aType = espertest.dummy.A.class.getName();
    Configuration cepConfig = new Configuration();
    cepConfig.addEventType("A",aType);
    EPServiceProvider epService = EPServiceProviderManager.getProvider("DummyProvider", cepConfig);
    Consumer c = new Consumer(epService);
    Producer p = new Producer(epService);
}

制作人:

public Producer(EPServiceProvider epsp) {
    Object o = new A();
    epsp.getEPRuntime().sendEvent(o);
}

消费者:

public Consumer(EPServiceProvider epsp) {
    EPStatement statement = epsp.getEPAdministrator().createEPL(input);
    statement.setSubscriber(this);
}
public void update(A event) {
    System.out.println("Consumer received event!");    
}