3
我是Esper的新手,我想得到一些幫助。我已經設法使用CSV文件來使用Esper,但現在我需要使用Java對象作爲通過套接字發送的事件,並且我無法在Internet上找到簡單示例作爲指導。從Esper + sockets開始
有沒有人基於他們一些簡單的例子?
無論如何,我在這裏讓我想要做的工作代碼。當我運行它時,沒有任何反應,它似乎套接字連接不起作用。
服務器類(它也包含事件類)。它被認爲是發送事件:
import java.io.* ;
import java.net.* ;
class Server {
static final int PORT=5002;
public Server() {
try {
ServerSocket skServer = new ServerSocket(PORT);
System.out.println("Listening at " + PORT);
Socket skClient = skServer.accept();
System.out.println("Serving to Esper");
OutputStream aux = skClient.getOutputStream();
ObjectOutputStream flux = new ObjectOutputStream(aux);
int i = 0;
while (i<10) {
flux.writeObject(new MeteoEvent(i,"A"));
i++;
}
flux.flush();
skClient.close();
System.out.println("End of transmission");
} catch(Exception e) {
System.out.println(e.getMessage());
}
}
public static void main(String[] arg) {
new Server();
}
class MeteoEvent{
private int sensorId;
private String GeoArea;
public MeteoEvent() {
}
public MeteoEvent(int sensorid, String geoarea) {
this.sensorId = sensorid;
this.GeoArea = geoarea;
}
public int getSensorId() {
return sensorId;
}
public void setSensorId(int sensorId) {
this.sensorId = sensorId;
}
public String getGeoArea() {
return GeoArea;
}
public void setGeoArea(String geoArea) {
GeoArea = geoArea;
}
}
}
而且基於Esper的類。
import java.io.File;
import java.io.IOException;
import java.sql.SQLException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import com.espertech.esper.client.Configuration;
import com.espertech.esper.client.EPAdministrator;
import com.espertech.esper.client.EPRuntime;
import com.espertech.esper.client.EPServiceProvider;
import com.espertech.esper.client.EPServiceProviderManager;
import com.espertech.esper.client.EPStatement;
import com.espertech.esper.client.EventBean;
import com.espertech.esper.client.UpdateListener;
import com.espertech.esper.event.map.MapEventBean;
import com.espertech.esperio.socket.EsperIOSocketAdapter;
import com.espertech.esperio.socket.config.ConfigurationSocketAdapter;
import com.espertech.esperio.socket.config.DataType;
import com.espertech.esperio.socket.config.SocketConfig;
public class Demo {
public static class CEPListener implements UpdateListener {
private String tag;
public CEPListener (String tag)
{
this.tag = tag;
}
public static void main(String[] args) throws IOException, InterruptedException {
Configuration configuration = new Configuration();
Map<String, Object> eventProperties = new HashMap<String, Object>();
eventProperties.put("sensorId", int.class);
eventProperties.put("GeoArea", String.class);
configuration.addEventType("MeteoEvent", eventProperties);
ConfigurationSocketAdapter socketAdapterConfig = new ConfigurationSocketAdapter();
SocketConfig socketConfig = new SocketConfig();
socketConfig.setDataType(DataType.OBJECT);
socketConfig.setPort(5002);
socketAdapterConfig.getSockets().put("MeteoSocket", socketConfig);
EPServiceProvider cepService = EPServiceProviderManager.getProvider("MeteoSocket",configuration);
EPRuntime cepServiceRT = cepService.getEPRuntime();
EPAdministrator cepAdmin = cepService.getEPAdministrator();
EsperIOSocketAdapter socketAdapter = new EsperIOSocketAdapter (socketAdapterConfig, "MeteoSocket");
socketAdapter.start();
EPStatement stmt = cepAdmin.createEPL("insert into JoinStream select * from MeteoEvent");
EPStatement outputStatementX = cepAdmin.createEPL("select * from JoinStream");
outputStatementX.addListener(new CEPListener("JS"));
cepService.initialize();
Object lock = new Object();
synchronized (lock)
{
lock.wait();
}
}
如果有人需要一些時間來幫助我,請提前非常感謝您。