批处理通常被视为遗留“大型机”技术,现在正逐渐成为一种重要的工作负载样式,Java™ 和分布式环境对其的需求越来越大。过去 40 年中,IBM® 一直是批处理系统的领先企业,本文将介绍 IBM® 推出的令人兴奋的 Java 批处理新功能。
Compute Grid 是 IBM WebSphere Extended Deployment V6.1 中推出的一项功能,提供了最完整的企业 Java 批处理编程解决方案。通过 Compute Grid,您可以得到:
虽然 Compute Grid 设计为与其他 WebSphere Extended Deployment 功能进行协作,但也可以独立对其进行购买和部署。在生产环境中,Compute Grid 使用 IBM WebSphere Application Server Network Deployment(此环境采用分布式多计算机配置)进行操作,但 Compute Grid 也提供了单元测试环境,可以在其中运行独立的 WebSphere Application Server。Compute Grid 还提供了基于 Eclipse 的开发体验,而且支持将 IBM Rational® Application Developer 作为全功能开发环境使用。
本文将说明如何使用 Compute Grid 进行 Java 批处理编程。不过,您首先务必充分了解批处理作业及 Compute Grid 所提供的用于构建批处理应用程序的编程模型。讨论了这两个主题后,本文将指导您使用 Compute Grid 和批处理模拟器测试实用工具开发简单的批处理应用程序。
有关 Compute Grid 及其功能的更多信息,请参见参考资料。
从较为抽象的角度而言,批处理作业是声明性构造,可指示执行一个或多个批处理应用程序组成的序列并指定其输入和输出。批处理作业按顺序执行这组任务,以完成特定业务功能。批处理应用程序是设计为在后台以非迭代方式执行的程序。输入和输出通常作为逻辑构造供批处理应用程序访问,而且由批处理作业定义映射到具体的数据资源。
批处理作业通常要处理大量输入/输出数据(通常是面向记录的),这些数据通常是关键业务数据,如客户账户、销售数据等等。批处理作业执行的业务处理任务的范围很广,包括发票生成、账户优化、商机分析等。批处理任务已经在 System z(大型机)环境中使用了几十年,直到今天还继续在很多大中型企业中作为中枢系统使用。
批处理作业的基本内容包括图 1 所示的组成部分。
作业定义描述了要执行的批处理步骤及其运行顺序。每个步骤都定义了要调用的特定批处理应用程序及其输入与输出数据。数据的常见源和目的地包括文件、数据库、事务系统、消息队列等。
Compute Grid 批处理应用程序由一组 POJO 组成,在 Compute Grid 批处理容器的控制下运行,而该容器本身作为标准 WebSphere Application Server 的扩展运行。图 2 显示了应用程序组件及其与批处理容器的关系。
批处理容器在异步 Bean 的控制下运行批处理作业,而您可以将异步 Bean 视为容器管理的线程。批处理容器处理作业定义并执行其生命周期(使用异步 Bean 作为执行单位)。
批处理应用程序由用户提供的以下组件组成:
批处理作业步骤:此 POJO 提供了在批处理作业中作为步骤执行的业务逻辑。批处理容器在处理作业定义的过程中调用批处理作业步骤。
批处理数据流:此 POJO 为批处理作业步骤提供对数据的访问权限。可以将批处理应用程序编写为访问一个或多个批处理数据流。可以将批处理数据流编写为提供对任何类型的数据的访问权限,这包括来自 RDB、文件系统、消息队列以及通过 J2C 连接器等的数据。
批处理容器负责在作业步骤的生命周期中对批处理数据流进行打开、关闭及检查点相关的回调操作。批处理作业步骤本身将对批处理数据流调用各个方法,以获取和放置数据。
批处理应用程序可以选择包括用户提供的以下组件:
检查点算法:批处理容器提供检查点/重新启动机制,以支持从已知的一致性点重新启动作业。可能需要中断作业然后在计划或非计划停机之后重新启动。批处理容器定期调用检查点,以确定是否应该设置检查点。
Compute Grid 提供了两个预置检查点算法,一个支持基于时间的检查点间隔,而另一个支持基于记录计数的检查点间隔。
结果算法:每个批处理作业步骤在完成时都会提供返回代码。结果算法可以获得批处理作业中的所有步骤的返回代码,并返回整个作业的最终总体返回代码。
Compute Grid 提供了预置的结构算法,该算法将数字最大的步骤的返回码作为总体作业返回代码。
通过对批处理容器(属于 Compute Grid 运行时)的简单介绍,您要记住关于批处理容器的以下重要事项:
关于批处理容器的其他详细信息不在本文的讨论范围之内。有关进一步的信息,请参见参考资料。
Compute Grid 批处理编程模型包括四个主要接口,其中两个是构建批处理应用程序的基本接口,而另外两个是可选接口,用于高级场景中:
BatchJobStepInterface 定义批处理容器和批处理应用程序间的交互。
表 1. BatchJobStepInterface 方法| 数据 | 方法摘要 |
|---|---|
| void | createJobStep() createJobStep 由批处理容器在调用 processJobStep 之前调用。 |
| int | destroyJobStep() 批处理容器完成作业步骤处理后调用 destroyJobStep,可以在这里添加任何清理代码。 |
| java.util.Properties | getProperties() 返回在 xJCL 中为批处理作业步骤指定的属性。 |
| int | processJobStep() processJobStep 应该包括批处理作业步骤的所有业务逻辑。 |
| void | setProperties(java.util.Properties properties) 由批处理容器调用,用于向批处理作业步骤提供在 xJCL 中指定的属性。 |
BatchDataStream 对批处理应用程序的特定输入源或输出目的地进行抽象,并定义 Compute Grid 和具体的 BatchDataStream 实现之间的交互。
表 2. BatchDataStream 方法| 数据 | 方法摘要 |
|---|---|
| void | close() 批处理容器将调用 close 方法来告知 BDS,BDS 的用户已经完成了 BDS 的使用。 |
| java.lang.String | externalizeCheckpointInformation() 批处理容器将在处理工作的检查点完成阶段调用 externalizeCheckpointInformation 方法。 |
| java.lang.String | getName() 返回此批处理数据流的逻辑名称。 |
| java.util.Properties | getProperties() 返回在 xJCL 中为此 BDS 指定的属性。 |
| void | initialize(java.lang.String logicalname, java.lang.String jobstepid) 批处理容器将在作业步骤的初始化期间调用 initialize 方法。这就允许 BDS 对流进行初始化,以供批处理步骤使用。 |
| void | intermediateCheckpoint() 批处理容器调用 intermediateCheckpoint 方法告知 BDS 刚刚完成了一个检查点。 |
| void | internalizeCheckpointInformation(java.lang.String chkptinfo) 批处理容器在重新启动批处理步骤期间调用 internalizeCheckpointInformation 方法,这样就可以让 BDS 将其内部状态重置为上次成功处理检查点时的状态。 |
| void | open() 批处理容器调用 open 方法来指示将要使用 BDS 和为进行操作而对 BDS 做准备。 |
| void | positionAtCurrentCheckpoint() 批处理容器调用 positionAtCurrentCheckpoint 方法来向 BDS 提供信号,指示应该在 internalizeCheckpointInformation 方法集中定义的点开始处理流。 |
| void | positionAtInitialCheckpoint() 批处理容器调用 positionAtInitialCheckpoint 向 BDS 提供信号,指示应该在 xJCL 输入所定义的初始点开始处理流。 |
| void | setProperties(java.util.Properties properties) 批处理容器调用 setProperties 来将 xJCL 中指定的 BDS 属性作为 java.util.Properties 传递到 BDS。 |
CheckpointPolicyAlgorithm 定义 Compute Grid 和自定义检查点策略实现间的交互。检查点策略用于确定 Compute Grid 将何时对正在运行的批处理作业进行检查点操作,以便在计划或非计划中断之后重新启动。Compute Grid 包括两个现成的检查点策略,如表 3 中所示。
ResultsAlgorithm 定义 Compute Grid 和自定义结果算法间的交互。结果算法用于提供作业的总体返回代码。该算法能够获取每个作业步骤的返回代码。Compute Grid 包括一个现成的结果算法,如表 3 中所示。
表 3. 现成算法| 算法 | 类名称 | 描述 |
|---|---|---|
| 检查点策略 | com.ibm.wsspi.batch.checkpointalgorithms.recordbased | 根据所处理的输入记录数量对批处理作业进行检查点操作。 |
| 检查点策略 | com.ibm.wsspi.batch.resultsalgorithms.jobsum | 返回编号最高的步骤的返回代码。 |
| 结果 | com.ibm.wsspi.batch.checkpointalgorithms.recordbased | 根据所处理的输入记录的数量对批处理作业步骤进行检查点操作。 |
我们已经了解了批处理作业的基本概念和 Compute Grid 批处理编程模型,接下来就要通过简单练习来应用这些概念,我们在其中将重点关注基本批处理接口,即 BatchJobStepInterface 和 BatchDataStream。本文剩下的部分将详细介绍如何使用 Eclipse 开发环境实现示例批处理作业步骤和批处理数据流并使用本文中所提供的实用工具(名为 Batch Simulator)来对其进行测试。
设置环境
安装 Eclipse V3.2 或更高版本(如果尚未安装 Batch Simulator 实用工具,请下载并安装。Batch Simulator 实用工具在本文中仅为演示目的而提供。)可以在下载文件中包括的 README PDF 文件中找到该实用工具的下载和安装说明。
创建批处理数据流
package com.ibm.websphere.samples;
import com.ibm.websphere.batch.BatchContainerDataStreamException;
import com.ibm.websphere.batch.BatchDataStream;
public abstract class TextFileBatchDataStream implements BatchDataStream {
...
} |
protected enum ACCESS_MODE {R,W};
protected ACCESS_MODE _access_mode;
protected TextFileBatchDataStream(ACCESS_MODE access_mode) {
_access_mode= access_mode;
} |
public void setProperties(Properties properties) {
props = properties;
} |
public void initialize(String logicalname, String jobstepid) {
Properties prop = getProperties();
fn = prop.getProperty("FILENAME");
} |
public void open() throws BatchContainerDataStreamException {
try {
if ( _access_mode == ACCESS_MODE.R) {
_file = new RandomAccessFile(_fn,"r");
}
else {
_file = new RandomAccessFile(_fn,"rw");
}
} catch (IOException e) {
throw new BatchContainerDataStreamException(e);
}
} |
public String getNextRecord() throws
BatchContainerDataStreamException {
String input = null;
try {
input = _file.readLine();
} catch (IOException e) {
throw new BatchContainerDataStreamException(e);
}
return input;
} |
public void putNextRecord(String r) throws
BatchContainerDataStreamException {
try {
_file.writeBytes(r);
_file.write('\n');
this._position+=r.length()+1;
} catch (IOException e) {
throw new BatchContainerDataStreamException(e);
} |
package com.ibm.websphere.samples;
public class TestInputBatchDataStream extends TextFileBatchDataStream {
public TestInputBatchDataStream() {
super(ACCESS_MODE.R);
}
} |
package com.ibm.websphere.samples;
public class TestOutputBatchDataStream extends TextFileBatchDataStream {
public TestInputBatchDataStream() {
super(ACCESS_MODE.W);
}
} |
创建批处理作业步骤
由于已经实现了批处理数据流,编写批处理作业步骤就相对比较简单。所有批处理作业步骤必须实现 BatchJobStepInterface,因此按以下所示类似的方法声明您的代码:
package com.ibm.websphere.samples;
import com.ibm.websphere.batch.BatchConstants;
import com.ibm.websphere.batch.BatchJobStepInterface;
import com.ibm.websphere.batch.BatchDataStreamMgr;
import com.ibm.websphere.batch.BatchContainerDataStreamException;
import com.ibm.websphere.batch.JobStepID;
import com.ibm.websphere.batch.context.JobStepContextMgr;
import com.ibm.websphere.batch.context.JobStepContext;
public class TestBatchJobStep implements BatchJobStepInterface {
...
} |
在此示例中,有五个方法要实现;这里我们所感兴趣的有 createJobStep 和 processJobStep 方法。(您可以在示例代码中看到 getProperties、setProperties 和 destroyJobStep 方法。)
public void createJobStep() {
try {
JobStepID id = getJobStepID();
_testInputBatchDataStream= (TestInputBatchDataStream)
BatchDataStreamMgr.getBatchDataStream( "input", id.getJobstepid() );
_testOutputBatchDataStream= (TestOutputBatchDataStream)
BatchDataStreamMgr.getBatchDataStream( "output", id.getJobstepid() );
}
catch (BatchContainerDataStreamException e) {
throw new RuntimeException (e);
}
} |
private JobStepID getJobStepID() {
JobStepContext ctx= JobStepContextMgr.getContext();
return ctx.getJobStepID();
} |
JobStepContext 对象公开以下接口:
表 4. JobStepContext 方法| 数据 | 方法摘要 |
|---|---|
| java.lang.String | getJobID() 返回当前作业的作业名称。 |
| JobStepID | getJobStepID() 返回当前步骤的 JobStepID 对象。 |
| java.lang.String | getStepID() 返回当前步骤的步骤名称。 |
| java.lang.Object | getUserData() 返回存储在此上下文中的用户数据。 |
| void | setUserData(java.lang.Object o) 在此上下文中设置用户数据对象。 |
public int processJobStep() {
try {
String l= _testInputBatchDataStream.getNextRecord();
if ( l == null ) {
return BatchConstants.STEP_COMPLETE;
}
else {
_testOutputBatchDataStream.putNextRecord(l);
return BatchConstants.STEP_CONTINUE;
}
}
catch (Exception e) {
throw new RuntimeException("TestBatchJobStep: error in
processJobStep: ",e);
}
} |
随本文提供的 Eclipse 项目中包括全部示例代码。
测试代码
正如前面提到的,Compute Grid 提供了在独立 WebSphere Application Server 中执行的测试环境,但为了进一步简化此示例(并避免其他应用程序打包和部署任务),本文提供了名为 Batch Simulator 的测试辅助工具。Batch Simulator 允许您在 Eclipse 环境内对批处理 POJO 进行基本测试。请记住,尽管 Batch Simulator 在 J2SE 环境中运行,但实际的 Compute Grid 执行环境是 Java EE,因为它在 WebSphere Application Server 内运行。但是 Batch Simulator 在测试 POJO 间的基本流和测试批处理作业步骤中的基本业务逻辑时非常有用。
运行 Batch Simulator:
您需要指定以下信息的 Eclipse 运行配置:
表 5. 现成算法| 属性 | 值 |
|---|---|
| 主类 | com.ibm.websphere.batch.BatchSimulator |
| 程序参数 | "\${workspace_loc:${project_name}}/jobdefs/testbatchjobstep.properties" |
BatchSimulator: start job JOB1 ... BatchSimulator: end job JOB1 - RC=0 |
com.ibm.websphere.batch.BatchSimulatorTask
该工作区提供了两个示例 Ant 脚本,可以将其用于启动此任务,如图 4 中所示。
WebSphere Extended Deployment Compute Grid 提供了批处理作业步骤及其输入输出的简单抽象。编程模型非常简洁,而且使用简单。通过内置的检查点/回滚机制,可以方便地构建可靠、可重新启动的 Java 批处理应用程序。
随本文提供的 Batch Simulator 实用工具提供了替代测试环境,可在您的 Eclipse(或 Rational Application Developer)开发环境中运行。它的 xJCL 生成功能可帮助您直接进入下一阶段在 Compute Grid 单元测试服务器中的测试。
来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/14789789/viewspace-419006/ ,如需转载,请注明出处,否则将追究法律责任。