java 反射 ibm_Java编程语言构建IBM InfoSphere Streams程序(3)

java 运算符 API

除了 Java 运算符模型描述文件之外，Streams 还提供了基于 Java 编程语言的一系列元组处 理类库以简化 Java 运算符的开发。这些类库统称为 Java 运算符 API，绝大多数接口与类都位于 com.ibm.streams.operator 包中，物理上位于 Streams 安装目录之下的 lib 目录。除此之外，Java 运算符 API 还依赖 于 Apache Commons Math 2.1 类库 (commons-math-2.1.jar)，其物理上位于 Streams 安装目录之下的 ext/lib 目录。

当使用 Java 运算符 API 开发 Java 运算符时，上述提到的两个 Jar 包需要显示地包含在 classpath 或者 Streams Studio 的 build 环境中，以确保相应的类能被 Java 编译器找到;但对于 Streams 运行时间，它们是自动进行 加载的，因此并不需要显示指定为 Java 运算符的依赖类库。另外，在 Streams 安装目录之下 lib 目录中 com.ibm.streams.operator.samples.jar 包含的大量 Java 运算符源代码示例，对于开始学习 Java 运算符的开发将是大 有裨益的。

下图描述的是 Java 运算符 API 中主要的类与接口的关系，其中红色标注的是 Streams2.0 版本新提 供的类或接口：

3. Java 运算符 API

Operator 接口：定义了运算度的接口。Streams 运行时将调用该接口的初始化方法，之后对每个收到的元组与结束符调 用相应的处理方法，直至收到关闭请求。

AbstractOperator 类：Streams 提供的抽象类，提供了很多实用的方法以简化运算符的开发，绝大多数情况下应该继承 该抽象类来实现自己的运算符。

OperatorContext 接口：提供了对运算符运行时上下文的访问功能，包括数据流以及参数的描述信息。除此之外，还提 供了用于 Java 异步操作的 ScheduledExectorService 和 ThreadFactory 的访问。当需要使用多线程的情况之下，应该使 用 OpeartorContext 提供的 ScheduledExectorService 或者 ThreadFactory 以保证 Streams 运行时在处理关闭请求时能 够对未完成的任务进行必要的清理工作。

StreamSchema 接口：提供了端口的模式定义，包括 Java 以及 SPL 语言对应的属性名称以及类型描述信息。

StreamingInput 接口：提供了对输入端口以及相关模式的定义。

StreamingOutput 接口：提供了输出端口的描述以及获取新的输出元组并且提交元组的方法。

Tuple 接口：提供了不可变通用元组对象的表示。

OutputTuple 接口：提供了可变通用输出元组的表示。

Logging 接口：提供了 ERROR，INFO，debug，TRACE 四个日志级别，以及 log 和 logException 等方法。

ProcessingElement 接口：提供了处理元素相关的信息查询功能，如：输出目录，数据目录，作业标识符，PE 标识符， 输入输出端口个数，运行模式等。

PEMetrics 接口：定义了 PE 层面上的输入 / 输出端口 Metric 枚举类型，以及对应的获取输入 / 输出 Metric 的方 法。

OperatorMetrics 接口：定义了运算符层面上的输入 / 输出端口 Metric 枚举类型，以及对应的获取输入 / 输出 Metric 的方法。

Streamwindow 接口：定义了窗口构成策略，如基于时间间隔的，基于计数器，结束符，或者数据的增量变化等，使得运 算符输入端口接收到元组以此形成一定的逻辑集合，进行数据聚合运算。

Java 运算符生命周期与线程安全

使用 Java 运算符 API 实现的运算符是 com.ibm.streams.operator.Operator 接口的实例，必须提供一个无参数的构造函数，以便运算符能够以反射的方式被实例化。实例化的运算符实例通过调用 initialize(OperatorContext) 方法来开始其生命周期。在所有端口准备就绪之前，运算符是不能接收和发送元组和结束符 的。一旦端口准备就绪，运算符将调用元组和结束符的处理方法，并在恰当的时候提交元组到输出端口。allPortsReady 方 法用于通知运算符所有端口准备就绪以接收并且提交元组。当包含运算符的处理元素 (PE) 被关闭时，运算符将收到 shutdown 方法的调用，但在此期间元组和结束符的处理方法可能正在执行中。因此 shutdown 方法应该保证所有异步操作 的完成以及相关资源的释放。

运算符的 process 方法将接收所有发送到输入端口的元组，元组在经过处理之后能够 被提交到输出端口。其参数 StreamingInput 描述了元组实例以及相应的接收端口。在 process 方法中，能够使用 Tuple 接口提供的类似 JDBC ResultSet 的方法通过属性下标或者属性名称来获取属性值，例如，getInt, getString, getDouble, getObject。由于元组对象本身是不可变的，这样即使经过多层次的调用，仍然能够安全的获取其属性值。 OutputTuple 接口代表了输出元组，继承自 Tuple 接口，提供了额外的方法来进行属性值的设置。OutputTuple 实例可通 过 StreamingOutput.newTuple 方法来创建，然后使用 StreamingOutput.submit 方法来进行提交。同样，属性值的设置方法也与 JDBC 类似，如，setInt, setString, setDouble, setObject。此外，还可使用 OutputTuple 的 assign 方法对其他元组中名称和类型相同的属性进 行批量复制。

下图总结了 Java 运算符的生命周期以及方法的顺序或者并发执行特性：

4. Java 运算符 生命周期与线程安全