一概述

Streams框架是管道和过滤构架模式的一种实现，主要应用于处理数据流的系统。其实现以Task框架为基础。Task框架有两个特性非常适用于Streams框架：一是Task框架可用于创建独立线程的并发环境，这适合应用于ACE Streams框架中的主动过滤器；二是Task框架有统一的数据传输结果——消息队列，这适用于Streams框架中的管道。

二ACE_Task类

这里主要介绍与Streams框架相关的部分。

成员变量

 Task_T.h

 class ACE_Task : public ACE_Task_Base

 {

 /// Queue of messages on the ACE_Task..

   ACE_Message_Queue<ACE_SYNCH_USE, TIME_POLICY> *msg_queue_;

   /// true if should delete Message_Queue, false otherwise.

   bool delete_msg_queue_;

   /// Back-pointer to the enclosing module.

   ACE_Module<ACE_SYNCH_USE, TIME_POLICY> *mod_;

   /// Pointer to adjacent ACE_Task.

   ACE_Task<ACE_SYNCH_USE, TIME_POLICY> *next_;

 }

put函数

在ACE_Task框架中，put函数没有实际作用，在默认情况下，该函数没有执行任何操作，仅仅返回0.但是在Streams框架中，put函数与put_next函数结合起来可以实现数据在过滤器间的传输。如果put函数将数据保存在消息队列中，通过独立的线程来处理这些消息，那么它将成为一个主动过滤器；反之，如果put函数直接对数据进行处理，然后交给下一个过滤器，那么它就是一个被动过滤器。

 Task.cpp

 /// Default ACE_Task put routine.

 int

 ACE_Task_Base::put (ACE_Message_Block *, ACE_Time_Value *)

 {

   ACE_TRACE ("ACE_Task_Base::put");

   return ;

 }

put_next函数

如果在数据处理流水线有下一个过滤器，那么put_next函数用于将数据交给下一个过滤器处理。如下：

 Task_T.inl

 // Transfer msg to the next ACE_Task.

 template <ACE_SYNCH_DECL, class TIME_POLICY> ACE_INLINE int

 ACE_Task<ACE_SYNCH_USE, TIME_POLICY>::put_next (ACE_Message_Block *msg, ACE_Time_Value *tv)

 {

   ACE_TRACE ("ACE_Task<ACE_SYNCH_USE, TIME_POLICY>::put_next");

   return this->next_ ==  ? - : this->next_->put (msg, tv);

 }

next_指向的是有序过滤器的下一个过滤器。通过put_next函数，可以将数据交给下一个过滤器处理。

Streams框架应用示例

在这个示例中，我们将一个数据流的处理分为4步，在Streams框架中，将每个处理步骤称为一个Module：

  • Logrec_Reader：从文件中读取记录，然后交给下一个步骤。
  • Logrec_Timer：在记录尾部加上“format_data”
  • Logrec_Suffix：在记录尾部加上一个后缀——suffix
  • Logrec_Write：将记录显示在终端上

Logrec_Reader类

其是ACE_Task的子类，是一个主动对象类，有独立的控制线程，线程处理函数是svc。在Streams框架中，Logrec_Reader类是一个主动过滤器，代码如下：

 class Logrec_Reader : public ACE_Task<ACE_MT_SYNCH>

 {

 private:

     ifstream fin;  //标准输入流

 public:

     Logrec_Reader(ACE_TString logfile)

     {

         fin.open(logfile.c_str()); //ACE_TString.c_str() 转换为char

     }

     virtual int open (void *)

     {

         return activate();

     }

     virtual int svc()

     {

         ACE_Message_Block *mblk;

         int len = ;

         const int LineSize = ;

         char file_buf[LineSize];

         while(!fin.eof())

         {

             fin.getline(file_buf, LineSize);

             len = ACE_OS::strlen(file_buf);

             ACE_NEW_RETURN(mblk, ACE_Message_Block(len + ), );

             if (file_buf[len - ] == '\r')

             {

                 len = len - ;

             }

             mblk->copy(file_buf, len);

             // 通过put_next函数，将消息传递给下一个过滤器

             put_next(mblk);

         }

         //构造一个MB_STOP消息

         ACE_NEW_RETURN(mblk, ACE_Message_Block (, ACE_Message_Block::MB_STOP), );

         put_next(mblk);

         fin.close();

         ACE_DEBUG((LM_DEBUG, "read svc return. \n"));

         return ;

     }

 };

