异步消息块

项目
06/16/2023

代理库提供了多种消息块类型，使你能够以线程安全的方式在应用程序组件之间传播消息。这些消息块类型通常与 concurrency::send、concurrency::asend、concurrency::receive 和 concurrency::try_receive 等各种消息传递例程配合使用。有关代理库定义的消息传递例程的详细信息，请参阅消息传递函数。

章节

本主题包含以下各节：

源和目标
消息传播
消息块类型概述
unbounded_buffer 类
overwrite_buffer 类
single_assignment 类
call 类
transformer 类
choice 类
join 和 multitype_join 类
timer 类
消息筛选
消息保留

源和目标

源和目标是消息传递的两个重要参与方。源是指发送消息的通信终结点。目标是指接收消息的通信终结点。可将源视为从中读取的终结点，将目标视为要写入到的终结点。应用程序将源和目标连接在一起以形成消息传递网络。

代理库使用两个抽象类来表示源和目标：concurrency::ISource 和 concurrency::ITarget。充当源的消息块类型派生自 ISource；充当目标的消息块类型派生自 ITarget。充当源和目标的消息块类型派生自 ISource 和 ITarget。

[返回页首]

消息传播

消息传播是将消息从一个组件发送到另一个组件的行为。向消息块提供某个消息时，消息块可以接受、拒绝或推迟该消息。每种消息块类型以不同的方式存储和传输消息。例如，unbounded_buffer 类存储无限数量的消息，overwrite_buffer 类一次存储一条消息，而 transformer 类存储每条消息的已更改版本。本文档稍后将更详细地介绍这些消息块类型。

当消息块接受消息时，它可以选择性地执行工作，如果该消息块是源，则会将生成的消息传递给网络的另一个成员。消息块可以使用筛选器函数来拒绝它不想要接收的消息。本主题稍后的消息筛选部分将更详细地介绍筛选器。推迟消息的消息块可以保留该消息并在以后再使用它。本主题稍后的消息保留部分将更详细地介绍消息保留。

代理库允许消息块以异步或同步方式传递消息。当你以同步方式（例如，使用 send 函数）将某个消息传递到消息块时，运行时会阻止当前上下文，直到目标块接受或拒绝该消息。当你以异步方式（例如，使用 asend 函数）将某个消息传递到消息块时，运行时会将该消息提供给目标，如果目标接受消息，则运行时将计划一个异步任务，用于将该消息传播到接收方。运行时使用轻量级任务以协作方式传播消息。有关轻量级任务的详细信息，请参阅任务计划程序。

应用程序将源和目标连接在一起以形成消息传递网络。通常，你会链接网络并调用 send 或 asend 以将数据传递到网络。若要将源消息块连接到目标，请调用 concurrency::ISource::link_target 方法。若要断开源块与目标的连接，请调用 concurrency::ISource::unlink_target 方法。若要断开源块与其所有目标的连接，请调用 concurrency::ISource::unlink_targets 方法。当某种预定义的消息块类型超出范围或被销毁时，它会自动断开与任何目标块的连接。某些消息块类型会限制它们可以写入到的最大目标数。以下部分描述了适用于预定义消息块类型的限制。

[返回页首]

消息块类型概述

下表简要描述了重要消息块类型的作用。

unbounded_buffer
存储消息队列。

overwrite_buffer
存储一条可以多次写入并从中读取的消息。

single_assignment
存储一条可以一次写入并从中多次读取的消息。

call
收到消息时执行工作。

transformer
收到数据时执行工作，并将该工作的结果发送到另一个目标块。 transformer 类可以作用于不同的输入和输出类型。

choice
从一组源中选择第一条可用消息。

join 和 multitype join
等待从一组源中接收所有消息，然后将这些消息组合成一条消息，供另一个消息块使用。

timer
按固定的间隔向目标块发送消息。

这些消息块类型具有不同的特征，使其适用于不同的情况。下面是部分特征：

传播类型：消息块是充当数据源、数据接收方还是两者。
消息顺序：消息块是否保留发送或接收消息的原始顺序。每个预定义的消息块类型都会保留其发送或接收消息的原始顺序。
源计数：消息块可从中读取的最大源数。
目标计数：消息块可以写入到的最大目标数。

下表显示了这些特征与各种消息块类型的关系。

