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

Question

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

例如:

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

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

此外，我可以从 java.util.concurrent 使用的唯一类是 Semaphore类。

Answer 1

在线程之间来回传递数据的一种简单方法是使用接口(interface) BlockingQueue<E> 的实现。 , 位于包裹 java.util.concurrent .

此接口(interface)具有将元素添加到具有不同行为的集合的方法:

• add(E) : 如果可能添加，否则抛出异常
• boolean offer(E) : 如果元素已添加则返回 true，否则返回 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)可以被任何线程再次调用，并返回保存在局部变量中的内容。

作为一个重要的评论，请注意对被传递对象的引用保存在一个私有(private)字段中，并且方法take()和 put(E)都是最终。这是最重要的，但经常被忽视。如果这些方法不是最终的(或者更糟，字段不是私有(private)的)，继承类将能够改变 take() 的行为。和 put(E)违约。

最后，您可以避免在 take() 中声明局部变量的需要使用 try {} finally {} 的方法如下:

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

这里，这个例子的重点只是为了展示 try/finally 的用法。没有经验的开发人员不会注意到这一点。显然，在这种情况下，没有真正的 yield 。

哦该死，我几乎已经完成了你的家庭作业。在报应中——为了测试您对信号量的了解——，您为什么不实现 BlockingQueue 契约定义的其他一些方法呢？例如，您可以实现 offer(E)方法和 take(E, long, TimeUnit) !

祝你好运。