消息处理过程 - 业务整合 - 精通 WebSphere Message Broker

业务整合

4.3 消息处理过程

图 4-8 消息工具 rfhutil 图 4-9 Enqueue 和 Dequeue

Enqueue 工具嵌入在 Toolkit 中，它的内置^MQ客户端环境为西文，所以连接的队列管理器 CCSID 必须为 819，这对于大多数基于中文环境的队列管理器（CCSID=1381）用户可能会感觉不便。此外， Enqueue 工具缺省认为所操作的是代理所在的队列管理器，通过 SYSTEM.BKR.CONFIG 通道接入到队列管理器，如果操作的只是普通队列管理器，则可以手工创建名为 SYSTEM.BKR.CONFIG 的服务器连接通道。

Dequeue 工具可以通过指定的操作参数将消息从队列中取出来并存入文件中。其中，操作参数的用法与 Enqueue 相同，通过 Save As...可以将消息内容存放到任意指定的文件中，这时文件中只含消息体，不含MQMD消息头也不含操作参数。

4.3 消息处理过程

WebSphere Message Broker 开发的消息流本质上就是对消息的处理过程，在这个过程中可以对消息进行识别和加工，也可以基于消息内容实现路由、转换及各种业务逻辑。下面就让我们通过一个例子来感受消息在消息流中的处理过程。

这个例子的应用场景是为了连接订单系统和物流系统。一个网上订单系统在接受到客户的订单后将订单消息放入队列 QIn 中，消息流经过消息格式映射、计算订单价格、填入客户地址等步骤，将订单消息（OrderForm）转换成送货单消息（DeliveryForm）放入队列 QOut 中，由物流系统收到后就可以安排送货并上门收款了。我们将这个例子称为 OrderDelivery（简称 OD），假定订单消息和送货单消息都是 XML 格式，消息流将 QIn 和 QOut 队列连接起来。

4.3.1 环境准备

首先，用 demo 用户登录并创建数据库 OD_DB。我们假定使用 DB2 数据库，可以在 db2 命令窗口中用 create db 命令创建。如果使用其他的数据库，可以用相应的数据库创建命令。

C:\>db2 "create db OD_DB"

C:\>db2 "connect to OD_DB user demo using demo4me"

接着，设置在系统的 ODBC 中添加对 OD_DB 数据库的数据源访问条目，不妨将数据源名也设置为 OD_DB，具体方法可参见“管理控制”相关章节。

然后，创建消息流需要的数据库表并填写相关的记录。

create table CUSTOMER(

CustomerId varchar(4) not null, CustomerName varchar(20),

CustomerAddress varchar(20), PRIMARY KEY(CustomerId));

insert into CUSTOMER values('0001', 'Jonathan', 'Shanghai');

insert into CUSTOMER values('0002', 'Cliff', 'Beijing');

insert into CUSTOMER values('0003', 'Jeffery', 'Guangzhou');

create table PRODUCT(

ProductId varchar(6) not null, ProductName varchar(20),

UnitPrice decimal(7,2), PRIMARY KEY(ProductId));

insert into PRODUCT values('000001', 'Bicycle', 405.50);

insert into PRODUCT values('000002', 'Telephone', 75.55);

insert into PRODUCT values('000003', 'T-Shirt', 19.99);

最后，让我们看一下输入的 OrderForm 消息和输出的 DeliveryForm 消息，如表 4-1 所示。

我们可以创建 OrderForm.xml 和 DeliveryForm.xml 文件分别存放消息内容。为了简单起见，假定订单中只含单一商品及数量，消息流需要根据 OrderForm 中的客户号（CustomerId）添加 DeliveryForm 中的客户姓名和地址，根据商品号（ProductId）和数量查出单价并计算出总价。

表 4-1 输入和输出的消息

OrderForm DeliveryForm

<?xml version="1.0"?>

</OrderForm>

<?xml version="1.0"?>

<CustomerName>Jonathan</CustomerName>

<CustomerAddress>Shanghai</CustomerAddress>

<ProductName>Telephone</ProductName>

</DeliveryForm>

4.3.2 定义消息格式

为了能让消息流识别 OrderForm 和 DeliveryForm 这两种消息格式，我们必须创建消息集并将两种消息在 WMB 中定义出来。

首先，创建消息集项目。切换到代理应用开发视图，在“资源导航”（Resource Navigator）

窗口中右键选择 New→Message Set Project，在 Project Name 中填写 OD_MsgSetProject 并结束。

这样，我们就创建了一个名为 OD_MsgSetProject 的消息集项目。

