S4-定制开发一个简单销售监控的示例

AliKevin 写道

   本系列文章不涉及过多的S4的理论内容，因为s4论文中描述相当清楚（我认为我实在是说的不会比论文中更清楚了：）呵呵），论文信息请看论文http://dl.iteye.com/topics/download/704e5924-0dd8-34df-b44f-2efbc91de071

    S4现有框架的应用开发十分方便，我们只需要实现标准接口，配置定制应用的spirng配置问题就，利用Spring的注入方式将你的应用集成到S4框架中。
所以我可以任何方式建立java工程,形成jar包和配置文档部署到s4的apps下面,但为了和s4统一开发方式我们还是推荐用gradle构建应用项目.


一、销售监控的示例场景描述

    销售监控的示例的应用场景是：假设我们关注所有产品的销售情况，我们监控所有产品的销售事件，当有产品售出的时候会产生一个Sell的事件流，事件流中包含产品的名称和销售数量，当销售数量大于10000时候，我们会触发另一个事件流Celebrate的事件，通知文艺团队进行庆祝演出准备（当然是假设，没有什么节目可以看哦：））。事件流的处理我们只是在控制台输出信息。

二、建立gradle的目录结构

root@slave:/kevin/sellMoniter# cd /kevin
root@slave:/kevin# mkdir sellMoniter
root@slave:/kevin# cd sellMoniter/
root@slave:/kevin/sellMoniter# mkdir -p src/main/resources
root@slave:/kevin/sellMoniter# mkdir src/main/java
root@slave:/kevin/sellMoniter# vi build.gradle

build.gradle内容如下：

/*
 * Copyright (c) 2011 Yahoo! Inc. All rights reserved.
 *
 * Licensed 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. See accompanying LICENSE file.
 */

/*
 * Twitter Topic Count Application Build Script
 *
 * Modify this script to create a build script for your app.
 *
 * NOTE: You must set up the environment variable S4_IMAGE before
 * running this script.
 *
 * Tasks:
 *
 *   gradlew TASK1, TASK2, ... (or gradle TASK1, ... if gradle is installed.)
 *
 *   build - builds the application
 *   install - creates scripts to run applications
 *   deploy - deploys the S4 app to the S4 image
 *   clean - cleans build dir and removes app from S4 image
 *   eclipse - creates an project for the Eclipse IDE
 */

/* Set a version number for your app. */
 version = new Version(major: 0, minor: 1, bugfix: 0)
 group = 'alibaba.com'

 /* Read S4_IMAGE environment variable. */
 env = System.getenv()
 s4Image = env['S4_IMAGE']

 if (s4Image == null) {
     logger.warn("\nEnvironment variable S4_IMAGE not set.")
     System.exit(-1)
 }

/* Search these repos to find artifacts. Gradle will download and cache. */
repositories {
    flatDir name: 's4core', dirs:  "${s4Image}/s4-core/lib"
    flatDir name: 's4driver', dirs:  "${s4Image}/s4-driver/lib"
    mavenLocal()
    mavenCentral()
}

/* Include handy Gradle plugins. */
apply plugin: 'eclipse'
apply plugin: 'java'
apply plugin: "application"

/* Set Java version. */
sourceCompatibility = 1.6
targetCompatibility = 1.6

/* Main application to run ... */
mainClassName = "com.alibaba.s4.SellMachine"
applicationName = "sellMachine"

/* Dependencies. */
dependencies {
 	compile('io.s4:s4-core:0.3.0' )
 	compile('io.s4:s4-driver:0.3.0' )
 	compile('org.json:json:20090211' )
 	compile('com.google.code.gson:gson:1.6' )
 	compile('log4j:log4j:1.2.15' )
 	compile('commons-cli:commons-cli:1.2' )
 	compile('commons-logging:commons-logging:1.1.1' )
 	compile('commons-io:commons-io:2.0.1' )
 	testCompile('junit:junit:4.4' )
}

