CSP是用于对并发对象之间的复杂交互进行建模的范例。 使用CSP的主要优点之一是能够精确地指定和验证程序每个阶段涉及的对象的行为。 CSP的理论和实践对并发设计和编程领域产生了深远的影响。 它是occam等编程语言的基础，并且对其他语言（例如Ada）的设计产生了影响。 正如本文第1部分中简要讨论的那样，CSP对Java开发人员来说也很有价值，因为它适用于Java平台上安全且优雅的多线程编程。

在我的Java平台上的CSP编程的三部分的第二部分中，我将重点介绍CSP的理论和实践，尤其是它适用于Java语言中的多线程编程。 我首先概述了CSP的理论，然后向您介绍了基于CSP设计的基于Java的JCSP库实现。

CSP基础

CSP中的基本构造是过程以及它们之间的各种通信形式。 CSP中的所有内容都是一个过程，甚至是（子）过程的网络。 但是，进程之间没有直接的交互-所有交互仅通过CSP同步对象发生，例如进程组订阅的通信通道和事件屏障。

CSP 流程与典型的Java对象的不同之处在于，封装在流程组件中的数据和用于操纵该数据的算法都是私有的。 也就是说，进程没有外部可调用的方法（当然，必须调用一种方法来启动进程），算法只能由进程在其自己的控制线程中执行。 如果将此与Java语言中的方法调用进行对比，则可以立即看到CSP如何消除对显式锁定的需求。

在Java语言中，在对象上调用的方法始终在调用者的线程中运行。 特定的控制线程通过系统中的多个对象来工作。 在大多数情况下，对象没有自己的生命-它们只是在运行线程调用对象上的方法时才短暂地激活。 因此，不同的执行线程最终可能会尝试同时在同一对象上调用相同的方法，如第1部分中所述 。 显然，这在CSP中永远不可能发生。

沟通渠道和过程网络

进程间通信的最简单机制是跨通道读取和写入数据。 CSP中的基本通道构造是同步的和点对点的 ； 也就是说，不包含内部缓冲并将一个进程连接到另一个进程。 从此基本通道开始，还可以构造多个读取器/写入器通道（即，一对一，任意对一和任意对任意）。

CSP中的流程构成了复杂系统的基本构建模块-一个流程可以与一个或多个其他流程（所有都设置为并行执行）连接起来，形成一个流程网络。 该网络本身可以被认为是一个过程，可以与其他过程（本身就是网络）或其他过程递归地组合成无数复杂的安排，这些安排可以最好地解决当前的问题。

单独考虑，进程只是一个仅与外部I / O设备交互的独立串行程序。 该程序无需担心I / O通道另一端的进程的存在或性质。

CSP的理论已在许多基于Java的框架中实现，包括Java的通信顺序过程（JCSP）库。

JCSP库

JCSP库是由英国坎特伯雷肯特大学的Peter Welch教授和Paul Austin开发的（请参阅参考资料 ）。 在本文的其余大部分内容中，我将重点介绍如何在JCSP中实现CSP概念。 因为Java语言没有对CSP构造提供原生支持，JCSP库内部使用本地并发构建语言不支持，如synchronized ， wait和notify 。 为了帮助您准确了解JCSP的工作原理，我将根据这些Java构造说明一些JCSP库类的内部实现。

请注意，以下各节中的示例基于和/或派生自JCSP库和/或JCSP主页上的演示幻灯片的javadocs中记录的示例。

JCSP中的流程

在JCSP中，过程本质上是实现CSProcess接口的类。 清单1显示了该接口：

清单1. CSProcess接口

package jcsp.lang;public interface CSProcess
{public void run();
}

注意， CSProcess接口看起来与Java语言的Runnable接口完全一样，并且起着类似的作用。 尽管目前使用标准Java API实现JCSP，但不一定如此，将来也可能不是。 因此，在JCSP中不直接使用Runnable接口。

