3

私はエスパーの初心者で、助けてもらいたいです。私はすでに CSV ファイルで Esper を使用することができましたが、ソケットを介して送信されるイベントとして Java オブジェクトを使用する必要があり、インターネット上でガイドとして使用できる簡単な例を見つけることができません。

それらに基づく簡単な例はありますか？

とにかく、私が作ろうとしているコードをここに置いておきます。実行しても何も起こりません。ソケット接続が機能していないようです。

サーバー クラス (イベント クラスも含まれます)。イベントを送信することになっています：

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;
        }   
    }
}

そしてエスパー系クラス。

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();
         }
}

誰かが私を助けようとするのに時間がかかる場合は、事前にどうもありがとう.

4

1 に答える 1

4

問題が解決しました！Esper Dev リストは非常に役に立ちました。ここにあるテストクラスを通じて、Esper + ソケットの使用方法を学びました

よろしくお願いします！

于 2012-10-03T08:26:38.237 に答える