接着，创建消息集。在“资源导航”窗口中选中 OD_MsgSetProject 项目，右键选择 New

→Message Set，在 Message Set Name 中填写 OD_MsgSet，选择 XML Wire Format 并填写格式名 OD_XML，这表示该消息集支持 XML 格式的消息组织方式。一个消息集项目中只能创建一个消息集。

然后，创建消息定义。在“资源导航”窗口中选中 OD_MsgSetProject 项目，右键选择 NewMessage Definition File，消息定义可以完全由头手工定义实现，也可以从已有的消息格式描述文件中导入，如 XML Schema 文件（*.xsd）、XML DTD 文件（*.dtd）、C 语言头文件（*.h）、

COBOL 语言文件（*.cpy）等。我们选择创建新的消息定义文件，在 File Name 中填写 OD_MsgDefFile 并结束。在成功创建消息定义后 Toolkit 会自动打开消息定义编辑器。

最后，创建 OrderForm 和 DeliveryForm 消息定义。在消息定义编辑器中选中 Messages 右键选择 Add Message 来添加消息定义。在 Messages 中将消息名改为 OrderForm，在 Types 中将消息类型改为 OrderFormType。按照表 4-1 中的消息格式，在 OrderForm 消息中添加本地元素（Add Local Element）OrderId、OrderDate、OrderTime、CustomerId、ProductId 和 Quantity，其中 Quantity 是 int 类型，其他都是 string 类型。类似地，定义 DeliveryForm，

如图 4-10 所示。

图 4-10 OrderForm 和 DeliveryForm 消息定义

定义了消息格式以后，消息流就可以按预定的结构识别并解析输入和输出的消息，但对消息内容的处理仍需要在消息流过程中完成。由于我们已经在前面体验过消息流的开发过程，

所以在以后章节中会扼要叙述。

下面让我们来看一看如何在 OrderForm 和 DeliveryForm 之间建立映射关系。

4.3.3 映射消息格式

首先，创建消息流。创建消息流项目 OD_MsgFlowProject，在项目中创建代理模式 OD_BrokerSchema，再在模式中创建消息流 OD_MsgFlow。

其次，编辑消息流。在消息流中使用 MQInput 节点、Mapping 节点、Compute 节点、MQOutput 节点，四个节点的 In 和 Out 端子首尾相连。MQInput 的 Queue Name 属性设为 QIn，Message Domain 设为 MRM，Message Set 设为 OD_MsgSet，Message Type 设为 OrderForm，Message Format 设为 OD_XML，这表示 WMB 会用我们前面创建的 OrderForm 消息定义来解析读入的消息。为了表达清楚，我们可以将 MQInput 和 MQOutput 节点名分别改为 QIn 和 QOut。

OD_MsgFlow 消息流如图 4-11 所示。我们希望 Mapping 节点之前的消息格式为 OrderForm，

之后的消息格式为 DeliveryForm。

图 4-11 OD_MsgFlow 消息流

然后，建立消息映射。右击 Mapping 节点，选择 Open Map 并接受缺省映射文件名。消息可以分为消息属性（Property）、消息头（Header）和消息体（Body），数据来源可以是消息或者数据库。我们需要指定映射的范围和消息来源，这里选择消息属性和消息体就足够了，数据来源选择 Input Message。在源消息中选择 OrderForm，在目标消息中选择 DeliveryForm，如图 4-12 所示。成功建立消息映射后，Toolkit 会自动打开消息映射编辑器。

图 4-12 选择映射消息源和目标

最后，映射消息字段。在消息映射编辑器中用拖拉的方式将源消息 OrderForm 中的 Properties、OrderId、CustomerId、ProductId、Quantity 字段映射到目标消息 DeliveryForm 中，

如图 4-13 所示。目标消息中的其他字段可以填入缺省值，如 CustomerName 可以填写 'CustomerName'，UnitPrice 可以填写 0.00。

4.3.4 计算消息内容

首先，设置输出消息格式和类型说明。在消息流 Compute 节点以前，消息的格式和类型说明在 MQInput（QIn）节点中设置，如图 4-11 所示。Compute 节点之后，消息要输出到 MQOutput

（QOut）节点对应的队列中，所以在 Compute 中必须设置输出消息的格式和类型说明。打开并编辑 Compute 节点对应的 ESQL 代码，在 BEGIN 和 END 之间填写下面的代码即可。

QIn Mapping Compute QOut

图 4-13 OrderForm 和 DeliveryForm 之间的映射 SET OutputRoot = InputRoot;

