製品、アイデアの販売企画 -...

Copyright © 2004 NTT DATA Corporation

JSR-224 オフライン会議資料

JAX-RPC 2.0における非同期Webサービス実現のための

アーキテクチャ提案

2004年3月11日

株式会社NTTデータ技術開発本部

木村　利幸

2

非同期Webサービスの必要性は？

ＷｅｂサービスＷｅｂサービスクライアント実装クライアント実装

処理ブロック処理ブロック

Ｗｅｂサービス実装Ｗｅｂサービス実装

クライアントクライアント SOAPSOAPサーバサーバ

■ 同期型Webサービス

HTTP 1HTTP 1接続接続

処理要求処理要求

結果取得結果取得

長時間処理には不向き長時間処理長時間処理には不向きには不向き

! ○ HTTP １接続で完結し、サービス構築が容易

! × 完了までクライアント処理がブロックされる

3

非同期Webサービスとは？■非同期型Webサービスの種類

■ コールバック（Push)型

• 「処理依頼」と「結果取得」が別のHTTP接続

• 処理終了時にサーバ側から結果がPush

■ ポーリング(Pull)型

• 「処理依頼」と「結果取得」が別のHTTP接続

• 「結果取得」がクライアント側からのPolling

4

非同期Webサービスの実装方法は？

J2EE PlatformJ2EE Platform

WebWebコンテナコンテナ


JMSJMS EJBEJBコンテナコンテナ

DBMS／EIS


※※ MDB (MessageMDB (Message--Driven Bean)Driven Bean)

※※ JMS (Java Message Service)JMS (Java Message Service)

※※ SLSB (Stateless Session Bean) SLSB (Stateless Session Bean)

MDBMDB

SLSBSLSB

Queue/

Topic

Queue/

Topic


JAX-RPCJAXJAX--RPCRPCJAX-RPCJAXJAX--RPCRPC

EndEnd--pointpoint

InvokeInvokeInvoke

ID取得IDID取得取得

結果取得結果取得結果取得

ポーリング型非同期Webサービスの実現

課題１: サーバ負荷上昇の問題

課題２: クライアント実装複雑化の問題

5


Ｗｅｂサービス実装Ｗｅｂサービス実装Ｗｅｂサービスクライアント実装Ｗｅｂサービスクライアント実装

　　ユーティリティ部　　ユーティリティ部

ポーリング型WSポーリング型ポーリング型WSWS

制御情報DB

AA

BBCC

対策１－サーバ負荷上昇への対応サーバがポーリング制御用の情報をクライアントへ送付する



<<asyncWorkflow> asyncWorkflow> 　　<asyncStaticParam><asyncStaticParam>　　<firstInteval><firstInteval>36003600</firstInterval></firstInterval>　　<iterateInterval><iterateInterval>360360</iterateInterval></iterateInterval>

<timeoutInterval><timeoutInterval>100000100000</timeoutInterval></timeoutInterval><maxRetryCount>50</maxRetryCount><maxRetryCount>50</maxRetryCount>

</asyncStaticParam></asyncStaticParam><asyncWorkflow><asyncWorkflow>

サーバ負荷制御、ポーリング・タイミング制御の実現サーバ負荷制御、ポーリング・タイミング制御の実現サーバ負荷制御、ポーリング・タイミング制御の実現

6

対策２－クライアント実装複雑化への対応

複雑なポーリング処理を隠蔽し、クライアント実装を簡略化

クライアントクライアント



JAX-RPCJAXJAX--RPCRPC

SOAPSOAPサーバサーバ


JAX-RPCJAXJAX--RPCRPC

Ｗｅｂサービス実装Ｗｅｂサービス実装

ユーティリティ部ユーティリティ部

InvokeInvokeInvoke

結果取得結果取得結果取得

同期

同期

//非同期共通

非同期共通

IFIF

制御部制御部

タイマタイマ管理部管理部

制御情報制御情報管理部管理部

実行状態管理部実行状態管理部

7

非同期Webサービスの実現可能性検証

通信部通信部

同期同期APIAPI

ｸﾗｲｱﾝﾄｸﾗｲｱﾝﾄAPAP

通信部通信部

ＳＯＡＰサーバＳＯＡＰサーバ

WebWebサービスサービス

Ｊ２ＥＥ Platform

拡張部拡張部既存部既存部【凡例】

SOAPSOAPクライアントクライアント

AXISAXISAXISAXIS

ﾜｰｸﾌﾛｰ制御ﾜｰｸﾌﾛｰ制御