/* Customize your jar files. */
manifest.mainAttributes(
    provider: 'gradle',
    'Implementation-Url': 'http://www.cn.alibaba-inc.com',
    'Implementation-Version': version,
    'Implementation-Vendor': 'The S4-SellMoniter Project',
    'Implementation-Vendor-Id': 'alibaba.com'
)

/* Bug workaround. */
eclipseClasspath {
    downloadSources = false; // required for eclipseClasspath to work
}

/* Create an inage to copy and archive your app. */
deployImage = copySpec {
    into ("s4-apps/" + project.name + "/lib") {
        from project.configurations.runtime
        from project.configurations.archives.allArtifactFiles
    }
    into ("s4-apps/" + project.name) {
        from project.sourceSets.main.resources
    }
}

/* Copy to the S4 Image. */
task deploy(type: Copy) {
    description = "Copy app files to deployment dir."
    destinationDir = file(s4Image)
    with deployImage
}

/* Add remove app to the clean task. */
task cleanDeployment(type: Delete) {
    delete("${s4Image}/s4-apps/${project.name}")
}
clean.dependsOn cleanDeployment

/* Generates the gradlew scripts.
http://www.gradle.org/1.0-milestone-3/docs/userguide/gradle_wrapper.html */
task wrapper(type: Wrapper) {
    gradleVersion = '1.0-milestone-3'
}

class Version {
    int major
    int minor
    int bugfix
    String releaseType

    String toString() {
        "$major.$minor.$bugfix${releaseType ? '-'+releaseType : ''}"
    }
}

三、生成eclipse工程

为了我们快速开发，我完成grade的目录机构后我们利用gradle命令生成eliplse工程：

root@slave:/kevin/sellMoniter# gradle eclipse
:eclipseClasspath
Download file:/root/.m2/repository/junit/junit/4.4/junit-4.4.pom
Download file:/root/.m2/repository/junit/junit/4.4/junit-4.4.jar
:eclipseJdt
:eclipseProject
:eclipse

BUILD SUCCESSFUL

Total time: 8.719 secs
root@slave:/kevin/sellMoniter#

四、用eclipse中开发相关的类见附件【sellMoniter.rar】

a.事件对象类Sell.java

/**
 * Project: sellMoniter
 *
 * File Created at 2011-10-17
 * $Id$
 *
 * Copyright 1999-2100 Alibaba.com Corporation Limited.
 * All rights reserved.
 *
 * This software is the confidential and proprietary information of
 * Alibaba Company. ("Confidential Information").  You shall not
 * disclose such Confidential Information and shall use it only in
 * accordance with the terms of the license agreement you entered into
 * with Alibaba.com.
 */
package com.alibaba.s4;

/**
 * TODO Comment of Sell
 *
 * @author jincheng.sunjc
 */
public class Sell {
    private String sellInfo; //format productName:count,notebook:5

    /**
     * @return the sellInfo
     */
    public String getSellInfo() {
        return sellInfo;
    }

    /**
     * @param sellInfo the sellInfo to set
     */
    public void setSellInfo(String sellInfo) {
        this.sellInfo = sellInfo;
    }

    @Override
    public String toString() {
        String[] info = this.sellInfo.split(":");
        return "[We had sell " + info[1] + " " + info[0] + " ! :)]";
    }

    public String getName() {
        return this.sellInfo.split(":")[0];
    }

    public int getCount() {
        return Integer.parseInt(this.sellInfo.split(":")[1].trim());
    }

    public static void main(String[] args) {
        Sell s = new Sell();
        s.setSellInfo("notebook:5");
        System.out.println(s.toString());
        System.out.println(s.getCount());
        System.out.println(s.getName());
    }
}

b.事件对象类Celebrate.java