Logrec_Timer类

也是ACE_Task的子类，但是它不是主动对象类，没有创建独立线程。其实现了put函数，这个函数被它的上一个过滤器（Logrec_Reader）调用，并且数据直接在这个函数中处理。

这里for循环用于处理消息链表，在示例中并没有使用链表因此for循环只会执行一次。

 class Logrec_Timer : public ACE_Task<ACE_SYNCH>

 {

 private:

     void format_data(ACE_Message_Block *mblk)

     {

         char *msg = mblk->data_block()->base();

         strcat(msg, "format_data");

     }

 public:

     virtual int put(ACE_Message_Block *mblk, ACE_Time_Value *)

     {

         for (ACE_Message_Block *temp = mblk;

             temp != ; temp = temp->cont())

         {

             if (temp->msg_type() != ACE_Message_Block::MB_STOP)

             {

                 format_data(temp);

             }

         }

         return put_next(mblk);

     }

 };

Logrec_Suffix类

类似Logrec_Timer

 class Logrec_Suffix : public ACE_Task<ACE_SYNCH>

 {

 public:

     void suffix(ACE_Message_Block *mblk)

     {

         char *msg = mblk->data_block()->base();

         strcat(msg, "suffix\n");

     }

     virtual int put(ACE_Message_Block *mblk, ACE_Time_Value *)

     {

         for (ACE_Message_Block *temp = mblk;

             temp != ; temp = temp->cont())

         {

             if (temp->msg_type() != ACE_Message_Block::MB_STOP)

             {

                 suffix(temp);

             }

         }

         return put_next(mblk);

     }

 };

Logrec_Write类

这里put函数由上一个过滤器（Logrec_Suffix）调用，其并没有对数据进行实际处理，只是将数据放入队列中，由线程独立处理。

 class Logrec_Write : public ACE_Task<ACE_SYNCH>

 {

 public:

     virtual int open(void*)

     {

         ACE_DEBUG((LM_DEBUG, "Logrec_Writer.\n"));

         return activate();

     }

     virtual int put (ACE_Message_Block *mblk, ACE_Time_Value *to)

     {

         return putq(mblk, to);

     }

     virtual int svc()

     {

         int stop = ;

         for (ACE_Message_Block *mb; !stop && getq(mb) != -;)

         {

             if (mb->msg_type() == ACE_Message_Block::MB_STOP)

             {

                 stop = ;

             }

             else{

                 ACE_DEBUG((LM_DEBUG, "%s", mb->base()));

             }

             put_next(mb);

         }

         return ;

     }

 };

Main

这里讲ACE_Module放入Streams中，ACE_Module才是真正数据处理的Module。ACE_Streams类有两个数据成员：stream_head_和stream_tail_，它们指向ACE_Module链表的首尾。对于每个Streams，其默认带有首尾两个Module，而后可以通过push将数据处理的Module放入执行链表中。每个Module包含两个Task，分别为读Task和写Task。在示例中仅注册了写Task，这些Module和Task通过next指针构成一个有序的串。