非同期非同期APIAPI

8

検証結果 ( 1/4 )ユーティリティ適用の効果 - 1

//処理要求用の設定

Servcie service = new Service(serviceName);

Call call = service.createCall();

call.setOperationName(startOperationName);

call.addParameter("in0", XMLType.XSD_STRING,ParameterMode.IN);

call.addParameter("in1", XMLType.XSD_LONG,ParameterMode.IN);

call.setReturnType(XSD_ANY);

call.setTargetEndpointAddress(endpointURL);

call.setMaintainSession(true);

call.invoke(new Object[]{"HELLO", new Integer(10)});//処理要求呼び出し

//パーサの準備

try{

DocumentBuilderFactory d = DocumentBuilderFactory.newInstance();

DocumentBuilderFactory factory = new org.apache.xerces.jaxp.DocumentBuilderFactoryImpl();

boolean isValidate = true;

factory.setValidating(isValidate);

factory.setNamespaceAware(true);

factory.setAttribute("http://apache.org/xml/features/validation/schema" ,new Boolean(true));

builder = factory.newDocumentBuilder();

builder.setErrorHandler(new ValidationErrorHandler());

}catch(ParserConfigurationException e){

throw new InternalException(e);

}

//ヘッダ要素の抽出開始

MessageContext msgContext = call.getMessageContext();

Message msg = (Message)msgContext.getMessage();

SOAPEnvelope envelope = msg.getSOAPEnvelope();

SOAPHeaderElement asyncSOAPHeader = envelope.getHeaderByName(Constants.ASYNCSOAP_NS,Constants.ASYNCSOAPHEADER_LOCAL);

MessageElement asyncSOAPElement = (MessageElement)asyncSOAPHeader.getChildren().get(0);

Document headerDoc = asyncSOAPElement.getAsDocument();

String headerString = new XMLUtils().DocumentToString(headerDoc);

ByteArrayInputStream headerStream = new ByteArrayInputStream(headerString.getBytes());

builder.parse(headerStream);//パース

headerStream.close();

//ステータスを取得

Element asyncServerStatusElement = (Element)headerDoc.getElementsByTagNameNS(Constants.ASYNCSOAP_NS, Constants.ASYNCSERVERSTATUS_LOCAL).item(0);

Element asyncWorkflowElement =(Element)headerDoc.getElementsByTagNameNS(Constants.ASYNCSOAP_NS, Constants.ASYNCWORKFLOW_LOCAL).item(0);

String statusCode = asyncServerStatusElement.getElementsByTagName(Constants.STATUSCODE_LOCAL).item(0).getFirstChild().getNodeValue();

String statusString = asyncServerStatusElement.getElementsByTagName(Constants.STATUSSTRING_LOCAL).item(0).getFirstChild().getNodeValue();

Element detail = (Element)asyncServerStatusElement.getElementsByTagName(Constants.DETAIL_LOCAL).item(0);

NodeList asyncStaticParamNodeList = headerDoc.getElementsByTagName(Constants.ASYNCSTATICPARAM_LOCAL);

//静的パラメータを取得

Element asyncStaticParamElement = (Element)asyncStaticParamNodeList.item(0);

Text firstIntervalText = (Text)asyncStaticParamElement.getElementsByTagName(Constants.FIRSTINTERVAL_LOCAL).item(0).getFirstChild();

long firstInterval = Long.parseLong(firstIntervalText.getNodeValue());

Text iterateIntervalText = (Text)asyncStaticParamElement.getElementsByTagName(Constants.ITERATEINTERVAL_LOCAL).item(0).getFirstChild();

long iterateInterval = Long.parseLong(iterateIntervalText.getNodeValue());

Text timeoutIntervalText = (Text)asyncStaticParamElement.getElementsByTagName(Constants.TIMEOUTINTERVAL_LOCAL).item(0).getFirstChild();

long timeoutInterval = Long.parseLong(timeoutIntervalText.getNodeValue());

Text maxRetryCountText = (Text)asyncStaticParamElement.getElementsByTagName(Constants.MAXRETRYCOUNT_LOCAL).item(0).getFirstChild();

int maxRetryCount = Integer.parseInt(maxRetryCountText.getNodeValue());

//ポーリング用の準備

Call pollingCall = service.createCall();

pollingCall.setOperationName(endOperationName);

pollingCall.setReturnType(XSD_STRING);

//第一回目のポーリングまで待機

Thread.sleep(firstInterval * 1000);//待機

//ポーリングのループ開始

boolean completed = false;//処理完了フラグ