消息块类型	传播类型（源、目标或两者）	消息顺序（有序或无序）	源计数	目标计数
`unbounded_buffer`	推送、请求和匿名	已订购	无界限	无界限
`overwrite_buffer`	两者	已订购	无界限	无界限
`single_assignment`	两者	已订购	无界限	无界限
`call`	目标	已订购	无界限	不适用
`transformer`	推送、请求和匿名	已订购	无界限	1
`choice`	两者	已订购	10	1
`join`	两者	已订购	无界限	1
`multitype_join`	两者	已订购	10	1
`timer`	Source	不适用	不适用	1

以下部分更详细地介绍了消息块类型。

[返回页首]

unbounded_buffer 类

concurrency::unbounded_buffer 类表示通用异步消息传递结构。此类存储先进先出 (FIFO) 消息队列，此消息队列可由多个源写入或从多个目标读取。在目标收到来自 unbounded_buffer 对象的消息时，将从消息队列中删除此消息。因此，虽然一个 unbounded_buffer 对象可以具有多个目标，但只有一个目标将接收每条消息。需将多条消息传递给另一个组件，且该组件必须接收每条消息时，unbounded_buffer 类十分有用。

示例

以下示例使用基本结构来演示如何使用 unbounded_buffer 类。此示例将三个值发送到 unbounded_buffer 对象，然后从该对象读回这些值。

// unbounded_buffer-structure.cpp
// compile with: /EHsc
#include <agents.h>
#include <iostream>

using namespace concurrency;
using namespace std;

int wmain()
{
   // Create an unbounded_buffer object that works with
   // int data.
   unbounded_buffer<int> items;

   // Send a few items to the unbounded_buffer object.
   send(items, 33);
   send(items, 44);
   send(items, 55);

   // Read the items from the unbounded_buffer object and print
   // them to the console.
   wcout << receive(items) << endl;
   wcout << receive(items) << endl;
   wcout << receive(items) << endl;
}

该示例产生下面的输出：

有关演示如何使用 unbounded_buffer 类的完整示例，请参阅如何：实现各种生成者-使用者模式。

[返回页首]

overwrite_buffer 类

concurrency::overwrite_buffer 类与 unbounded_buffer 类类似，只不过 overwrite_buffer 对象仅存储一条消息。此外，当目标接收来自 overwrite_buffer 对象的消息时，不会从缓冲区中删除该消息。因此，多个目标将接收到该消息的副本。

如果你要将多条消息传递到另一个组件，而该组件只需要最新的值，则 overwrite_buffer 类很有用。需向多个组件广播消息时，此类也很有用。

示例

以下示例使用基本结构来演示如何使用 overwrite_buffer 类。此示例将三个值发送到 overwrite _buffer 对象，然后从该对象读取当前值三次。此示例类似于 unbounded_buffer 类的示例。但是，overwrite_buffer 类仅存储一条消息。此外，运行时不会在读取消息后从 overwrite_buffer 对象中删除该消息。

// overwrite_buffer-structure.cpp
// compile with: /EHsc
#include <agents.h>
#include <iostream>

using namespace concurrency;
using namespace std;

int wmain()
{
   // Create an overwrite_buffer object that works with
   // int data.
   overwrite_buffer<int> item;

   // Send a few items to the overwrite_buffer object.
   send(item, 33);
   send(item, 44);
   send(item, 55);

   // Read the current item from the overwrite_buffer object and print
   // it to the console three times.
   wcout << receive(item) << endl;
   wcout << receive(item) << endl;
   wcout << receive(item) << endl;
}

该示例产生下面的输出：

有关演示如何使用 overwrite_buffer 类的完整示例，请参阅如何：实现各种生成者-使用者模式。

[返回页首]

single_assignment 类

concurrency::single_assignment 类与 overwrite_buffer 类类似，只不过只能写入到 single_assignment 对象一次。与 overwrite_buffer 类相似，在目标收到来自 single_assignment 对象的消息时，不会从该目标删除此消息。因此，多个目标将接收到该消息的副本。如果你要向多个组件广播一条消息时，single_assignment 类很有用。

示例

以下示例使用基本结构来演示如何使用 single_assignment 类。此示例将三个值发送到 single_assignment 对象，然后从该对象读取当前值三次。此示例类似于 overwrite_buffer 类的示例。尽管 overwrite_buffer 和 single_assignment 类都存储一条消息，但只能写入到 single_assignment 类一次。

// single_assignment-structure.cpp
// compile with: /EHsc
#include <agents.h>
#include <iostream>

using namespace concurrency;
using namespace std;