这里注意push有顺序要求，最后push即栈顶的为先执行的Module。

 int ACE_TMAIN(int argc, ACE_TCHAR *argv[])

 {

     if (argc != )

     {

         ACE_ERROR_RETURN((LM_ERROR, "usage: %s logfile\n", argv[]), );

     }

     ACE_TString logfile (argv[]);

     ACE_Stream<ACE_SYNCH> stream;

     ACE_Module<ACE_MT_SYNCH> *module[];

     module[] = new ACE_Module<ACE_MT_SYNCH>("Reader", new Logrec_Reader(logfile), , , ACE_Module<ACE_SYNCH>::M_DELETE_READER);

     module[] = new ACE_Module<ACE_MT_SYNCH>("Formatter", new Logrec_Timer(), , , ACE_Module<ACE_SYNCH>::M_DELETE_READER);

     module[] = new ACE_Module<ACE_MT_SYNCH>("Separator", new Logrec_Suffix(), , , ACE_Module<ACE_SYNCH>::M_DELETE_READER);

     module[] = new ACE_Module<ACE_MT_SYNCH>("Writer", new Logrec_Write(), , , ACE_Module<ACE_SYNCH>::M_DELETE_READER);

     for ( int i = ; i >= ; --i )

     {

         if (stream.push(module[i]) == -)

         {

             ACE_ERROR_RETURN((LM_ERROR, "push %s module into stream failed.\n", module[i]->name()), );

         }

         ACE_DEBUG((LM_DEBUG, "push %s module into stream success. \n", module[i]->name()));

     }

     ACE_Thread_Manager::instance()->wait();

 }

附完整代码及结果：https://github.com/ShiningZhang/ACE_Learning/tree/master/streams

 /*************************************************************************

     > File Name: logrec.cpp

     > Author:

     > Mail:

     > Created Time: Fri 13 Oct 2017 04:19:39 PM CST

  ************************************************************************/

  #include <fstream>

  #include <ace/Synch.h>

  #include <ace/Task.h>

  #include <ace/Message_Block.h>

  #include <ace/Stream.h>

  #include "ace/Thread_Manager.h"

  #include <ace/Time_Value.h>

  #include <ace/Module.h>

 using namespace std;

 class Logrec_Reader : public ACE_Task<ACE_MT_SYNCH>

 {

 private:

     ifstream fin;  //标准输入流

 public:

     Logrec_Reader(ACE_TString logfile)

     {

         fin.open(logfile.c_str()); //ACE_TString.c_str() 转换为char

     }

     virtual int open (void *)

     {

         return activate();

     }

     virtual int svc()

     {

         ACE_Message_Block *mblk;

         int len = ;

         const int LineSize = ;

         char file_buf[LineSize];

         while(!fin.eof())

         {

             fin.getline(file_buf, LineSize);

             len = ACE_OS::strlen(file_buf);

             ACE_NEW_RETURN(mblk, ACE_Message_Block(len + ), );

             if (file_buf[len - ] == '\r')

             {

                 len = len - ;

             }

             mblk->copy(file_buf, len);

             // 通过put_next函数，将消息传递给下一个过滤器

             put_next(mblk);

         }

         //构造一个MB_STOP消息

         ACE_NEW_RETURN(mblk, ACE_Message_Block (, ACE_Message_Block::MB_STOP), );

         put_next(mblk);

         fin.close();

         ACE_DEBUG((LM_DEBUG, "read svc return. \n"));

         return ;

     }

 };

 class Logrec_Timer : public ACE_Task<ACE_SYNCH>

 {

 private:

     void format_data(ACE_Message_Block *mblk)

     {

         char *msg = mblk->data_block()->base();

         strcat(msg, "format_data");

     }

 public:

     virtual int put(ACE_Message_Block *mblk, ACE_Time_Value *)

     {

         for (ACE_Message_Block *temp = mblk;

             temp != ; temp = temp->cont())

         {

             if (temp->msg_type() != ACE_Message_Block::MB_STOP)

             {

                 format_data(temp);

             }

         }

         return put_next(mblk);

     }

 };

 class Logrec_Suffix : public ACE_Task<ACE_SYNCH>

 {

 public:

     void suffix(ACE_Message_Block *mblk)

     {

         char *msg = mblk->data_block()->base();

         strcat(msg, "suffix\n");

     }

     virtual int put(ACE_Message_Block *mblk, ACE_Time_Value *)

     {

         for (ACE_Message_Block *temp = mblk;

             temp != ; temp = temp->cont())

         {

             if (temp->msg_type() != ACE_Message_Block::MB_STOP)

             {

                 suffix(temp);

             }

         }

         return put_next(mblk);

     }

 };

 class Logrec_Write : public ACE_Task<ACE_SYNCH>

 {

 public:

     virtual int open(void*)

     {

         ACE_DEBUG((LM_DEBUG, "Logrec_Writer.\n"));

         return activate();

     }

     virtual int put (ACE_Message_Block *mblk, ACE_Time_Value *to)

     {

         return putq(mblk, to);

     }

     virtual int svc()

     {

         int stop = ;

         for (ACE_Message_Block *mb; !stop && getq(mb) != -;)

         {

             if (mb->msg_type() == ACE_Message_Block::MB_STOP)

             {

                 stop = ;

             }

             else{

                 ACE_DEBUG((LM_DEBUG, "%s", mb->base()));

             }

             put_next(mb);

         }

         return ;

     }

 };

 int ACE_TMAIN(int argc, ACE_TCHAR *argv[])

 {

     if (argc != )

     {

         ACE_ERROR_RETURN((LM_ERROR, "usage: %s logfile\n", argv[]), );

     }

     ACE_TString logfile (argv[]);

     ACE_Stream<ACE_SYNCH> stream;

     ACE_Module<ACE_MT_SYNCH> *module[];

     module[] = new ACE_Module<ACE_MT_SYNCH>("Reader", new Logrec_Reader(logfile), , , ACE_Module<ACE_SYNCH>::M_DELETE_READER);

     module[] = new ACE_Module<ACE_MT_SYNCH>("Formatter", new Logrec_Timer(), , , ACE_Module<ACE_SYNCH>::M_DELETE_READER);

     module[] = new ACE_Module<ACE_MT_SYNCH>("Separator", new Logrec_Suffix(), , , ACE_Module<ACE_SYNCH>::M_DELETE_READER);

     module[] = new ACE_Module<ACE_MT_SYNCH>("Writer", new Logrec_Write(), , , ACE_Module<ACE_SYNCH>::M_DELETE_READER);

     for ( int i = ; i >= ; --i )

     {

         if (stream.push(module[i]) == -)

         {

             ACE_ERROR_RETURN((LM_ERROR, "push %s module into stream failed.\n", module[i]->name()), );

         }

         ACE_DEBUG((LM_DEBUG, "push %s module into stream success. \n", module[i]->name()));

     }

     ACE_Thread_Manager::instance()->wait();

 }