JCSP定义了两个用于读取和写入通道的接口。 从通道读取的接口称为ChannelInput ，由一个名为read()方法组成。 在实现ChannelInput接口的对象上调用此方法的进程将阻塞，直到该对象在通道另一端的进程实际写到通道上为止。 一旦此类对象在通道上可用，它就会返回到调用过程。 同样， ChannelOutput接口由一个称为write(Object o)方法组成。 在实现ChannelOutput接口的对象上调用此方法的过程将阻塞，直到该对象被通道接受为止。 如前所述，最简单的通道类型没有任何缓冲，因此，直到其他（读取）端的进程调用read() ，它才会接受该对象。

从这里开始，我将使用代码示例来演示这些以及其他JCSP构造如何工作。 在清单2中，您可以看到一个非常简单的过程，该过程输出1到100之间的所有偶数整数：

清单2.生成1到100之间的偶数整数的过程

import jcsp.lang.*;public class SendEvenIntsProcess implements CSProcess
{private ChannelOutput out;public SendEvenIntsProcess(ChannelOutput out){this.out = out;}public void run(){for (int i = 2; i <= 100; i = i + 2){out.write (new Integer (i));}}
}

必须有一个与每个写作过程相对应的阅读过程。 如果没有这样的过程，则会在ChannelOutput对象out上的第一次写操作之后立即导致SendEvenIntsProcess无限期地阻塞。 清单3显示了一个简单的读取过程，该过程与清单2中所示的写入过程相对应：

清单3.相应的消费者流程

import jcsp.lang.*;public class ReadEvenIntsProcess implements CSProcess
{private ChannelInput in;public ReadEvenIntsProcess(ChannelInput in){this.in = in;}public void run(){while (true){Integer d = (Integer)in.read();System.out.println("Read: " + d.intValue());}}
}

JCSP中的频道

至此，我只有两个独立的过程。 我的下一步是使用公共通道作为共享同步机制将它们连接起来，然后将它们逐个踢开。 通道接口是JCSP的ChannelInput和ChannelOutput接口的子接口，并且是用于读写对象的公共接口。 该接口有许多可能的实现，如下所述：

• 顾名思义，类One2OneChannel实现了“单个One2OneChannel器-单个读取器”类型的通道。
• 类One2AnyChannel实现一个“单写多读”对象通道。 （注意：这不是广播机制，因为多个阅读器实际上会相互竞争以从频道中进行阅读；在任何给定时间，只有一个阅读器可以与作者一起使用该频道。）
• 类Any2OneChannel实现了一个“多作者-单读者”对象通道。 与上述情况一样，写入过程相互竞争以使用通道。 在任何给定时间，只有读者和多个作者之一才能实际使用该频道。
• 类Any2AnyChannel实现了“多作者-多读者”对象通道。 阅读过程与写作过程相互竞争以使用渠道。 实际上，任何时候只有一位读者和一位作家可以实际使用该频道。

在清单3的示例中，我只有一个One2OneChannel器和一个阅读器进程，因此One2OneChannel类就足够了。 清单4中显示了驱动程序的示例代码：

清单4.驱动程序

import jcsp.lang.*;public class DriverProgram
{public static void main(String[] args){One2OneChannel chan = new One2OneChannel();new Parallel(new CSProcess[]{new SendEvenIntsProcess (chan),new ReadEvenIntsProcess (chan)}).run ();}
}

如代码所示，我首先实例化一个新的One2OneChannel对象，然后将其传递给SendEvenIntsProcess和ReadEventIntsProcess进程的构造函数。 这样做是One2OneChannel因为One2OneChannel实现了两个接口ChannelInput和ChannelOutput 。

通道内部

由于渠道是JC​​SP中的重要概念，因此请确保在继续之前先了解它们的工作方式。 如前所述，默认情况下通道是非缓冲的，但是也可以使它们成为缓冲的。 这是由于通道本身不处理缓冲特征并将该职责委托给另一个类而使之成为可能，该类必须实现一个称为ChannelDataStore的接口。 JCSP为此接口提供了多种内置实现，包括：