int wmain()
{
   // Create an single_assignment object that works with
   // int data.
   single_assignment<int> item;

   // Send a few items to the single_assignment object.
   send(item, 33);
   send(item, 44);
   send(item, 55);

   // Read the current item from the single_assignment object and print
   // it to the console three times.
   wcout << receive(item) << endl;
   wcout << receive(item) << endl;
   wcout << receive(item) << endl;
}

该示例产生下面的输出：

有关演示如何使用 single_assignment 类的完整示例，请参阅演练：实现 Future。

[返回页首]

call 类

concurrency::call 类充当消息接收方，它在接收数据时执行工作函数。此工作函数可以是 lambda 表达式、函数对象或函数指针。 call 对象的行为与普通函数调用不同，因为它与向它发送消息的其他组件并行运行。如果 call 对象在接收消息时正在执行工作，则它会将该消息添加到队列。每个 call 对象按消息的接收顺序处理排队的消息。

示例

以下示例使用基本结构来演示如何使用 call 类。此示例创建一个 call 对象，它会将收到的每个值输出到控制台。然后，该示例将三个值发送到 call 对象。由于 call 对象在单独的线程上处理消息，因此此示例还使用了计数器变量和 event 对象来确保 call 对象在 wmain 函数返回之前处理所有消息。

// call-structure.cpp
// compile with: /EHsc
#include <agents.h>
#include <iostream>

using namespace concurrency;
using namespace std;

int wmain()
{
   // An event that is set when the call object receives all values.
   event received_all;

   // Counts the 
   long receive_count = 0L;
   long max_receive_count = 3L;

   // Create an call object that works with int data.
   call<int> target([&received_all,&receive_count,max_receive_count](int n) {
      // Print the value that the call object receives to the console.
      wcout << n << endl;
      
      // Set the event when all messages have been processed.
      if (++receive_count == max_receive_count)
         received_all.set();
   });

   // Send a few items to the call object.
   send(target, 33);
   send(target, 44);
   send(target, 55);

   // Wait for the call object to process all items.
   received_all.wait();
}

该示例产生下面的输出：

有关演示如何使用 call 类的完整示例，请参阅如何：为 call 和 transformer 类提供工作函数。

[返回页首]

transformer 类

concurrency::transformer 类既充当消息接收方，也充当消息发送方。 transformer 类与 call 类类似，因为它在接收数据时执行用户定义的工作函数。但是，transformer 类还会工作函数的结果发送到接收方对象。与 call 对象一样，transformer 对象与向其发送消息的其他组件并行运行。如果 transformer 对象在接收消息时正在执行工作，则它会将该消息添加到队列。每个 transformer 对象按消息的接收顺序处理其排队的消息。

transformer 类将其消息发送到一个目标。如果将构造函数中的 _PTarget 参数设置为 NULL，则稍后可以通过调用 concurrency::link_target 方法来指定目标。

与代理库提供的所有其他异步消息块类型不同，transformer 类可以作用于不同的输入和输出类型。这种将数据从一种类型转换为另一种类型的能力使得 transformer 类成为许多并发网络中的关键组件。此外，可以在 transformer 对象的工作函数中添加更高粒度的并行功能。

示例

以下示例使用基本结构来演示如何使用 transformer 类。此示例创建一个 transformer 对象，该对象将输入的每个 int 值乘以 0.33，以生成一个 double 值作为输出。然后，该示例从该 transformer 对象接收转换后的值并将其输出到控制台。

// transformer-structure.cpp
// compile with: /EHsc
#include <agents.h>
#include <iostream>

using namespace concurrency;
using namespace std;

int wmain()
{
   // Create an transformer object that receives int data and 
   // sends double data.
   transformer<int, double> third([](int n) {
      // Return one-third of the input value.
      return n * 0.33;
   });

   // Send a few items to the transformer object.
   send(third, 33);
   send(third, 44);
   send(third, 55);

   // Read the processed items from the transformer object and print
   // them to the console.
   wcout << receive(third) << endl;
   wcout << receive(third) << endl;
   wcout << receive(third) << endl;
}

该示例产生下面的输出：

10.8914.5218.15

有关演示如何使用 transformer 类的完整示例，请参阅如何：在数据管道中使用转换器。

[返回页首]

choice 类

concurrency::choice 类从一组源中选择第一条可用消息。 choice 类表示控制流机制而不是数据流机制（主题异步代理库介绍了数据流与控制流之间的差别）。

