ICE系列培训(二)

1、ICE系列培训(二),黄亮 统一网管平台专项,培训内容,ICE中的异步程序设计方法(AMD,AMI) 1.AMI:Asynchronous Method Invocation 异步方法调用 2.AMD:Asynchronous Method Dispatch 异步方法派发,1. 概述,1.1 同步模型与异步模型区别 1.2 AMI概述 1.3 AMD概述 1.4 元数据(metadata)与AMI，AMD,1.1 同步模型与异步模型区别,两者的区别： 同步模型中发出调用的线程会阻塞到操作返回。异步模型则正好相反，发出调用的线程不会阻塞的等待调用结果的返回。 Ice 在本质上是一个异步的中间件平

2、台，出于对应用 （及其程序员）的考虑而模拟了同步的行为。当 Ice 应用通过代理、向远地对象发出同步的双向调用时，发出调用的线程会阻塞起来，以模拟同步的方法调用。在此期间， Ice run time在后台运行，处理消息，直到收到所需的答复为止，然后发出调用的线程就可以解除阻塞并解编结果。,1.2 AMI概述,AMI： 异步方法调用这个术语描述的是客户端的异步编程模型支持。 使用 AMI发出远地调用，在 Ice run time 等待答复的同时，发出调用的线程不会阻塞。相反，发出调用的线程可以继续进行各种活动，当答复最终到达时， Ice run time会通知应用。通过回调发给应用对象通知,1.

3、3 AMD概述,AMD: 使用 AMD 时，服务器可以接收一个请求，然后挂起其处理，以尽快释放分派线程。当处理恢复、结果已得出时，服务器要使用 Ice run time 提供的回调对象，显式地发送响应。 AMD 操作通常会把请求数据 （也就是，回调对象和 操作参数）放入队列 ，供应用的某个线程 （或线程池）随后处理用。这样，服务器就使分派线程的使用率降到了最低限度，能够高效地支持数千并发客户。,1.3 AMD概述,为什么要使用AMD： （1）默认情况下：一个服务器在同一时刻所能支持的同步请求数受到 Ice run time的服务器线程池的尺寸限制 。如果所有线程都在忙于分派长时间运行的操作，那

4、么就没有线程可用于处理新的请求，客户就会体验到不可接受的无响应状态。 （2）Ice run time的服务器线程池尺寸受限于主机的CPU数。,1.4 元数据与AMI，AMD,1）ICE自3.4以后，提供了一套全新的AMI(异步调用方式)的API，新的API已不需要在slice文件中标记“ami”这样的元数据。 2）ICE提供向下兼容性，支持“ami”的语言标记，但已标注为deprecated. 3) 服务端使用AMD异步派发方式编程，仍然必须在slice文件中，对slice定义标注“amd”元数据。 Tips: 给客户提供某个调用方法的同步和异步版本是有益的，但如果在服 务器中这样做，程序员就

5、必须实现两种版本的分派方法，并不能带来切实的好处，而且还会带来若干潜在的缺陷。,1.4 元数据与AMI，AMD,AMD元数据标记方式： 1）接口级(interface) amd interface I bool isValid(); float computeRate(); ; 2) 操作级(operation) interface J amd void startProcess(); int endProcess(); ;,1.4 元数据与AMI，AMD,AMD元数据的标记方式 Tips: 在操作层面、而不是接口或类的层面指定元数据，不仅能使所生成的代码的数量降到最低限度，而且，更重要的是，

6、它还能使复杂度降到最低限度。尽管异步模型更为灵活，它的使用也更复杂。因此，最好只把它用于那些能从中获得某种好处的操作；对其他操作则使用更简单的同步模型。,2. AMI编程,2.1 基本API 2.2 AsynResult类 2.3 轮询方式的完成通知(Polling for Completion) 2.4 通用的完成通知回调函数(Generic Completion Callbacks) 2.5 类型安全的回调(Type-Safe Completion Callbacks) 2.6 单路调用(Oneway Invocations) 2.7 流量控制(Flow Control) 2.8 批量请求