/**
 * Project: sellMoniter
 *
 * File Created at 2011-10-17
 * $Id$
 *
 * Copyright 1999-2100 Alibaba.com Corporation Limited.
 * All rights reserved.
 *
 * This software is the confidential and proprietary information of
 * Alibaba Company. ("Confidential Information").  You shall not
 * disclose such Confidential Information and shall use it only in
 * accordance with the terms of the license agreement you entered into
 * with Alibaba.com.
 */
package com.alibaba.s4;

/**
 * TODO Comment of Celebrate
 *
 * @author jincheng.sunjc
 */
public class Celebrate {
    private String celebrateInfo;

    /**
     * @return the celebrateInfo
     */
    public String getCelebrateInfo() {
        return celebrateInfo;
    }

    /**
     * @param celebrateInfo the celebrateInfo to set
     */
    public void setCelebrateInfo(String celebrateInfo) {
        this.celebrateInfo = celebrateInfo;
    }

    public String getCelebrate() {
        return "1";
    }

    public void setCelebrate(String id) {
        // do nothing
    }

    @Override
    public String toString() {
        return this.celebrateInfo;
    }
}

c.Sell事件流处理类SellPE.java

/**
 * Project: sellMoniter
 *
 * File Created at 2011-10-17
 * $Id$
 *
 * Copyright 1999-2100 Alibaba.com Corporation Limited.
 * All rights reserved.
 *
 * This software is the confidential and proprietary information of
 * Alibaba Company. ("Confidential Information").  You shall not
 * disclose such Confidential Information and shall use it only in
 * accordance with the terms of the license agreement you entered into
 * with Alibaba.com.
 */
package com.alibaba.s4;

import io.s4.dispatcher.Dispatcher;
import io.s4.processor.AbstractPE;

/**
 * TODO Comment of SellPE
 *
 * @author jincheng.sunjc
 */
public class SellPE extends AbstractPE {
    /**
     * Dispatcher that will dispatch events on <code>Sentence *</code> stream.
     */
    private Dispatcher dispatcher;

    public Dispatcher getDispatcher() {
        return dispatcher;
    }

    public void setDispatcher(Dispatcher dispatcher) {
        this.dispatcher = dispatcher;
    }

    public void processEvent(Sell sell) {
        System.out.println("Received: " + sell);

        if (sell.getCount() > 10000) {
            System.out.print("well done we need a celebrate...");
            Celebrate c = new Celebrate();
            c.setCelebrateInfo("Hi,Because " + sell + " ,So we must have a celebration meeting....");
            // dispatch a Sentence event
            dispatcher.dispatchEvent("Celebrate", c);
        }
    }

    @Override
    public void output() {
        // TODO Auto-generated method stub
    }

    @Override
    public String getId() {
        return this.getClass().getName();
    }
}

d.Celebrate事件流处理类CelebratePE.java

/**
 * Project: sellMoniter
 *
 * File Created at 2011-10-17
 * $Id$
 *
 * Copyright 1999-2100 Alibaba.com Corporation Limited.
 * All rights reserved.
 *
 * This software is the confidential and proprietary information of
 * Alibaba Company. ("Confidential Information").  You shall not
 * disclose such Confidential Information and shall use it only in
 * accordance with the terms of the license agreement you entered into
 * with Alibaba.com.
 */
package com.alibaba.s4;

import io.s4.processor.AbstractPE;

/**
 * TODO Comment of SellPE
 *
 * @author jincheng.sunjc
 */
public class CelebratePE extends AbstractPE {
    public void processEvent(Celebrate celebrate) {
        System.out.println("Received: " + celebrate.getCelebrateInfo());
    }

    @Override
    public void output() {
        // TODO Auto-generated method stub
    }

    @Override
    public String getId() {
        return this.getClass().getName();
    }

}