• ZeroBuffer ，它对应于默认的非缓冲特性。
• Buffer ，它为与之关联的通道提供阻塞的FIFO缓冲语义。
• InfiniteBuffer ，它还提供了FIFO的语义，不同之处在于，如果缓冲区为空，则只能阻止读取器。 由于缓冲区容量可以无限扩展，或者至少直到达到底层存储系统施加的限制，才不会阻塞写入器。

工作频道

考虑一个工作渠道的例子。 创建清单4中所示的One2OneChannel实例时，我将其内部ChannelDatasource设置为ZeroBuffer的新实例。 ZeroBuffer只能存储一个对象（或整数）。 它具有一个内部状态变量，该变量以值EMPTY开头，并在将对象放入后立即变为FULL 。

当SendEvenIntsProcess进程在其输出通道上执行write时会发生什么？ 好吧， One2OneChannel类的write()方法是synchronized()方法。 因此，在其上运行发送方进程的线程（稍后将看到在不同线程上运行的发送方进程和读取器进程）获取与此通道实例关联的监视器锁，并继续该方法。 在该方法中，首要任务是通过调用对象上的put方法将对象（或在这种情况下为整数） ZeroBuffer内部保存的ZeroBuffer实例。 这会将缓冲区的状态更改为FULL 。 此时，调用线程将调用wait ，导致线程进入监视器的等待集 ，然后释放监视器锁定，并阻塞线程。

在稍后的时间点，读取器线程在通道上调用read操作（这也是一种同步方法，因此读取器线程必须在继续之前获取监视器锁定）。 因为内部缓冲区的状态为FULL ，所以将返回可用数据并发出notify() 。 此notify()唤醒发送方线程，然后退出监视器等待集并收回监视器锁定。

在相反的情况下，如果读取器线程在其内部缓冲区处于EMPTY状态的通道上调用了read方法，则它必须wait ，在这种情况下，发送方线程在将数据对象写入到对象中后会通知它。内部缓冲区。

并行构造

您可能已经在清单4中注意到，驱动程序引入了一个名为Parallel的新类。 JCSP将Parallel类作为预定义的CSProcess提供，该CSProcess接受一组单独的CSProcess实例并“并行”运行它们（除最后一个进程外，所有进程都在不同的线程中运行；最后一个进程由Parallel对象在其各自的线程中运行）自己的控制线程）。 Parallel进程的run仅在其所有组件进程终止时终止。 因此， Parallel进程是一种使用通道（在驱动程序示例中）作为“电线”将它们连接在一起的一种将多个单独进程组合在一起的机制。

另一种查看Parallel构造的方法是说，它使由较小和较简单的组件组成更高层次的过程成为可能。 实际上，通过在每个新迭代中将在每个先前迭代中创建的组件连接在一起， Parallel允许您在多个迭代中创建一个任意复杂度的过程的整个互连网络 。 可以将所得的过程网络公开，并用作另一个CSProcess对象。

一个并行的例子

JCSP库提供了即插即用组件的集合，这些组件仅用于教育目的，非常适合我的目的：深入研究其中的一些内部实现应该很好地演示如何构成网络并发流程在JCSP中。 我使用以下示例过程来说明JCSP中Parallel构造的内部工作方式：

• PlusInt在其两个输入流上接受整数，将它们加在一起，然后将结果输出到其输出流。
• Delta2Int并行地将到达其输入流的每个整数广播到其两个输出通道。
• PrefixInt在其整数输入流之前（用户配置）整数。 （也就是说，此过程的第一个输出（在其输入通道上没有任何整数可用之前，是配置的整数本身）。以下输出是从输入流中直接获取的整数。）
• IntegrateInt是使用Parallel构造由前三个组成的过程。 它的功能是输出从其输入通道下降的整数的运行总和。

清单5显示了IntegrateInt类的run方法：

清单5. IntegrateInt流程