读取 choice 对象类似于在将 bWaitAll 参数设置为 FALSE 的情况下调用 Windows API 函数 WaitForMultipleObjects。但是，choice 类会将数据绑定到事件本身而不是外部同步对象。

通常，你会将 choice 类与 concurrency::receive 函数一起使用来驱动应用程序中的控制流。必须在具有不同类型的消息缓冲区中进行选择时，请使用 choice 类。必须在具有相同类型的消息缓冲区中进行选择时，请使用 single_assignment 类。

将源链接到 choice 对象的顺序非常重要，因为这可以确定选择了哪条消息。例如，假设你要将已包含一条消息的多个消息缓冲区链接到 choice 对象。 choice 对象将从它链接到的第一个源中选择该消息。链接所有源后，choice 对象会保留每个源接收消息的顺序。

示例

以下示例使用基本结构来演示如何使用 choice 类。此示例使用 concurrency::make_choice 函数创建一个 choice 对象，该对象在三个消息块中进行选择。然后，该示例计算各种斐波那契数并将每个结果存储在不同的消息块中。然后，该示例将基于首先完成的运算的消息输出到控制台。

// choice-structure.cpp
// compile with: /EHsc
#include <agents.h>
#include <ppl.h>
#include <iostream>

using namespace concurrency;
using namespace std;

// Computes the nth Fibonacci number.
// This function illustrates a lengthy operation and is therefore
// not optimized for performance.
int fibonacci(int n)
{
   if (n < 2)
      return n;
   return fibonacci(n-1) + fibonacci(n-2);
}

int wmain()
{
   // Although the following thee message blocks are written to one time only, 
   // this example illustrates the fact that the choice class works with 
   // different message block types.

   // Holds the 35th Fibonacci number.
   single_assignment<int> fib35;
   // Holds the 37th Fibonacci number.
   overwrite_buffer<int> fib37;
   // Holds half of the 42nd Fibonacci number.
   unbounded_buffer<double> half_of_fib42;   

   // Create a choice object that selects the first single_assignment 
   // object that receives a value.
   auto select_one = make_choice(&fib35, &fib37, &half_of_fib42);

   // Execute a few lengthy operations in parallel. Each operation sends its 
   // result to one of the single_assignment objects.
   parallel_invoke(
      [&fib35] { send(fib35, fibonacci(35)); },
      [&fib37] { send(fib37, fibonacci(37)); },
      [&half_of_fib42] { send(half_of_fib42, fibonacci(42) * 0.5); }
   );

   // Print a message that is based on the operation that finished first.
   switch (receive(select_one))
   {
   case 0:
      wcout << L"fib35 received its value first. Result = " 
            << receive(fib35) << endl;
      break;
   case 1:
      wcout << L"fib37 received its value first. Result = " 
            << receive(fib37) << endl;
      break;
   case 2:
      wcout << L"half_of_fib42 received its value first. Result = " 
            << receive(half_of_fib42) << endl;
      break;
   default:
      wcout << L"Unexpected." << endl;
      break;
   }
}

此示例产生以下示例输出：

fib35 received its value first. Result = 9227465

由于无法保证计算第 35 个斐波那契数的任务首先完成，因此此示例的输出可能有所不同。

此示例使用 concurrency::parallel_invoke 算法并行计算斐波那契数。有关 parallel_invoke 的详细信息，请参阅并行算法。

有关演示如何使用 choice 类的完整示例，请参阅如何：在已完成的任务中进行选择。

[返回页首]

join 和 multitype_join 类

concurrency::join 和 concurrency::multitype_join 类允许你等待一组源中的每个成员接收消息。 join 类作用于具有通用消息类型的源对象。 multitype_join 类作用于可具有不同消息类型的源对象。

读取 join 或 multitype_join 对象类似于在将 bWaitAll 参数设置为 TRUE 的情况下调用 Windows API 函数 WaitForMultipleObjects。但是，与 choice 对象一样，join 和 multitype_join 对象使用将数据绑定到事件本身而不是外部同步对象的事件机制。

读取 join 对象会生成 std::vector 对象。读取 multitype_join 对象会生成 std::tuple 对象。元素按照其相应源缓冲区链接到 join 或 multitype_join 对象的相同顺序出现在这些对象中。由于将源缓冲区链接到 join 或 multitype_join 对象的顺序与生成的 vector 或 tuple 对象中的元素顺序相关联，因此我们建议不要从联接中取消链接现有的源缓冲区。这样做可能会导致未指定的行为。

贪婪与非贪婪联接

