活动对象

活动对象模式将执行与对象的成员函数解耦，每个对象会留在在自己的控制线程中。其目标是通过使用异步方法，处理调度器的请求，从而触发并发。维基百科：Active object。所以，这种模式也称为并发对象模式。

客户端的调用会转到代理，代理表现为活动对象的接口。服务提供活动对象的实现，并在单独的线程中运行。代理在运行时将客户端的调用转换为对服务的调用，调度程序将方法加入到激活列表中。调度器与服务在相同的线程中活动，并将方法调用从激活列表中取出，再将它们分派到相应的服务上。最后，客户端可以通过future从代理处获取最终的结果。

组件

活动对象模式由六个组件组成:

  1. 代理为活动对象的可访问方法提供接口。代理将触发激活列表的方法，并请求对象的构造。并且，代理和客户端运行在相同的线程中。
  2. 方法请求类定义了执行活动对象的接口。
  3. 激活列表的目标是维护挂起的请求，激活列表将客户端线程与活动对象线程解耦。代理对入队请求的进行处理，而调度器将请求移出队列。
  4. 调度器与代理可在不同的线程中运行。调度器会在活动对象的线程中运行，并决定接下来执行激活列表中的哪个请求。
  5. 可以通过服务实现活动对象，并在活动对象的线程中运行，服务也支持代理接口。
  6. future是由代理创造的，客户端可以从future上获取活动对象调用的结果。客户端可以安静等待结果，也可以对结果进行轮询。

下面的图片显示了消息的顺序。

活动对象 - 图1

代理

代理设计模式是《设计模式:可重用的面向对象软件的元素》中的经典模式，代理是其他对象的代表。典型的代理可以是远程代理CORBA、安全代理、虚拟代理或智能指针，如std::shared_ptr。每个代理会为它所代表的对象添加额外的功能。远程代理代表远程对象，并使客户端产生本地对象的错觉。安全代理通过对数据进行加密和解密，将不安全的连接转换为安全的连接。虚拟代理以惰性的方式封装对象的创建，智能指针将接管底层内存的生存期。

活动对象 - 图2

  • 代理具有与RealSubject相同的接口，用于管理引用，还有subject的生命周期。
  • 与Subject具有相同的接口，如代理和RealSubject。
  • RealSubject用于提供具体的功能。

关于代理模式的更多细节，可以参考Wikipedia页面。

优点和缺点

介绍Active Object模式的最小实现前，先了解一下它的优点和缺点。

  • 优点:
    • 同步只需要在活动对象的线程上进行，不需要在客户端的线程上进行。
    • 客户端(用户)和服务器(实现者)之间的解耦，同步的挑战则在实现者的一边。
    • 由于客户端为异步请求，所以系统的吞吐量提高了，从而调用处理密集型方法不会阻塞整个系统。
    • 调度器可以实现各种策略来执行挂起请求，因此可以按不同的顺序执行入队请求。
  • 缺点:
    • 如果请求的粒度太细，则活动对象模式(如代理、激活列表和调度器)的性能开销可能过大。
    • 由于调度器的调度策略和操作系统的调度互相影响，调试活动对象模式通常非常困难，尤其是以不同顺序执行请求的情况下。

具体实现

下面的示例展示了活动对象模式的简单实现。我没有定义一个请求，这应该由代理和服务实现。而且，当请求调度程序执行下一个请求时，服务应该只执行这个请求。