import jcsp.lang.*;public class IntegrateInt implements CSProcess
{private final ChannelInputInt in;private final ChannelOutputInt out;public IntegrateInt (ChannelInputInt in, ChannelOutputInt out){this.in = in;this.out = out;}public void run(){One2OneChannelInt a = new One2OneChannelInt ();One2OneChannelInt b = new One2OneChannelInt ();One2OneChannelInt c = new One2OneChannelInt ();new Parallel (new CSProcess[]{new PlusInt (in, c, a),new Delta2Int (a, out, b),new PrefixInt (0, b, c)}).run ();}
}

请注意，与清单4相比，此示例中使用了不同种类的通道。 IntegrateInt类使用ChannelInputInt和ChannelOutputInt通道， ChannelOutputInt ，它们可用于传输int类型的int 。 相比之下，清单4中的驱动程序使用ChannelInput和ChannelOutput ，它们是对象通道，可以用来从发送者到接收者的通道向下发送任意对象。 因此，在清单4中传输它们之前，我必须将int值包装为Integer对象。

您还可以看到清单5中的代码吗？ 本质上， PrefixInt进程的第一个输出为0，由PlusInt进程将其添加到第一个整数以到达输入通道。 将该结果写入通道a ，通道a形成Delta2Int进程的输入通道。 Delta2Int进程将整数结果写出 （该进程的整体输出通道），并将其发送到PrefixInt进程。 然后， PrefixInt进程将原样的整数发送给PlusInt进程，以将其添加到流中的第二个整数，依此类推。

图1显示了IntegrateInt流程的组成的示意图：

图1. IntegrateInt流程

网络中的网络

由三个较小的过程组成的IntegrateInt过程本身可以用作组成过程。 JCSP库提供了一个称为SquaresInt的过程， SquaresInt ，该过程生成整数流，该整数流是自然数的平方（1、2、3、4等）。 清单6显示了此过程的代码：

清单6. SquaresInt流程

public class SquaresInt implements CSProcess
{private final ChannelOutputInt out;public SquaresInt (ChannelOutputInt out){this.out = out;}public void run(){One2OneChannelInt a = new One2OneChannelInt ();One2OneChannelInt b = new One2OneChannelInt ();new Parallel (new CSProcess[]{new NumbersInt (a),new IntegrateInt (a, b),new PairsInt (b, out)}).run ();}
}

我确定您在清单6中注意到了两个新过程NumbersInt是一个内置过程，仅在其输出通道上输出从0开始的自然数。 PairsInt是一个过程，它将连续的输入值对相加并输出结果。 这两个新流程与IntegrateInt构成了SquaresInt流程，如图2中的图所示：

图2. SquaresInt流程

SquaresInt如何运作

在SquaresInt之前，让我们考虑一下SquaresInt流程的内部工作原理。 在下面，您可以看到SquaresInt各个通道上的流量如何SquaresInt ：

Channel "a": [0, 1, 2, 3, 4, 5, 6, 7, 8, ...ad infinitum]
Channel "b":  [0, 1, 3, 6, 10, 15, 21, 28, 36, ...ad infinitum]
Channel "out":    [1, 4, 9, 16, 25, 36, 49, 64, 81 ...ad infinitum]

您是否看到了如何将整数写入通道a导致将整数写入通道b进而写入通道out的模式 ？ 在第一个“滴答”中， NumbersInt进程将整数0写入通道a 。 IntegrateInt进程还将整数0（毕竟，这是运行总计的当前值）写入通道b 。 PairsInt进程在PairsInt不会产生任何结果，因为它需要两个输入才能使用。 在第二个滴答之后， NumbersInt进程将整数1写入其输出通道。 这使IntegrateInt进程将运行总数更改为0+1=1 ，从而将整数1写入通道b 。

在这一点上， PairsInt有两个整数输入要使用-前一个刻度的整数0和当前整数的整数1。 它将它们加在一起并将输出0+1=1写入channel out 。 请注意，1是1的平方，因此我们可能在这里。 以示例前进到下一个（第三个）滴答声， NumbersInt进程将整数2写入通道a 。 这使IntegrateInt进程将运行总计更新为1 （以前的总计） + 2 （新值） = 3并将此整数写到通道b上 。

