使用 Java 在线程之间传输数据

Question

我正在编写一个模仿电影院的多线程应用程序。每个参与的人都是自己的线程，并发必须完全由信号量来完成。我遇到的唯一问题是如何基本上链接线程以便它们可以通信（例如通过管道）。

例如：

Customer[1] 是一个线程，它获取一个信号量，让它走到票房。现在，客户[1] 必须告诉票房代理他们想看电影“X”。然后 BoxOfficeAgent[1] 也是一个线程，必须检查以确保电影未满，然后出售门票或告诉 Customer[1] 选择另一部电影。

如何来回传递数据，同时仍保持与信号量的并发性？

另外，我可以从 java.util.concurrent 使用的唯一类是 信号量 类。

Answer 1

在线程之间来回传递数据的一种简单方法是使用接口 BlockingQueue，位于包java.util.concurrent。

该接口具有以不同行为向集合添加元素的方法：

• add(E)：如果可能，则添加，否则抛出异常
• boolean Offer(E)：如果元素已添加，否则为 false
• boolean Offer(E, long, TimeUnit)：尝试添加元素，等待指定的时间
• put(E)：阻止调用线程直到添加元素

为止它还定义了具有类似行为的元素检索方法：

• take()：阻塞直到有可用元素
• poll(long, TimeUnit)：检索元素或返回 null

我最常用的实现是： ArrayBlockingQueue，LinkedBlockingQueue 和 SynchronousQueue。

第一个 ArrayBlockingQueue 具有固定大小，由传递给其构造函数的参数定义。

第二个 LinkedBlockingQueue 的大小不受限制。它总是接受任何元素，即 offer 会立即返回 true，add 永远不会抛出异常。

第三个，对我来说最有趣的一个，SynchronousQueue，就是一个管道。您可以将其视为大小为 0 的队列。它永远不会保留元素：只有当有其他线程尝试从中检索元素时，该队列才会接受元素。相反，如果有另一个线程尝试推送某个元素，则检索操作只会返回该元素。

为了满足仅使用信号量完成同步的家庭作业要求，您可以从我给您提供的有关 SynchronousQueue 的描述中获得启发，并编写一些非常相似的内容：

class Pipe<E> {
  private E e;

  private final Semaphore read = new Semaphore(0);
  private final Semaphore write = new Semaphore(1);

  public final void put(final E e) {
    write.acquire();
    this.e = e;
    read.release();
  }

  public final E take() {
    read.acquire();
    E e = this.e;
    write.release();
    return e;
  }
}

请注意，此类呈现与我描述的 SynchronousQueue 类似的行为。

一旦调用方法 put(E) ，它就会获取写入信号量，该信号量将保留为空，以便对同一方法的另一个调用将在其第一行阻塞。然后，此方法存储对正在传递的对象的引用，并释放读取的信号量。此版本将使任何调用 take() 方法的线程都可以继续执行。

take() 方法的第一步自然是获取读取信号量，以禁止任何其他线程同时检索该元素。检索元素并将其保存在局部变量中后（练习：如果删除该行 E e = this.e 会发生什么？），该方法释放写入信号量，以便put(E) 方法可以由任何线程再次调用，并返回局部变量中保存的内容。

作为一个重要的注释，请注意对正在传递的对象的引用保存在私有字段中，并且方法take()和put(E)< /code> 都是最终。这是极其重要的，但也常常被忽视。如果这些方法不是最终方法（或更糟糕的是，该字段不是私有的），继承类将能够改变 take() 和 put(E) 的行为合同。

最后，您可以通过使用 try {} finally {} 来避免在 take() 方法中声明局部变量，如下所示：

class Pipe<E> {
  // ...
  public final E take() {
    try {
      read.acquire();
      return e;
    } finally {
      write.release();
    }
  }
}

这里是本示例的要点如果只是为了展示没有经验的开发人员不会注意到的 try/finally 的使用。显然，在这种情况下，没有真正的收益。

哦该死，我已经帮你完成大部分作业了。为了报应——并且为了测试你对信号量的了解——，为什么不实现 BlockingQueue 合约定义的一些其他方法呢？例如，您可以实现一个 offer(E) 方法和一个 take(E, long, TimeUnit)！

祝你好运。

Answer 2

将其视为带有读/写锁的共享内存。

1. 创建一个缓冲区来放置消息。
2. 应使用锁/信号量来控制对缓冲区的访问。
3. 使用此缓冲区进行线程间通信。

问候

PKV