所涉及的类型为future<vector<future<pair<bool, int>>>>，这个类型的标识有点长。为了提高可读性，我使用了声明(第16 - 37行)。

  1. // activeObject.cpp
  3. #include <algorithm>
  4. #include <deque>
  5. #include <functional>
  6. #include <future>
  7. #include <iostream>
  8. #include <memory>
  9. #include <mutex>
  10. #include <numeric>
  11. #include <random>
  12. #include <thread>
  13. #include <utility>
  14. #include <vector>
  16. using std::async;
  17. using std::boolalpha;
  18. using std::cout;
  19. using std::deque;
  20. using std::distance;
  21. using std::endl;
  22. using std::for_each;
  23. using std::find_if;
  24. using std::future;
  25. using std::lock_guard;
  26. using std::make_move_iterator;
  27. using std::make_pair;
  28. using std::move;
  29. using std::mt19937;
  30. using std::mutex;
  31. using std::packaged_task;
  32. using std::pair;
  33. using std::random_device;
  34. using std::sort;
  35. using std::thread;
  36. using std::uniform_int_distribution;
  37. using std::vector;
  39. class IsPrime {
  40. public:
  41.   pair<bool, int> operator()(int i) {
  42.     for (int j = 2; j * j <= i; ++j) {
  43.       if (i % j == 0)return std::make_pair(false, i);
  44.     }
  45.     return std::make_pair(true, i);
  46.   }
  47. };
  49. class ActivaeObject {
  50. public:
  52.   future<pair<bool, int>> enqueueTask(int i) {
  53.     IsPrime isPrime;
  54.     packaged_task<pair<bool, int>(int)> newJob(isPrime);
  55.     auto isPrimeFuture = newJob.get_future();
  56.     auto pair = make_pair(move(newJob), i);
  57.     {
  58.       lock_guard<mutex> lockGuard(activationListMutex);
  59.       activationList.push_back(move(pair));
  60.     }
  61.     return isPrimeFuture;
  62.   }
  64.   void run() {
  65.     thread servant([this] {
  66.       while (!isEmpty()) {
  67.         auto myTask = dequeueTask();
  68.         myTask.first(myTask.second);
  69.       }
  70.       });
  71.     servant.join();
  72.   }
  74. private:
  76.   pair<packaged_task<pair<bool, int>(int)>, int> dequeueTask() {
  77.     lock_guard<mutex> lockGuard(activationListMutex);
  78.     auto myTask = std::move(activationList.front());
  79.     activationList.pop_front();
  80.     return myTask;
  81.   }
  83.   bool isEmpty() {
  84.     lock_guard<mutex> lockGuard(activationListMutex);
  85.     auto empty = activationList.empty();
  86.     return empty;
  87.   }
  89.   deque<pair<packaged_task<pair<bool, int>(int)>, int >> activationList;
  90.   mutex activationListMutex;
  91. };
  93. vector<int> getRandNumber(int number) {
  94.   random_device seed;
  95.   mt19937 engine(seed());
  96.   uniform_int_distribution<> dist(1000000, 1000000000);
  97.   vector<int> numbers;
  98.   for (long long i = 0; i < number; ++i) numbers.push_back(dist(engine));
  99.   return numbers;
  100. }
  102. future<vector<future<pair<bool, int>>>> getFutures(ActivaeObject& activeObject,
  103.   int numberPrimes) {
  104.   return async([&activeObject, numberPrimes] {
  105.     vector<future<pair<bool, int>>> futures;
  106.     auto randNumbers = getRandNumber(numberPrimes);
  107.     for (auto numb : randNumbers) {
  108.       futures.push_back(activeObject.enqueueTask(numb));
  109.     }
  110.     return futures;
  111.     });
  112. }
  115. int main() {
  117.   cout << boolalpha << endl;
  119.   ActivaeObject activeObject;
  121.   // a few clients enqueue work concurrently
  122.   auto client1 = getFutures(activeObject, 1998);
  123.   auto client2 = getFutures(activeObject, 2003);
  124.   auto client3 = getFutures(activeObject, 2011);
  125.   auto client4 = getFutures(activeObject, 2014);
  126.   auto client5 = getFutures(activeObject, 2017);
  128.   // give me the futures
  129.   auto futures = client1.get();
  130.   auto futures2 = client2.get();
  131.   auto futures3 = client3.get();
  132.   auto futures4 = client4.get();
  133.   auto futures5 = client5.get();
  135.   // put all futures together
  136.   futures.insert(futures.end(), make_move_iterator(futures2.begin()),
  137.     make_move_iterator(futures2.end()));
  139.   futures.insert(futures.end(), make_move_iterator(futures3.begin()),
  140.     make_move_iterator(futures3.end()));
  142.   futures.insert(futures.end(), make_move_iterator(futures4.begin()),
  143.     make_move_iterator(futures4.end()));
  145.   futures.insert(futures.end(), make_move_iterator(futures5.begin()),
  146.     make_move_iterator(futures5.end()));
  148.   // run the promises
  149.   activeObject.run();
  151.   // get the results from the futures
  152.   vector<pair<bool, int>> futResults;
  153.   futResults.reserve(futResults.size());
  154.   for (auto& fut : futures)futResults.push_back(fut.get());
  156.   sort(futResults.begin(), futResults.end());
  158.   // separate the primes from the non-primes
  159.   auto prIt = find_if(futResults.begin(), futResults.end(),
  160.     [](pair<bool, int>pa) {return pa.first == true; });
  162.   cout << "Number primes: " << distance(prIt, futResults.end()) << endl;
  163.   cout << "Primes: " << endl;
  164.   for_each(prIt, futResults.end(), [](auto p) {cout << p.second << " "; });
  166.   cout << "\n\n";
  168.   cout << "Number no primes: " << distance(futResults.begin(), prIt) << endl;
  169.   cout << "No primes: " << endl;
  170.   for_each(futResults.begin(), prIt, [](auto p) {cout << p.second << " "; });
  172.   cout << endl;
  174. }

示例的基本思想是，客户端可以在激活列表上并发地安排作业。线程的工作是确定哪些数是质数。激活列表是活动对象的一部分，而活动对象在一个单独的线程上进行入队操作，并且客户端可以在激活列表中查询作业的结果。

程序的详情：5个客户端通过getFutures将工作(第121 - 126行)入队到activeObject。numberPrimes中的数字是1000000到1000000000之间(第96行)的随机数，将这些数值放入vector<future<pair<bool, int>>中。future<pair<bool, int>持有一个bool和int对，其中bool表示int值是否是质数。再看看第108行：future .push_back(activeObject.enqueueTask(numb))。此调用将触发新作业进入激活列表的队列，所有对激活列表的调用都必须受到保护，这里激活列表是一个promise队列(第89行)：deque<pair<packaged_task<pair<bool, int>(int)>, int >>。

每个promise在调用执行函数对象IsPrime(第39 - 47行)时，会返回一个bool和int对。现在，工作包已经准备好了，开始计算吧。所有客户端在第129 - 133行中返回关联future的句柄，并把所有的future放在一起(第136 - 146行)，这样会使工作更加容易。第149行中的调用activeObject.run()启动执行。run(第64 - 72行)启动单独的线程，并执行promises(第68行)，直到执行完所有作业(第66行)。isEmpty(第83 - 87行)确定队列是否为空，dequeTask会返回一个新任务。通过在每个future上调用futResults.push_back(fut.get())(第154行)，所有结果都会推送到futResults上。第156行对成对的向量进行排序:vector<pair<bool, int>>。其余代码则是给出了计算结果，第159行中的迭代器prIt将第一个迭代器指向一个素数对。

程序打印素数数量为distance(prIt, futResults.end())(第162行)，并(第164行)逐一显示。

活动对象 - 图3

拓展阅读