PairsInt进程看到的最后两个整数是什么？ 它们是1 （在上一个刻度线期间）和3 （在当前的刻度线期间）。 因此，该过程将这两个整数相加并将1+3=4写入channel out 。 您会注意到4是2的平方，表示SquaresInt正在正常工作。 实际上，此后您可以继续运行该程序任何滴答声，并且可以验证写入通道输出的整数始终是下一行整数的平方。 我将在下一部分中精确地做到这一点。

数学题题

万一您想知道，我可以解释平方如何生成的数学基础。 假设您在NumbersInt进程已经输出不超过某个n-1的整数时偷看了框内。 IntegrateInt过程最后产生的运行总和（并通过共享通道b馈入PairsInt过程）将是[1+2+3+...+(n-1)] = (n-1)(n-2)/2 。

在下一个滴答中， NumbersInt将输出n ，导致IntegrateInt进程中的运行总和增加到(1+2+3+...+n) = n(n-1)/2 。 然后，该总和将通过共享通道b馈入PairsInt过程。 PairsInt会将这两个数字加在一起以生成[(n-1)(n-2)/2 + n(n-1)/2] = [(n-2) + n](n-1)/2 = (2n-2)(n-1)/2 = (n-1)exp2 。

接下来， NumbersInt进程将产生（n + 1） 。 与此对应， IntegrateInt进程会将n(n+1)/2输入PairsInt进程。 PairsInt然后将生成[n(n-1)/2 + n(n+1)/2] = nexp2 。 根据需要对所有n进行通用化将得出所有平方。

JCSP中的确定性

上面的示例演示了CSP的组成语义-即如何使用Parallel构造由细粒度的无状态组件组成分层网络。 通信并行过程的所有此类分层网络的卖点在于它们是完全确定的。 在这种情况下， 确定性是什么意思？ 这意味着这种分层网络的输出仅取决于提供给它的输入，而与网络所运行的运行时环境（JVM）的特性无关。也就是说，过程网络独立于调度JVM的策略以及将其分发到多个处理器的策略。 （我在这里假设一个节点；但是，没有任何内在的方法可以阻止将同一论点传递到一个过程网络中，该过程网络实际上分布在多个节点之间，并且进程通过有线方式进行通信。）

确定性是工具包中包含的功能强大的工具，因为它使您可以清楚地推断程序的行为，而不必担心运行时环境可能会对程序产生什么影响。 同时，确定性并不是并发编程的唯一可能或必要的方法。 正如下一个（也是最后一个）工作示例所示，在JCSP库中，非确定性在工作中同样具有强大的概念。

JCSP中的非确定性

在许多实际应用中，不确定性是一个因素，其中可见结果是事件发生顺序的函数。 换句话说，非确定性在并发应用程序中起作用，在并行应用程序中，结果取决于设计而不是偶然进行调度。 就像您将看到的那样，JCSP明确处理了此类问题。

例如，假设一个过程在下一步做什么方面有很多选择。 每个备选方案都具有与之关联的防护装置 ，该防护装置必须“准备就绪”，才能考虑将备选方案考虑为可能的选择。 该过程从可用（即就绪）选项中选择一种替代方法。 选择本身可以基于不同的策略，即任意选择，最高优先级选择或公平选择。

活动选择策略

在JCSP的特定上下文中，提供了一个称为Guard的抽象类，该抽象类必须由争用进程选择的事件对象所继承。 流程本身使用另一个预先提供的类（ Alternative ，这些保护对象必须作为对象数组传递到其构造函数中。 Alternative类提供了三种类型的事件选择策略的方法。

Alternative类的select()方法对应于任意选择策略。 select()方法调用将阻塞，直到一个或多个警卫就绪为止（请记住，所有竞争的警卫对于Alternative类都是已知的）。 可以随时选择一个就绪的警卫，并返回其索引（在传入的警卫数组中）。

priSelect()方法对应于最高优先级策略。 也就是说，如果准备好多个警卫，则返回索引最低的警卫； 假设已将传递给Alternative构造函数的数组中的防护按优先级从高到低的顺序排序。