int loop_count = 0;

while(completed){//処理が完了するまでループ

Object ret = pollingCall.invoke(new Object[]{"HELLO", new Integer(10)});//結果確認呼び出し

loop_count++;










builder.parse(headerStream);


//パラメータの抽出






//ステータスの確認

if(!statusCode.equals("RUNNING")){

completed = true;

}

//最大リトライ数のチェック

if(loop_count >= maxRetryCount){

throw new Exception("最大ポーリング回数を超過しました.");

}

Thread.sleep(iterateInterval * 1000);

}

//結果の表示

System.out.println("return : " + ret);

//処理要求用の設定

Servcie service = new Service(serviceName);


call.setOperationName(startOperationName);

call.addParameter("in0", XMLType.XSD_STRING,ParameterMode.IN);

call.addParameter("in1", XMLType.XSD_LONG,ParameterMode.IN);

call.setReturnType(XSD_ANY);


call.setMaintainSession(true);

call.invoke(new Object[]{"HELLO", new Integer(10)});//処理要求呼び出し

//パーサの準備

try{

DocumentBuilderFactory d = DocumentBuilderFactory.newInstance();

DocumentBuilderFactory factory = new org.apache.xerces.jaxp.DocumentBuilderFactoryImpl();

boolean isValidate = true;

factory.setValidating(isValidate);

factory.setNamespaceAware(true);

factory.setAttribute("http://apache.org/xml/features/validation/schema" ,new Boolean(true));

builder = factory.newDocumentBuilder();

builder.setErrorHandler(new ValidationErrorHandler());

}catch(ParserConfigurationException e){

throw new InternalException(e);

}










builder.parse(headerStream);//パース


//ステータスを取得






NodeList asyncStaticParamNodeList = headerDoc.getElementsByTagName(Constants.ASYNCSTATICPARAM_LOCAL);

//静的パラメータを取得

Element asyncStaticParamElement = (Element)asyncStaticParamNodeList.item(0);

Text firstIntervalText = (Text)asyncStaticParamElement.getElementsByTagName(Constants.FIRSTINTERVAL_LOCAL).item(0).getFirstChild();

long firstInterval = Long.parseLong(firstIntervalText.getNodeValue());

Text iterateIntervalText = (Text)asyncStaticParamElement.getElementsByTagName(Constants.ITERATEINTERVAL_LOCAL).item(0).getFirstChild();

long iterateInterval = Long.parseLong(iterateIntervalText.getNodeValue());

Text timeoutIntervalText = (Text)asyncStaticParamElement.getElementsByTagName(Constants.TIMEOUTINTERVAL_LOCAL).item(0).getFirstChild();

long timeoutInterval = Long.parseLong(timeoutIntervalText.getNodeValue());

Text maxRetryCountText = (Text)asyncStaticParamElement.getElementsByTagName(Constants.MAXRETRYCOUNT_LOCAL).item(0).getFirstChild();

int maxRetryCount = Integer.parseInt(maxRetryCountText.getNodeValue());

//ポーリング用の準備

Call pollingCall = service.createCall();

pollingCall.setOperationName(endOperationName);

pollingCall.setReturnType(XSD_STRING);

//第一回目のポーリングまで待機

Thread.sleep(firstInterval * 1000);//待機

//ポーリングのループ開始

boolean completed = false;//処理完了フラグ

int loop_count = 0;

while(completed){//処理が完了するまでループ

Object ret = pollingCall.invoke(new Object[]{"HELLO", new Integer(10)});//結果確認呼び出し

loop_count++;










builder.parse(headerStream);


//パラメータの抽出






//ステータスの確認

if(!statusCode.equals("RUNNING")){

completed = true;

}

//最大リトライ数のチェック

if(loop_count >= maxRetryCount){

throw new Exception("最大ポーリング回数を超過しました.");

}

Thread.sleep(iterateInterval * 1000);

}

//結果の表示

System.out.println("return : " + ret);

"" 利用しない場合利用しない場合

// ポーリングによる実装例

ServiceFactory factory = ServiceFactory.newInstance();

Service service = ServiceFactory.createService(serviceName);


call.setOperationName(new QName(targetNS, operationName));


call.addParameter("in0", XMLType.XSD_STRING, ParameterMode.IN);

call.addParameter("in1", XMLType.XSD_LONG, ParameterMode.IN);

call.setReturnType(XMLType.XSD_STRING);

call.setOperationModel(OperationModel.ASYNC_PULL);

call.invokeAsync(new Object[]{inputData});

long pollingInterval = 2000;