join 和 multitype_join 类支持贪婪和非贪婪联接的概念。当消息可用时，贪婪联接会接受其每个源中的消息，直到所有消息都可用。非贪婪联接分两个阶段接收消息。首先，非贪婪联接会一直等到从其每个源为其提供消息。然后，在所有源消息都可用后，非贪婪联接会尝试保留其中的每条消息。如果它可以保留每条消息，则会使用所有消息并将其传播到目标。否则，它会释放或取消消息保留，并再次等待每个源接收消息。

贪婪联接比非贪婪联接的性能更好，因为它们立即接受消息。但是，在极少数情况下，贪婪联接会导致死锁。如果有多个联接包含一个或多个共享源对象，请使用非贪婪联接。

示例

以下示例使用基本结构来演示如何使用 join 类。此示例使用 concurrency::make_join 函数创建一个从三个 single_assignment 对象接收消息的 join 对象。此示例计算不同的斐波那契数，将每个结果存储在不同的 single_assignment 对象中，然后将 join 对象保存的每个结果输出到控制台。此示例类似于 choice 类的示例，不同之处在于，join 类会等待所有源消息块接收消息。

// join-structure.cpp
// compile with: /EHsc
#include <agents.h>
#include <ppl.h>
#include <iostream>

using namespace concurrency;
using namespace std;

// Computes the nth Fibonacci number.
// This function illustrates a lengthy operation and is therefore
// not optimized for performance.
int fibonacci(int n)
{
   if (n < 2)
      return n;
   return fibonacci(n-1) + fibonacci(n-2);
}

int wmain()
{
   // Holds the 35th Fibonacci number.
   single_assignment<int> fib35;
   // Holds the 37th Fibonacci number.
   single_assignment<int> fib37;
   // Holds half of the 42nd Fibonacci number.
   single_assignment<double> half_of_fib42;   

   // Create a join object that selects the values from each of the
   // single_assignment objects.
   auto join_all = make_join(&fib35, &fib37, &half_of_fib42);

   // Execute a few lengthy operations in parallel. Each operation sends its 
   // result to one of the single_assignment objects.
   parallel_invoke(
      [&fib35] { send(fib35, fibonacci(35)); },
      [&fib37] { send(fib37, fibonacci(37)); },
      [&half_of_fib42] { send(half_of_fib42, fibonacci(42) * 0.5); }
   );

   auto result = receive(join_all);
   wcout << L"fib35 = " << get<0>(result) << endl;
   wcout << L"fib37 = " << get<1>(result) << endl;
   wcout << L"half_of_fib42 = " << get<2>(result) << endl;
}

该示例产生下面的输出：

fib35 = 9227465fib37 = 24157817half_of_fib42 = 1.33957e+008

此示例使用 concurrency::parallel_invoke 算法并行计算斐波那契数。有关 parallel_invoke 的详细信息，请参阅并行算法。

有关演示如何使用 join 类的完整示例，请参阅如何：在已完成的任务中进行选择和演练：使用联接来防止死锁。

[返回页首]

timer 类

concurrency::timer 类充当消息源。 timer 对象在指定的时间段过后向目标发送消息。必须延迟发送消息或要按固定的间隔发送消息时，timer 类很有用。

timer 类将其消息仅发送到一个目标。如果将构造函数中的 _PTarget 参数设置为 NULL，则稍后可以通过调用 concurrency::ISource::link_target 方法来指定目标。

timer 对象可以重复或不重复。若要创建重复计时器，请在调用构造函数时为 _Repeating 参数传递 true。否则，请为 _Repeating 参数传递 false 以创建非重复计时器。如果计时器是重复的，则它会在每个间隔后向其目标发送相同的消息。

代理库创建处于未启动状态的 timer 对象。若要启动 timer 对象，请调用 concurrency::timer::start 方法。若要停止 timer 对象，请销毁该对象或调用 concurrency::timer::stop 方法。若要暂停重复计时器，请调用 concurrency::timer::pause 方法。

示例

以下示例使用基本结构来演示如何使用 timer 类。该示例使用 timer 和 call 对象来报告长时间运行的操作的进度。

// timer-structure.cpp
// compile with: /EHsc
#include <agents.h>
#include <iostream>

using namespace concurrency;
using namespace std;

// Computes the nth Fibonacci number.
// This function illustrates a lengthy operation and is therefore
// not optimized for performance.
int fibonacci(int n)
{
   if (n < 2)
      return n;
   return fibonacci(n-1) + fibonacci(n-2);
}