最后，方法fairSelect在一个以上的预备警卫之间进行选择是公平的 ：在此方法的连续调用中，不会选择两次单独的预备警卫，而未选择另一个预备警卫和可用警卫。 因此，如果防护的总数为n ，则在最坏的情况下，对于超过n个连续的选择操作，没有就绪的防护将不会被取消选择。

任意选择策略最适合无关紧要的过程，如果准备好了多个防护措施，如何进行选择； 对于不希望出现饥饿或最坏情况下的服务时间的流程（例如在实时系统中），它的效果不佳。 在前一种情况下，建议使用fairSelect方法，而在后一种情况下，最好使用priSelect()方法。

警卫类型

大致而言，JCSP提供了三种类型的防护：

• 通道防护始终与进程正在等待从中读取数据的通道相对应。 即，仅当通道另一端的过程已输出到通道保护并且此过程尚未输入此数据时，通道保护才就绪。
• 计时器保护始终对应于设置（绝对）超时。 也就是说，如果计时器保护超时，则它已准备就绪。
• 防盗器随时准备就绪。

JCSP中的通道保护可以是以下类型： AltingChannelInput / AltingChannelInputInt ，只要对象/整数数据（分别）在相应通道中暂挂，它们就准备就绪； 或AltingChannelAccept ，如果通道上有未接受的“ CALL”挂起，则准备就绪（有关更多信息，请AltingChannelAccept ）。 这些是抽象类，并具有One2One和Any2One类型通道的形式的具体实现。 JCSP中的计时器保护器为CSTimer类型，而跳过保护器作为“ Skip类提供。

守卫在工作

我以一个简单的示例结束了对JCSP的介绍，该示例演示了如何使用JCSP防护措施来促进并发应用程序中的不确定性。 假设您必须开发一个乘法（或缩放 ）设备，该设备读取在其输入通道上以固定速率到达的整数，将它们乘以某个因子，然后将其写出到其输出通道。 设备可以以初始因子开始，但是该因子每五秒自动增加一倍。

故事中的故事如下：系统中存在第二个控制器进程，该进程可以通过专用通道向设备发送suspend operation信号。 这使设备挂起自身，并通过第二个通道将乘数的当前值发送到控制器。

挂起时，设备应仅允许所有传入的整数在其输出通道上不变地通过。 控制器进程-也许在使用设备发送给它的乘数作为输入执行某些计算之后-通过专用通道将新因子向下发送给设备。 （请注意，设备处于挂起状态时，必须接受该因素。）

将更新的因子注入设备也可以用作其唤醒信号。 该设备现在恢复其放大操作，将输入整数乘以新更新的因子。 此时计时器也会重置，因此将新的乘数设置为在五秒钟后加倍，依此类推。

图3中的图描述了缩放设备：

图3.缩放设备

ScaleInt流程

清单7中显示了缩放设备的源代码。在本示例中，不确定性是由于以下事实： 输出值基于in和inject流中的值（以及这些值的顺序）到达）。

清单7. ScaleInt流程