AsyncStatus status = call.getAsyncStatus();

while(status.isRunnging()){

Thread.sleep(pollingInterval);

status = call.getAsyncStatus();

}

String ret = (String)call.getAsyncReturnValue();

System.out.println(ret);

// ポーリングによる実装例

ServiceFactory factory = ServiceFactory.newInstance();

Service service = ServiceFactory.createService(serviceName);


call.setOperationName(new QName(targetNS, operationName));


call.addParameter("in0", XMLType.XSD_STRING, ParameterMode.IN);

call.addParameter("in1", XMLType.XSD_LONG, ParameterMode.IN);

call.setReturnType(XMLType.XSD_STRING);

call.setOperationModel(OperationModel.ASYNC_PULL);

call.invokeAsync(new Object[]{inputData});

long pollingInterval = 2000;

AsyncStatus status = call.getAsyncStatus();

while(status.isRunnging()){

Thread.sleep(pollingInterval);

status = call.getAsyncStatus();

}

String ret = (String)call.getAsyncReturnValue();

System.out.println(ret);

"" 利用した場合利用した場合

実行行数: 75実行行数: 75実行行数: 17実行行数: 17

約4.5倍、ユーザ実装量を減少倍倍、ユーザ実装量を減少約約4.54.5 、ユーザ実装量を減少

9


同期

同期

//非同期共通

非同期共通

IFIF

クライアントクライアント

ＷｅｂサービスＷｅｂサービスクライアント実装クライアント実装 JAXJAX--RPC 1.0, 1.1RPC 1.0, 1.1規定済み規定済み

Call . Invoke ()Call . Invoke ()Call . Invoke ()

Call . invokeOneWay ()Call . invokeOneWay ()Call . invokeOneWay ()

{ Call . invokeAsync () } ?{ { Call . invokeAsync () Call . invokeAsync () } ?} ?

JAXJAX--RPC 2.0RPC 2.0規定範囲規定範囲((議論中議論中))

同期呼び出し同期呼び出し同期呼び出し

一方向呼び出し一方向呼び出し一方向呼び出し

非同期呼び出し非同期呼び出し非同期呼び出し

既存の同期型APIに準ずる用法で非同期呼び出しが可能既存の同期型既存の同期型APIAPIに準ずる用法で非同期呼び出しが可能に準ずる用法で非同期呼び出しが可能

10




同期

同期

//非同期共通

非同期共通

IFIF

同期型API同期型同期型APIAPI

コールバック型APIコールバック型コールバック型APIAPI

ポーリング型APIポーリング型ポーリング型APIAPI

同期型WS同期型同期型WSWS


API層、トランスポート層のアクセス・モデルを完全分離APIAPI層、トランスポート層のアクセス・モデルを完全分離層、トランスポート層のアクセス・モデルを完全分離

11

検証結果 ( 4/4 )



　　ユーティリティ部　　ユーティリティ部


制御情報DB

AA

BBCC

ユーティリティ適用の効果 - 4



<<asyncWorkflow> asyncWorkflow> 　　<asyncStaticParam><asyncStaticParam>　　<firstInteval><firstInteval>36003600</firstInterval></firstInterval>　　<iterateInterval><iterateInterval>360360</iterateInterval></iterateInterval>

<timeoutInterval><timeoutInterval>100000100000</timeoutInterval></timeoutInterval><maxRetryCount>50</maxRetryCount><maxRetryCount>50</maxRetryCount>

</asyncStaticParam></asyncStaticParam><asyncWorkflow><asyncWorkflow>

サーバ負荷制御、ポーリング・タイミング制御の実現サーバ負荷制御、ポーリング・タイミング制御の実現サーバ負荷制御、ポーリング・タイミング制御の実現

12

まとめ■ 非同期型Webサービス対応の３ステップ

■ レベル１：　クライアント側のみ

• Microsoft社は、Delegateモデルに対応済み

■ レベル２：　クライアント/サーバ両方　　　　　　　　　（但し、ポーリング制御を除く）

• 幾つかのベンダは独自方式で対応済み

■ レベル３：　クライアント/サーバ両方　　　　　　　　　（ポーリング制御を含む）

• 本提案のアーキテクチャ

13

ー　The　End　－

本資料で利用される、Sun, Sun Microsystems, Java, J2EEなどの語句は、Sun Microsystems 社殿の商標もしくは登録商標です。また、それ以外の製品名等は一般的にそれら各企業の商標ですが、表示を省略しています。

製品、アイデアの販売企画 -...

Documents