e.Spring配置文件sellMoniter-conf.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">

	<bean id="sellCatcher" class="com.alibaba.s4.SellPE">
		<property name="dispatcher" ref="dispatcher" />
		<property name="keys">
			<list>
				<value>Sell *</value>
			</list>
		</property>
	</bean>

	<bean id="celebrateCatcher" class="com.alibaba.s4.CelebratePE">
		<property name="keys">
			<list>
				<value>Celebrate *</value>
			</list>
		</property>
	</bean>

	<bean id="dispatcher" class="io.s4.dispatcher.Dispatcher"
		init-method="init">
		<property name="partitioners">
			<list>
				<ref bean="celebratePartitioner" />
			</list>
		</property>
		<property name="eventEmitter" ref="commLayerEmitter" />
		<property name="loggerName" value="s4" />
	</bean>

	<bean id="celebratePartitioner" class="io.s4.dispatcher.partitioner.DefaultPartitioner">
		<property name="streamNames">
			<list>
				<value>Celebrate</value>
			</list>
		</property>
		<property name="hashKey">
			<list>
				<value>celebrate</value>
			</list>
		</property>
		<property name="hasher" ref="hasher" />
		<property name="debug" value="true" />
	</bean>

</beans>

f.客户端测试类SellMachine.java

/**
 * Project: sellMoniter
 *
 * File Created at 2011-10-17
 * $Id$
 *
 * Copyright 1999-2100 Alibaba.com Corporation Limited.
 * All rights reserved.
 *
 * This software is the confidential and proprietary information of
 * Alibaba Company. ("Confidential Information").  You shall not
 * disclose such Confidential Information and shall use it only in
 * accordance with the terms of the license agreement you entered into
 * with Alibaba.com.
 */
package com.alibaba.s4;

import io.s4.client.Driver;
import io.s4.client.Message;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.Reader;

/**
 * TODO Comment of SellMachine
 *
 * @author jincheng.sunjc
 */
public class SellMachine {
    public static void main(String[] args) {
        String hostName = "localhost";
        hostName = "192.168.254.129";
        int port = 2334;
        String streamName = "Sell";
        String clazz = "com.alibaba.s4.Sell";

        Driver d = new Driver(hostName, port);
        Reader inputReader = null;
        BufferedReader br = null;
        try {
            if (!d.init()) {
                System.err.println("Driver initialization failed");
                System.exit(1);
            }

            if (!d.connect()) {
                System.err.println("Driver initialization failed");
                System.exit(1);
            }

            inputReader = new InputStreamReader(System.in);
            br = new BufferedReader(inputReader);

            for (String inputLine = null; (inputLine = br.readLine()) != null;) {
                String sellInfo = "{\"sellInfo\":\"" + inputLine + "\"}";
                System.out.println("sellInfo-> " + sellInfo);
                Message m = new Message(streamName, clazz, sellInfo);
                d.send(m);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                d.disconnect();
            } catch (Exception e) {
            }
            try {
                br.close();
            } catch (Exception e) {
            }
            try {
                inputReader.close();
            } catch (Exception e) {
            }
        }

    }
}

OK,这些简单的代码已经足以完成我们的业务场景的需求了：）。

五、部署sellMoniter

root@slave:/kevin/s4/build/s4-image# rm -fr $S4_IMAGE/s4-apps/*
root@slave:/kevin/s4/build/s4-image# rm $S4_IMAGE/s4-core/logs/s4-core/*
root@slave:/kevin/s4/build/s4-image# rm $S4_IMAGE/s4-core/lock/*
root@slave:cd /kevin/sellMoniter
root@slave:/kevin/sellMoniter# :/kevin/sellMoniter# gradle install

:compileJava UP-TO-DATE
:processResources UP-TO-DATE
:classes UP-TO-DATE
:jar UP-TO-DATE
:startScripts UP-TO-DATE
:installApp

BUILD SUCCESSFUL

Total time: 7.258 secs

root@slave:/kevin/sellMoniter# gradle deploy
:compileJava UP-TO-DATE
:processResources UP-TO-DATE
:classes UP-TO-DATE
:jar UP-TO-DATE
:deploy