import jcsp.lang.*;
import jcsp.plugNplay.ints.*;public class ScaleInt implements CSProcess
{private int s;private final ChannelOutputInt out, factor;private final AltingChannelInputInt in, suspend, inject;public ScaleInt (int s, AltingChannelInputInt suspend, AltingChannelInputInt in, ChannelOutputInt factor, AltingChannelInputInt inject, ChannelOutputInt out){this.s = s;this.in = in;this.out = out;this.suspend = suspend;this.factor = factor;this.inject = inject;}public void run(){final long second = 1000;               // Java timings are in millisecsfinal long doubleInterval = 5*second;final CSTimer timer = new CSTimer ();final Alternative normalAlt = new Alternative (new Guard[] {suspend, timer, in});final int NORMAL_SUSPEND=0, NORMAL_TIMER=1, NORMAL_IN = 2;final Alternative suspendedAlt = new Alternative (new Guard[] {inject, in});final int SUSPENDED_INJECT=0, SUSPENDED_IN = 1;long timeout = timer.read () + doubleInterval;timer.setAlarm (timeout);while (true){switch (normalAlt.priSelect ()){case NORMAL_SUSPEND:suspend.read ();              // don't care what's sentfactor.write (s);             // reply with the crucial informationboolean suspended = true;while (suspended){switch (suspendedAlt.priSelect ()){case SUSPENDED_INJECT:    // this is the resume signal as wells = inject.read ();     // get the new scaling factorsuspended = false;      // and resume normal operationstimeout = timer.read () + doubleInterval;timer.setAlarm (timeout);break;case SUSPENDED_IN:out.write (in.read ());break;}}break;case NORMAL_TIMER:timeout = timer.read () + doubleInterval;timer.setAlarm (timeout);s = s*2;break;case NORMAL_IN:out.write (s * in.read ());break;}}}
}import jcsp.lang.*;
import jcsp.plugNplay.ints.*;public class Controller implements CSProcess
{private long interval;private final ChannelOutputInt suspend, inject;private final ChannelInputInt factor;public Controller (long interval, ChannelOutputInt suspend, ChannelOutputInt inject, ChannelInputInt factor){ this.interval = interval;this.suspend = suspend;this.inject = inject;this.factor = factor;}public void run (){int currFactor = 0;final CSTimer tim = new CSTimer ();long timeout = tim.read ();while (true){timeout += interval;tim.after (timeout);        // blocks until timeout reachedsuspend.write (0);          // suspend signal (value irrelevant)currFactor = factor.read ();          currFactor ++;              // compute new factorinject.write (currFactor);  // inject new factor}}
}import jcsp.lang.*;
import jcsp.plugNplay.ints.*;public class DriverProgram
{public static void main(String args[]){try{final One2OneChannelInt temp = new One2OneChannelInt ();final One2OneChannelInt in = new One2OneChannelInt ();final One2OneChannelInt suspend = new One2OneChannelInt ();final One2OneChannelInt factor = new One2OneChannelInt ();final One2OneChannelInt inject = new One2OneChannelInt ();final One2OneChannelInt out = new One2OneChannelInt ();new Parallel(new CSProcess[]{new NumbersInt (temp),new FixedDelayInt (1000, temp, in),new ScaleInt (2, suspend, in, factor, inject, out),new Controller (6000, suspend, inject, factor),new PrinterInt (out, "--> ", "\n")}).run ();}catch (Exception e){e.printStackTrace();}}
}

上面的ScaleInt类对应于缩放设备。 如前所述，此类必须实现CSProcess接口。 由于上面的代码演示了许多概念，因此我将一一讨论它的各个方面。

两种选择

ScaleInt类中第一个感兴趣的方法是run() 。 我在run()方法中所做的第一件事是创建Alternative类的两个实例，每个实例具有一组不同的Guards对象。

用变量normalAlt表示的第一个Alternative实例旨在在设备正常运行时使用。 与之关联的防护措施列表如下：

• 暂停是One2OneChannelInt的实例。 如前所述，一个One2OneChannelInt实现了一个零缓冲且完全同步的读取器/写入器整数通道。 这是控制器进程将挂起信号发送到设备的通道。
• timer是CSTimer一个实例，该实例设置为每五秒钟触发一次，届时设备会将乘数的当前值加倍。
• in是设备接收输入整数的One2OneChannelInt的实例。

第二个Alternative实例（由suspendedAlt表示）旨在在设备先前已由Controller暂停的情况下使用。 与之关联的防护措施列表如下：

• 注入是One2OneChannelInt的实例，控制器进程将使用它来向设备发送新的乘数（也用作唤醒信号）。
• in是您之前看到的One2OneChannelInt的相同实例； 设备在此通道上接收输入整数。

这两个Alternative实例在不同的情况下用于等待警卫人员准备就绪，列出的顺序为隐式优先级顺序。 例如，如果normalAlt的suspend和timer防护normalAlt恰好同时准备就绪，则将首先处理与该suspend防护措施相对应的事件。

警惕！

下一个关注点是当每个警卫准备就绪时会发生什么。 我首先处理normalSelect ，假设设备运行正常（即尚未挂起）：

• 如果控制器已向设备发送了暂停信号，则将以最高优先级处理此事件。 作为响应，设备通过称为factor的通道将乘法因子的当前值发送到控制器。 然后，它将一个称为suspended的内部标志设置为true并进入循环，等待发送信号以恢复。 在此循环内，设备在第二个Alternative实例（ suspendedAlt ）上调用priSelect()方法。

  此Alternative实例包括两个保护措施：第一个保护措施表示事件，其中控制器将新的乘法因子发送到设备，第二个保护措施表示整数到达设备的输入通道。 在前一种情况下，设备使用从inject通道读取的新值更新因子（保存在变量s ），将suspended标志设置回false （从而确保在下一次迭代中退出内部循环），复位使用当前计时器值作为基础的警报，并发出警报。 在后一种情况下，设备仅从其输入通道读取整数并将其写入输出通道（这是要求在设备挂起时不使用乘法因子）。

• 优先级次高的事件是警报响起的事件。 这将导致设备将当前倍增系数加倍，使用当前定时器值作为基础来重置警报，然后返回等待下一个事件。
• 第三种可能的事件是在设备的输入通道上接收到整数。 作为响应，设备读取整数，将其乘以当前因子s ，并将结果写入其输出通道。

The Controller class

The next class to consider is the Controller class. Remember, the job of the controller class is to inject new multiplying factor values into the device process on a periodic basis (based on a complex computation, presumably). In this example, the periodic basis is just a timer that goes off at a regular, configurable, interval. Every time that happens, the controller writes an 0 on the suspend channel (that is, it suspends the device) and reads the current multiplying factor on its input channel called factor .

At this point, the controller merely increments this value by one and injects it back into the device over the one-to-one channel (called inject ) dedicated to this purpose. This signals the device to resume, at which point the timer is reset to go off after the appropriate interval.

The DriverProgram class

The only class that remains is the driver class DriverProgram . This class creates the appropriate channels and an array of CSProcess instances. It uses the JCSP-provided class NumbersInt to generate a sequence of natural numbers that are fed through the temp channel to another built-in class, called FixedDelayInt . As its name indicates, FixedDelayInt passes data coming in through its input channel onto its output channel after a fixed delay -- one second in the example code.

This stream of natural numbers one second apart is then fed into the in channel of the ScaleInt process. The output from the out channel of the ScaleInt process is fed to the JCSP-provided PrinterInt process, which in turn prints the integer values onto System.out .

第2部分的结论

In this second part of my three-part introduction to CSP for Java programmers, I've both explained and demonstrated the theory of CSP in concurrent programming. Following an overview of CSP constructs, I introduced you to the most popular of the Java-based CSP libraries, JCSP. Since the Java language doesn't offer native support for CSP constructs, the JCSP library internally uses the native concurrency constructs that Java does support, such as synchronized() , wait() , and notify() . To help you understand exactly how JCSP works, I explained the internal implementation of some of the JCSP library classes in terms of these Java constructs, and then demonstrated their use in several working examples.

The discussion here serves as an excellent foundation for the final article in this series, where I explain the parallels between CSP and AOP, briefly compare the CSP approach to concurrency to that of the new java.util.concurrent package, and introduce you to a number of techniques for advanced synchronization with JCSP .

致谢

我要感谢在撰写本系列文章期间我从Peter Welch教授那里得到的鼓励。 尽管他的日程安排很忙，但他还是花了一些时间对草案版本进行了非常彻底的审查，并为提高系列的质量和准确性提供了许多宝贵的意见。 所有其余错误都是我的！ 我在文章中使用的示例基于和/或派生自Javadocs中针对JCSP库和/或JCSP网站上提供的Powerpoint演示幻灯片的文档。 这两种资源均提供了大量可供探索的信息。

翻译自: https://www.ibm.com/developerworks/java/library/j-csp2/index.html