Siddhi初探

Posted 2020-09-23 coshaho

官方对Siddhi的介绍如下：

Siddhi CEP is a lightweight, easy-to-use Open Source Complex Event Processing Engine (CEP) under Apache Software License v2.0.

Siddhi是一个轻量级的，简单的开源的复杂事件流程引擎。它使用类SQL的语言描述事件流任务，可以很好的支撑开发一个可扩展的，可配置的流式任务执行引擎。

性能管理系统之中，告警模块采用storm作为告警生成组件。传统设计之中，为了支持不同的告警规则类型，我们需要编写不同的业务逻辑代码，但是使用了Siddhi之后，我们只需要配置不同的流任务Siddhiql，即可以支持不同的告警业务。

作为Siddhi入门文章，本文只介绍如何引用Siddhi核心类库以及HelloWorld代码。

1、 maven依赖

  <dependencies>
    <dependency>
      <groupId>org.wso2.siddhi</groupId>
      <artifactId>siddhi-core</artifactId>
      <version>4.0.0-M7</version>
    </dependency>
    
    <dependency>
      <groupId>org.wso2.siddhi</groupId>
      <artifactId>siddhi-query-api</artifactId>
      <version>4.0.0-M7</version>
    </dependency>

    <dependency>
      <groupId>org.wso2.siddhi</groupId>
      <artifactId>siddhi-query-compiler</artifactId>
      <version>4.0.0-M7</version>
    </dependency>
  </dependencies>

 2、 HelloWorld代码（简单的进行数据过滤）

/*
 * Copyright (c) 2016, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. licenses this file to you under the Apache License,
 * Version 2.0 (the "License"); you may not use this file except
 * in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package com.coshaho.learn.siddhi;
import org.wso2.siddhi.core.SiddhiAppRuntime;
import org.wso2.siddhi.core.SiddhiManager;
import org.wso2.siddhi.core.event.Event;
import org.wso2.siddhi.core.query.output.callback.QueryCallback;
import org.wso2.siddhi.core.stream.input.InputHandler;
//import org.wso2.siddhi.core.util.EventPrinter;

/**
 * 
 * SimpleFilterSample.java Create on 2017年6月19日 下午10:54:41    
 *    
 * 类功能说明:   siddhi官方例子，数据过滤
 *
 * Copyright: Copyright(c) 2013 
 * Company: COSHAHO
 * @Version 1.0
 * @Author coshaho
 */
public class SimpleFilterSample 
{
    public static void main(String[] args) throws InterruptedException 
    {
        // Creating Siddhi Manager
        SiddhiManager siddhiManager = new SiddhiManager();

        String siddhiApp = "" +
                "define stream cseEventStream (symbol string, price float, volume long); " +
                "" +
                "@info(name = \'query1\') " +
                "from cseEventStream[volume < 150] " +
                "select symbol,price " +
                "insert into outputStream ;";

        // Generating runtime
        SiddhiAppRuntime siddhiAppRuntime = siddhiManager.createSiddhiAppRuntime(siddhiApp);

        // Adding callback to retrieve output events from query
        siddhiAppRuntime.addCallback("query1", new QueryCallback() 
        {
            @Override
            public void receive(long timeStamp, Event[] inEvents, Event[] removeEvents) 
            {
                // EventPrinter.print(timeStamp, inEvents, removeEvents);
                System.out.print(inEvents[0].getData(0) + " ");
            }
        });

        // Retrieving InputHandler to push events into Siddhi
        InputHandler inputHandler = siddhiAppRuntime.getInputHandler("cseEventStream");

        // Starting event processing
        siddhiAppRuntime.start();

        // Sending events to Siddhi
        inputHandler.send(new Object[]{"Welcome", 700f, 100L});
        inputHandler.send(new Object[]{"WSO2", 60.5f, 200L});
        inputHandler.send(new Object[]{"to", 50f, 30L});
        inputHandler.send(new Object[]{"IBM", 76.6f, 400L});
        inputHandler.send(new Object[]{"siddhi!", 45.6f, 50L});
        Thread.sleep(500);

        // Shutting down the runtime
        siddhiAppRuntime.shutdown();

        // Shutting down Siddhi
        siddhiManager.shutdown();
    }
}

3、 执行结果

4、 Siddhi仓库

 <repositories>
    <repository>
      <id>wso2.releases</id>
      <name>WSO2 internal Repository</name>
      <url>http://maven.wso2.org/nexus/content/repositories/releases/</url>
      <releases>
        <enabled>true</enabled>
        <updatePolicy>daily</updatePolicy>
        <checksumPolicy>ignore</checksumPolicy>
      </releases>
      </repository>
  </repositories>