int wmain()
{
   // Create a call object that prints characters that it receives 
   // to the console.
   call<wchar_t> print_character([](wchar_t c) {
      wcout << c;
   });

   // Create a timer object that sends the period (.) character to 
   // the call object every 100 milliseconds.
   timer<wchar_t> progress_timer(100u, L'.', &print_character, true);

   // Start the timer.
   wcout << L"Computing fib(42)";
   progress_timer.start();

   // Compute the 42nd Fibonacci number.
   int fib42 = fibonacci(42);

   // Stop the timer and print the result.
   progress_timer.stop();
   wcout << endl << L"result is " << fib42 << endl;
}

此示例产生以下示例输出：

Computing fib(42)..................................................result is 267914296

有关演示如何使用 timer 类的完整示例，请参阅如何：按固定的间隔发送消息。

[返回页首]

消息筛选

创建消息块对象时，可以提供一个筛选器函数来确定消息块是接受还是拒绝消息。筛选器函数是保证消息块仅接收特定值的有效方式。

以下示例演示如何创建一个使用筛选器函数来仅接受偶数的 unbounded_buffer 对象。 unbounded_buffer 对象会拒绝奇数，因此不会将奇数传播到其目标块。

// filter-function.cpp
// compile with: /EHsc
#include <agents.h>
#include <iostream>

using namespace concurrency;
using namespace std;

int wmain()
{
   // Create an unbounded_buffer object that uses a filter
   // function to accept only even numbers.
   unbounded_buffer<int> accept_evens(
      [](int n) {
         return (n%2) == 0;
      });

   // Send a few values to the unbounded_buffer object.
   unsigned int accept_count = 0;
   for (int i = 0; i < 10; ++i)
   {
      // The asend function returns true only if the target
      // accepts the message. This enables us to determine
      // how many elements are stored in the unbounded_buffer
      // object.
      if (asend(accept_evens, i))
      {
         ++accept_count;
      }
   }

   // Print to the console each value that is stored in the 
   // unbounded_buffer object. The unbounded_buffer object should
   // contain only even numbers.
   while (accept_count > 0)
   {
      wcout << receive(accept_evens) << L' ';
      --accept_count;
   }
}

该示例产生下面的输出：

0 2 4 6 8

筛选器函数可以是 lambda 函数、函数指针或函数对象。每个筛选器函数采用以下格式之一。

bool (T)
bool (T const &)

为了消除不必要的数据复制，需要按值传播聚合类型时，请使用第二种格式。

消息筛选支持数据流编程模型，其中的组件在接收数据时会执行计算。有关使用筛选器函数控制消息传递网络中的数据流的示例，请参阅如何：使用消息块筛选器、演练：创建数据流代理和演练：创建图像处理网络。

[返回页首]

消息保留

消息保留使消息块能够保留消息供以后使用。通常，消息保留不是直接使用的。但是，了解消息预留有助于更好地了解某种预定义消息块类型的行为。

考虑非贪婪和贪婪联接。这两种联接都使用消息保留来保留消息供以后使用。如前所述，非贪婪联接分两个阶段接收消息。在第一阶段，非贪婪 join 对象等待其每个源接收消息。然后，非贪婪联接尝试保留其中的每个消息。如果它可以保留每条消息，则会使用所有消息并将其传播到目标。否则，它会释放或取消消息保留，并再次等待每个源接收消息。

贪婪联接也从多个源读取输入消息，它使用消息保留来读取其他消息，同时等待从每个源接收消息。例如，考虑一个从消息块 A 和 B 接收消息的贪婪联接。如果贪婪联接从 B 收到了两条消息，但尚未收到来自 A 的消息，则贪婪联接会保存来自 B 的第二条消息的唯一消息标识符。在贪婪联接收到来自 A 的消息并传播这些消息后，它会使用保存的消息标识符来查看来自 B 的第二条消息是否仍然可用。

当你实现自己的自定义消息块类型时，可以使用消息保留。有关如何创建自定义消息块类型的示例，请参阅演练：创建自定义消息块。

[返回页首]

另请参阅

异步代理库

异步消息块

章节

源和目标

消息传播

消息块类型概述

unbounded_buffer 类

示例

overwrite_buffer 类

示例

single_assignment 类

示例

call 类

示例

transformer 类

示例

choice 类

示例

join 和 multitype_join 类

贪婪与非贪婪联接

示例

timer 类

示例

消息筛选

消息保留

另请参阅

反馈

反馈

其他资源