7、(Batch Requests) 2.9 并发(Concurrency) 2.10 限制(Limitations) 2.11 示例,2.1 Slice介绍,（1）Slice是什么 Slice（Specification Language for Ice)是分离对象接口和实现的重要抽象机制。Slice建立了客户和服务器之间的契约，这些契约描述了应用所需的类型及对象接口。这些描述是中立与语言的，所以客户端和服务端可以使用不同的实现语言。 Slice提供编译工具将Slice定义编译为具体语言的实现代码，即客户端的Proxy code和服务端的Skeleton,这样用户只用把注意力放在业务逻辑的处理上

8、。,2.1 Basic Asynchronous API,1.异步方法通知(AMI)API ICE从3.4开始采用新的AMI API： module Demo interface Employees string getName(int number); ; ; 1）不必再显式的注明“ami”元数据，slice工具会自动的生产同步调用与异步调用的stub代码。 Ice:AsyncResultPtr begin_getName(Ice:Int number); Ice:AsyncResultPtr begin_getName(Ice:Int number, const Ice:Context,2

9、.1 Basic Asynchronous API,(2）begin_xxxMethod负责向队列中插入一个调用的请求，当请求插入到队列后，便可立即返回，不必阻塞的等待调用完成。 (3)end_xxxMethod负责收集异步调用的结果。当异步调用的结果还未处理完成时，调用end_xxxMethod方法会阻塞到异步调用完成处理结果。换句话说，如果结果已完成，调用end_xxxMethod方法会立即返回。 示例： EmployeesPrx e = .; Ice:AsyncResultPtr r = e-begin_getName(99); / Continue to do other things

10、 here. string name = e-end_getName(r); Tips： AsyncResultPtr是AsynResult的智能指针，存储了ICE运行时对于异步调用的状态追踪的信息。在调用begin_xxxMethod方法相关的end_xxxMethod方法时，必须传入相应的AsyncResultPtr对象,2.1 Basic Asynchronous API,（4）异步方法调用中的参数 Slice定义的方法参数在AMI的begin_,end_方法中有相应的变化： double op(int inp1, string inp2, out bool outp1, out lon

11、g outp2); AMI： Ice:AsyncResultPtr begin_op(Ice:Int inp1, const :std:string 注意： 1）slice定义中的输入参数转移为begin_op中的唯一输入参数。 2）slice定义中的输出参数作为end_op的输出参数，end_op的返回值为原slice定义的返回值。 3）Begin_op返回AsyncResultPtr作为end_op的输入参数。,2.1 Basic Asynchronous API,（5）异常处理(Exception Handling) 1)通常情况下，当异步调用过程中发生异常，在end_xxxMethod

12、中抛出，即使异常发生在begin_xxxMethod中。 2）但有2种例外的情况： 当销毁了当前的Communicator后，再调用异步方法，会抛出CommunicatorDestroyedException异常。 当调用方式出错时，抛出IceUtil:IllegalArgumentException异常（begin_,end_配对错误，调用的方法参数错误）,2.2AsynResult类,（1）AsynResult类声明 class AsyncResult : virtual public IceUtil:Shared, private IceUtil:noncopyable public:

13、virtual bool operator=(const AsyncResult,AsyncResult类由begin_op方法返回，封装了异步调用的有关信息,2.2AsynResult类,(2) 方法说明 bool operator=(const AsyncResult ; /同步发送 FileHandle file = open(.); FileTransferPrx ft = .; const int chunkSize = .; Ice:Int offset = 0; while (!file.eof() ByteSeq bs; / Read a chunk bs = file.rea

14、d(chunkSize); / Send the chunk ft-send(offset, bs); offset += bs.size(); ,FileHandle file = open(.); FileTransferPrx ft = .; const int chunkSize = .; Ice:Int offset = 0; list results; const int numRequests = 5; while (!file.eof() ByteSeq bs; bs = file.read(chunkSize); / Send up to numRequests + 1 ch

15、unks asynchronously. Ice:AsyncResultPtr r = ft-begin_send(offset, bs); offset += bs.size(); / 等待请求完成发送 r-waitForSent(); results.push_back(r); /超过了水位标则阻塞等待. while (results.size() numRequests) Ice:AsyncResultPtr r = results.front(); results.pop_front(); r-waitForCompleted(); /等待剩余的请求完成. while (!result

16、s.empty() Ice:AsyncResultPtr r = results.front(); results.pop_front(); r-waitForCompleted(); ,2.6 完成通知,1.轮询方式的完成通知 1）同步方式与异步方式效率上的差距 2）异步方式AsyncResult队列控制 提高传输效率 控制内存占用,2.6 完成通知,2.通用回调函数方式的完成通知 相对于轮询方式的优点： 解除应用逻辑与异步调用机制的耦合性。利用Ice 运行时完成客户端的异步处理逻辑。 Slice提供的重载方法，红色高亮部分是回调完成通知的特征。 Ice:AsyncResultPtr beg

17、in_getName( Ice:Int number, const Ice:CallbackPtr Ice:CallbackPtr: Ice运行时提供的回调函数智能指针，存储了自定义的回调函数实例。 当发生回调时，Ice运行时会执行回调函数对象上绑定的方法。,2.6 完成通知,3.自定义的回调对象示例： class MyCallback : public IceUtil:Shared public: void finished(const Ice:AsyncResultPtr 回调对象封装了异步调用实质的业务逻辑，异步调用本身是发出非阻塞的调用请求，但重要的是发出了请求以后，应用逻辑该如何执行

18、。,/ 应用程序AMI调用示例： EmployeesPrx e = .; MyCallbackPtr cb = new MyCallback; Ice:CallbackPtr d = Ice:newCallback(cb, ,2.6 完成通知,4.回调对象的编写要点 1）继承IceUtil:Shared 2) 回调对象中的回调方法，必须符合以下声明方式： void callback_method(AsyncResultPtr 5)回调方法必须捕获所有可能由end_xxxMethod抛出的异常，如果遗漏了某异常，Ice运行时默认会在日志中记录，但也会忽略这个异常(设置Ice.Warn.AMICa

19、llback属性为0，可以不记录日志)。,2.6 完成通知,5.应用程序AMI调用示例： EmployeesPrx e = .; MyCallbackPtr cb = new MyCallback; Ice:CallbackPtr d = Ice:newCallback(cb, 注意: 1)Ice:newCallback辅助函数，将自定义的回调对象智能指针与回调方法绑定后，返回Ice:CallbackPtr类型的对象。 2)begin_xxxMethod，需要输入一个Ice:CallbackPtr对象。,2.6 完成通知,6.回调中的Cookie Ice:AsyncResultPtr begi

20、n_getName( Ice:Int number, const Ice:CallbackPtr Cookie对象在异步方法调用中用于begin_xxxMethod与end_xxxMethod之间传递信息。 应用场景： 用户在界面上点击产生大量的异步调用，当这些异步调用完成后，每个调用需要更新不同用户界面组件。在这种情景下，每个调用需要携带它所需要更新的组件信息才能完成操作。 在这种异步调用与回调函数之间需要交互的情况下，可以通过Cookie在应用中绑定相关信息。,2.6 完成通知,7.Cookie的编写 为了满足不同的应用场景，Cookie对象可以自定义。 Cookie编写的唯一要求就是要继

21、承Ice:LocalObject 示例： class Cookie : public Ice:LocalObject public: Cookie(WidgetHandle h) : _h(h) WidgetHandle getWidget() return _h; private: WidgetHandle _h; ; typedef IceUtil:Handle CookiePtr; / CookiePtr cookie1 = new Cookie(widgetHandle1); / Make cookie for call to getName(42); CookiePtr cookie

22、2 = new Cookie(widgetHandle2); / Invoke the getName operation with different cookies. e-begin_getName(99, getNameCB, cookie1); e-begin_getName(24, getNameCB, cookie2);,void MyCallback:getName(const Ice:AsyncResultPtr 1)在应用逻辑中创建了Cookie对象 2)将Cookie对象传递给begin_getName方法 3)在对应的回调方法中重新获得Cookie(需要向下转型) 4)从

23、Cookie对象中获得需要的信息,2.6 完成通知,8.类型安全回调函数方式的完成通知 (1)通用回调方法存在一些类型安全方面的隐患 在回调方法中调用end_xxxMethod前必须将代理对象向下转型为正确的代理对象类型 EmployeesPrx e = EmployeesPrx:uncheckedCast(r-getProxy(); 调用的end_xxxMethod必须与begin_xxxMethod配对 如果使用了Cookie，那么也需要将Cookie向下转型为正确的Cookie类型。 CookiePtr cookie = CookiePtr:dynamicCast(r-getCookie

24、(); 在调用end_xxxMethod方法时，必须要捕获所有可能的异常。否则无法获知调用在何时失败 Slice工具提供了一套Callback类API解决上述类型安全问题。,2.6 完成通知,8.类型安全回调函数方式的完成通知 （2）Slice工具与类型安全的回调 Slice工具在编译slice文件时，生成一系列相关的回调函数类别，在这些自动生成的代码中，将类型转换，调用end_xxxMethod等通用回调方法中的程式化方法以自动生成的方式代替手工编程。（C+中是以模板的方式实现不同回调函数的模板实例）,2.6 完成通知,8.类型安全回调函数方式的完成通知 （3）类型安全的回调函数实现要点 S

25、lice工具为slice定义的接口中的接口方法自动生成类型安全的回调类型。 自定义回调类型实现具体的回调操作，处理业务逻辑。 必须声明一个正常情况下处理回调的方法，该方法返回void类型，参数列表与slice定义一致。当slice返回类型不是void的时候，则第一个参数用于返回值，后续参数与slice定义中的参数列表一致。 必须声明一个异常情况下，处理异常的方法（无返回值，只有一个const Ice:Exception MyCallbackPtr cb = new MyCallback; Ice:CallbackPtr d = Ice:newCallback(cb, ,2.6 完成通知,10.

26、类型安全的Cookie调用方式 MyCallbackPtr cb = new MyCallback; Callback_Employees_getNamePtr getNameCB = newCallback_Employees_getName( cb, ,class MyCallback : public IceUtil:Shared public: void getNameCB(const string ,2.6 完成通知,11.类型安全回调方式使用总结 声明自定义回调对象实例 声明类型安全的回调对象实例 声明类型安全回调对象实例时，利用辅助函数构造回调对象实例 辅助函数中需要传递： 自定

27、义回调对象实例 回调方法 失败处理的回调方法,2.7 单路调用,1）通用的回调方式由于回调类别中回调函数具有输入参数， void callback_method(AsyncResultPtr,2.8 流量控制,1.流量控制的起因： 客户端并发请求非常频繁，服务端处理能力赶不上客户端发送异步请求的速度。客户端将请求发往客户端Ice运行时请求队列，队列不断增长，在不加流量控制的情况下，最终耗尽内存。 2.流量控制的功能： Ice提供了一套流量控制管理的API，应用了流量管理的api后，当客户端的请求数超过了限定的阈值，则阻塞客户端发起的新的操作请求直到队列中完成部分请求的处理，有剩余的空间容纳入队

28、的请求。,2.8 流量控制,4.流量控制回调函数定义 1）通用回调方式 void sent(const Ice:AsyncResult 调用情景1与情景2的判断由Ice运行时传递的sendSynchronously参数判断。,2.8 流量控制,3.流量控制的编程要点： 流量控制的基本思想是自定义流量控制的方法。 class MyCallback : public IceUtil:Shared public: void finished(const Ice:AsyncResultPtr 情景1：当Ice运行时将请求发送到客户端的发送端上后，由调用begin_xxxMethod的方法的线程接着调用

29、send方法。 情景2：当Ice运行时将请求入队时，则由另一线程在Ice运行时将队列中请求发送到发送端后，调用send方法。 以上2种情况调用send的const Ice:AsyncResultPtr Ice:ObjectPrx ice_batchDatagram() const; void ice_flushBatchRequests(); / . ; ,2.9 批量调用,3.显式调用与隐式调用 ： 显式同步调用： ice_flushBatchRequests 显式异步调用： begin_ice_flushBatchRequests end_ice_flushBatchRequests 隐式

30、调用： Ice.BatchAutoFlush=1（默认） Ice.MessageSizeMax=1MB(默认),2.10 AMI的并发,Ice运行时通常情况下由独立的线程调用异步回调方法中的回调函数。因此在回调函数中可以使用非递归锁而不怕出现死锁的情况。 但如果定义了流量控制回调函数，则如前所述的规则，在不同的情景下，有不同的线程选择规则。,2.11 AMI的限制,采用了AMI异步回调，则不能采用collocated optimization优化。否则抛出CollocationOptimizationException。 collocated optimization优化 参见ICE手册32.

31、21节,2.12 示例1,slice文件： #ifndef HELLO_ICE #define HELLO_ICE module Demo exception RequestCanceledException ; interface Hello / 注意元数据定义 ami, amd idempotent void sayHello(int delay) throws RequestCanceledException; void shutdown(); ; ; #endif,2.12 示例1,int AsyncClient:run(int argc, char* argv) if(argc 1)

32、 cerr propertyToProxy(Hello.Proxy); if(!hello) cerr argv0 : invalid proxy endl; return EXIT_FAILURE; menu(); CallbackPtr cb = new Callback(); char c;,do try cout ; cin c; if(c = i) hello-sayHello(0); else if(c = d) hello-begin_sayHello(5000, newCallback_Hello_sayHello(cb, ,3. AMD编程,3.1元数据标记 3.2 AMD语

33、言映射（C+） 3.3 AMD方式下的异常 3.4示例,3. AMD编程,回顾AMD概述 (1.3节） AMD: 使用 AMD 时，服务器可以接收一个请求，然后挂起其处理，以尽快释放分派线程。当处理恢复、结果已得出时，服务器要使用 Ice run time 提供的回调对象，显式地发送响应。 AMD 操作通常会把请求数据 （也就是，回调对象和 操作参数）放入队列 ，供应用的某个线程 （或线程池）随后处理用。这样，服务器就使分派线程的使用率降到了最低限度，能够高效地支持数千并发客户。,3. 2 AMD方式的语言映射,2. AMD的语言映射(C+) C+ 代码生成器为每个AMD 操作生成以下代码：

34、一个抽象的回调类:实现用它来通知 Ice run time，操作已完成。 类名规则:AMD_class_op 例如，对于在接口I 中定义的名叫foo 的操作 ，对应的类的名字是 AMD_I_foo。 void ice_response(); 服务器可以用 ice_response 方法报告操作已成功完成。如果操作的返回类型不是 void， ice_response 的第一个参数就是返回值。与操作的out 参数对应的参数跟在返回值后面，其次序是声明时的次序。 void ice_exception(const Ice:Exception ; 下面是为操作foo 生成的回调类： class AMD_

35、I_foo : public . public: void ice_response(Ice:Int, Ice:Long); void ice_exception(const Ice:Exception ,3. 3 AMD的异常传递,在两种处理上下文中， AMD 操作的逻辑实现可能需要报告异常：分派线程 ( 也就是，接收调用的线程)，以及响应线程 (也就是，发送响应的线程 )。 尽管建议你用回调对象来把所有异常报告给客户，实现抛出异常也是合法的，但只能在分派线程中抛出。 Ice run time 无法捕捉从响应线程抛出的异常；应用 的运行时环境将决定怎样处理这样的异常。因此，响应线程必须确保抓

36、住所有异常，并用回调对象发送适当的响应。否则，如果响应线程被未捕捉的异常终止，请求就可能不会完成，客户将无限期地等待响应。,3. 4 示例,Slice定义 interface Hello ami, amd idempotent void sayHello(int delay) throws RequestCanceledException; void shutdown(); ;,3. 4 示例,服务端声明文件代码： #include #include class HelloI : virtual public Demo:Hello public: HelloI(const WorkQueueP

37、tr,3. 4 示例,服务端实现文件代码： HelloI:HelloI(const WorkQueuePtr ,void HelloI:shutdown(const Ice:Current ,3. 4 示例,服务端实现文件代码： int AsyncServer:run(int argc, char* argv) if(argc 1) cerr createObjectAdapter(Hello); _workQueue = new WorkQueue(); Demo:HelloPtr hello = new HelloI(_workQueue); /定义服务端servrant adapter-add(hello, communicator()-stringToIdentity(“hello”)