BUILD SUCCESSFUL

Total time: 6.81 secs

这样我们将我们开发的应用部署到s4中了，可以进行运行测试了：）

六、启动s4服务和adapetr服务

$S4_IMAGE/scripts/start-s4.sh -r client-adapter
appName=s4
dequeuer number: 6
[]
[/kevin/s4/build/s4-image/s4-apps/sellMoniter/sellMoniter-conf.xml]
Adding processing element with bean name sellCatcher, id com.alibaba.s4.SellPE
adding pe: com.alibaba.s4.SellPE@14d5bc9
Using ConMapPersister ..
Adding processing element with bean name celebrateCatcher, id com.alibaba.s4.CelebratePE
adding pe: com.alibaba.s4.CelebratePE@1202d69
Using ConMapPersister ..

如上信息说明我们部署了sellMoniter应用，并加载了sellMoniter-conf.xml配置文件,

root@slave:/kevin/s4# $S4_IMAGE/scripts/run-client-adapter.sh -s client-adapte-g s4 -d $S4_IMAGE/s4-core/conf/default/client-stub-conf.xml
.client.Adapter -t default -c /kevin/s4/build/s4-image/s4-core -d /kevin/s4/build/s4-image/s4-core/conf/default/client-stub-conf.xml
appName=client-adapter
dequeuer number: 12
Adding InputStub genericStub
Adding OutputStub genericStub

如上说明adapter已经启动。

七、启动测试类，并查看s4服务端信息

a.客户端

root@slave:/kevin/sellMoniter/build/install/sellMachine/bin# ./sellMachine
iphone:500
sellInfo-> {"sellInfo":"iphone:500"}
iphone:50000
sellInfo-> {"sellInfo":"iphone:50000"}

b.s4服务端

Received: [We had sell 500 iphone ! :)]
Received: [We had sell 50000 iphone ! :)]
well done we need a celebrate...{
   java.lang.String celebrateInfo
   java.lang.String celebrate
}

Using fast path!
Value 1, partition id 0
Received: Hi,Because [We had sell 50000 iphone ! :)] ,So we must have a celebration meeting....

如上信息证明当我们销售5个iphone的时候我们只是处理的一个sell的事件流，当我们的产品销售数量超过10000的时候我们就要触发一个celebrate的事件流，进行庆祝通知。：）

好了，到此我们针对S4的客户定制开发的例子也开发完毕了，让我们小小的庆祝一下吧：）Cheers\!

评论

3 楼 kspengjun 2011-12-27  

你这个例子怎么跑不通啊 ？ 求赐教？  QQ :515842619


Exception in thread "main" java.lang.NullPointerException
        at io.s4.client.Driver.init(Driver.java:279)
        at com.alibaba.s4.SellMachine.main(SellMachine.java:43)
user@server8:~/s4/sellMoniter/build/install/sellMachine/bin$

2 楼 kspengjun 2011-12-27  

1.root@slave:/kevin/s4# $S4_IMAGE/scripts/run-client-adapter.sh -s client-adapte-g s4 -d $S4_IMAGE/s4-core/conf/default/client-stub-conf.xml 
.client.Adapter -t default -c /kevin/s4/build/s4-image/s4-core -d /kevin/s4/build/s4-image/s4-core/conf/default/client-stub-conf.xml 


.client.Adapter  这个参数是什么意思啊？

无法读取 /home/user/s4/build/s4-image/s4-core/conf/.client.Adapter/s4-core.properties-header:

报了个错， 能解释一下么？谢谢

1 楼 kspengjun 2011-12-27  

root@slave:/kevin/s4# $S4_IMAGE/scripts/run-client-adapter.sh -s client-adapte-g s4 -d $S4_IMAGE/s4-core/conf/default/client-stub-conf.xml 


你的以上这个命令里的这个client-adapte-g  少了一个"r" 应该是adapter ，望改进额