test：

最新文章

  1. P4基函数
  2. js 点击默认另存 ，不是打开 Blob 操作
  3. C#高级编程（第8版）
  4. [未完成]关于DOM的总结
  5. ubuntu下安装opencv 2.4.9 脚本，支持摄像头和cuda
  6. MySQL将表a中查询的数据插入到表b中
  7. hdu 4864 Task---2014 Multi-University Training Contest 1
  8. response.sendRedirect 报 java.lang.IllegalStateException 异常的解决思路
  9. hdu--1072--Nightmare（bfs回溯）
  10. 通俗易懂的信息熵与信息增益（IE, Information Entropy; IG, Information Gain）
  11. Linux下的crontab定时、执行任务命令详解 oracle 自动备份
  12. Oracle 的几种循环方式介绍
  13. php 学习资料
  14. WPF 和 百度 eChart 交互
  15. 【LOJ】#2071. 「JSOI2016」最佳团体
  16. cloudera安装笔记
  17. windows“启动”对应的路径
  18. iOS开发用如何用类&quot;SKStoreProductViewController&quot;跳转AppStore点赞评分？
  19. css属性之flex属性
  20. MySQL学习记录一

热门文章

  1. poj 2104 静态主席树
  2. 9.3n+1问题
  3. HDU 4611 Balls Rearrangement 数学
  4. bzoj 1789: [Ahoi2008]Necklace Y型项链 贪心
  5. Webpack使用指南
  6. Ubuntu 16.09开启iptables的日志实现调试
  7. mysql中如何比较日期
  8. VS2010 + IDA SDK 搭建IDA Plugin开发环境
  9. FIS3中使用less
  10. 分享一个基于 Node.js 的 Web 开发框架 - Nokitjs