springboot中使用esper入門

序

esper是一個比較經典的CEP(Complex Event Processing)的開源實現(開源協議爲GPL v2)，這裏簡單介紹下如何在springboot中使用。

maven

<dependency>
            <groupId>com.espertech</groupId>
            <artifactId>esper</artifactId>
            <version>6.1.0</version>
        </dependency>

config

@Bean
    public EPServiceProvider epServiceProvider() {
        com.espertech.esper.client.Configuration config = new com.espertech.esper.client.Configuration();

        //add event type
        config.addEventType(PersonEvent.class);

        EPServiceProvider epServiceProvider = EPServiceProviderManager.getDefaultProvider(config);
        // epServiceProvider.initialize();
        return epServiceProvider;
    }

    @Bean
    public EPAdministrator epAdministrator() {
        return epServiceProvider().getEPAdministrator();
    }

舊版的話，須要epServiceProvider.initialize();新版已經不用了，在esper-6.1.0-sources.jar!/com/espertech/esper/client/EPServiceProviderManager.java已經初始化了

public static EPServiceProvider getProvider(String providerURI, Configuration configuration) throws ConfigurationException {
        String providerURINonNull = (providerURI == null) ? EPServiceProviderSPI.DEFAULT_ENGINE_URI : providerURI;

        if (runtimes.containsKey(providerURINonNull)) {
            EPServiceProviderSPI provider = runtimes.get(providerURINonNull);
            if (provider.isDestroyed()) {
                provider = getProviderInternal(configuration, providerURINonNull);
                runtimes.put(providerURINonNull, provider);
            } else {
                provider.setConfiguration(configuration);
            }
            return provider;
        }

        // New runtime
        EPServiceProviderSPI runtime = getProviderInternal(configuration, providerURINonNull);
        runtimes.put(providerURINonNull, runtime);
        runtime.postInitialize();

        return runtime;
    }

在esper-6.1.0-sources.jar!/com/espertech/esper/core/service/EPServiceProviderImpl.java的構造器中也調用了初始化

/**
     * Constructor - initializes services.
     *
     * @param configuration is the engine configuration
     * @param engineURI     is the engine URI or "default" (or null which it assumes as "default") if this is the default provider
     * @param runtimes      map of URI and runtime
     * @throws ConfigurationException is thrown to indicate a configuraton error
     */
    public EPServiceProviderImpl(Configuration configuration, String engineURI, Map<String, EPServiceProviderSPI> runtimes) throws ConfigurationException {
        if (configuration == null) {
            throw new NullPointerException("Unexpected null value received for configuration");
        }
        if (engineURI == null) {
            throw new NullPointerException("Engine URI should not be null at this stage");
        }
        this.runtimes = runtimes;
        this.engineURI = engineURI;
        verifyConfiguration(configuration);

        serviceListeners = new CopyOnWriteArraySet<EPServiceStateListener>();

        configSnapshot = takeSnapshot(configuration);
        statementListeners = new CopyOnWriteArraySet<EPStatementStateListener>();
        doInitialize(null);
    }

這裏的調用了兩個初始化，一個是EPServiceProviderImpl構造器的初始化doInitialize(null);一個是runtime.postInitialize();正好跟epServiceProvider.initialize();的兩個初始化方法對應

public void initialize() {
        initializeInternal(null);
    }

    private void initializeInternal(Long currentTime) {
        doInitialize(currentTime);
        postInitialize();
    }

監聽事件

EPStatement statement = epAdministrator.createEPL(epl);
statement.addListener( (newData, oldData) -> {
  String name = (String) newData[0].get("name");
  int age = (int) newData[0].get("age");
  System.out.println(String.format("Name: %s, Age: %d", name, age));
});

發送事件

epServiceProvider.getEPRuntime().sendEvent(new PersonEvent("Peter", 10));

doc

• esper-reference