SET OutputRoot.Properties.MessageSet = 'OD_MsgSet';

SET OutputRoot.Properties.MessageType = 'DeliveryForm';

SET OutputRoot.Properties.MessageFormat = 'OD_XML';

接着，通过 ESQL 代码填写输出消息中的客户姓名和地址。这需要通过输入消息的客户号（ InputRoot.MRM.CustomerId ）查询 CUSTOMER 表获得客户信息。代码大致为 SELECT...FROM CUSTOMER AS T WHERE T.CustomerId = InputRoot.MRM.CustomerId，这种代码的返回值可以是一个结果集，THE 函数表示取结果集中的第一条记录。我们将结果先在 Environment 中暂存，再将其内容写入输出消息的域中，以取代消息映射时设置的缺省值。

SET Environment.CN = THE(SELECT T.CustomerName FROM Database.CUSTOMER AS T WHERE T.CustomerId = InputRoot.MRM.CustomerId);

SET Environment.CA = THE(SELECT T.CustomerAddress FROM Database.CUSTOMER AS T WHERE T.CustomerId = InputRoot.MRM.CustomerId);

SET OutputRoot.MRM.CustomerName = Environment.CN.CustomerName;

SET OutputRoot.MRM.CustomerAddress = Environment.CA.CustomerAddress;

然后，通过 ESQL 代码填写商品信息并计算订单价格。这需要通过输入商品号

（InputRoot.MRM.ProductId）查询 PRODUCT 表获得单价，再乘以订购的商品数量获得总价。

注意，由于在数据库表定义中单价（UnitPrice）为 decimal 类型，在 ESQL 中会自动对应 decimal 类型并参与计算。

SET Environment.PN = THE(SELECT T.ProductName FROM Database.PRODUCT AS T WHERE T.ProductId = InputRoot.MRM.ProductId);

SET Environment.UP = THE(SELECT T.UnitPrice FROM Database.PRODUCT AS T WHERE T.ProductId = InputRoot.MRM.ProductId);

SET OutputRoot.MRM.ProductName = Environment.PN.ProductName;

SET OutputRoot.MRM.UnitPrice = Environment.UP.UnitPrice;

SET OutputRoot.MRM.TotalPrice = Environment.UP.UnitPrice * InputRoot.MRM.Quantity;

4.3.5 验证处理结果

经过以上消息集和消息流的开发，OrderDelivery 例子基本完成，但细心的读者会发现 Problems 窗口会留有一些警告，表示无法确认 ESQL 代码中对数据库表及其字段的引用。这是因为 Toolkit 在扫描 ESQL 代码时无法确定这些表是否存在，也不知道这些表的结构。这些警告其实并不影响项目的编译、部署和运行，但作为一个好的编程习惯，我们应该导入表结构以消除所有的警告。

从 Toolkit 切换到数据视图（Data Perspective），在“数据库浏览器”（Database Explorer）

窗口中右键选择 New Connection，Connection Name 可以是任何字串（我们在这里不妨用数据库名 OD_DB），选择 Database manager and JDBC driver 即以数据库 Client 端方式连接数据库服务器。在下一步的界面中选择数据库类型（DB2 Universal Database V8.2），输入服务器地址

（Host）和端口（Port）以及连接的用户名和密码，测试该连接通过后结束。这时 Toolkit 会连接数据库服务器并解析所有的用户表结构，完成后会提示是否导入到项目中，选择 Yes 及相关目录，该目录下会生成一套完整的数据库结构描述文件，代理开发视图中的警告也随之消失。

在归档时我们需要同时选择消息流项目（ OD_MsgFlowProject ）和消息集项目

（ OD_MsgSetProject ）。部署时要注意将原先的 MyMsgFlow 消息流停止或删除，因为 MyMsgFlow 和 OD_MsgFlow 都从 QIn 队列中读取消息，两者会竞争输入消息。测试时我们可以将先前准备的 OrderForm.xml 文件中的消息放入 QIn 队列，看 QOut 队列中的输出是否和 DeliveryForm.xml 文件中的内容完全一致。

我们在前面的章节中已经对 WebSphere Message Broker 的开发过程有了初步的认识，在这一章中我们将详细介绍消息流的设计。消息流是由节点和连线组成的，每个节点表示消息流中对消息处理的一个环节，它可以简单到只是一个标记，也可以复杂到需要 ESQL 或 Java 程序来表达相关的处理逻辑。一般来说，节点本身代表了一段消息处理逻辑，它需要在开发环境中

在文檔中精通 WebSphere Message Broker (頁 172-178)