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

Question

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

例如：

Customer1是一个线程，它获得一个信号量，让它走到票房。现在Customer1必须告诉票房代理他们想看电影"X“。然后BoxOfficeAgent1也是一个线程，必须检查以确保电影没有满，然后要么卖票，要么告诉Customer1选择另一部电影。

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

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

Answer 1

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

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

如果可能，则返回

• add(E)：；如果元素已添加，则返回exception
• boolean offer(E)：；如果元素已添加，则返回false；如果元素已添加，则返回false；otherwise
• boolean offer(E, long, TimeUnit)：尝试添加元素，等待指定数量的time
• put(E)：将阻塞调用线程，直到元素已添加

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

• take()：将一直阻塞，直到存在元素available
• 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)都是最终的。这一点非常重要，而且经常被忽略。如果这些方法不是put(E) (或者更糟糕的是，字段不是私有的)，继承类将能够改变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

从共享内存和读/写锁的角度来考虑它。

为了线程间通信的目的，应使用lock/semaphore.

• Use

1. 来控制对此缓冲区的访问。
2. 创建一个缓冲区来放置消息。

问候

PKV