使用Java的阻塞队列的生产者-消费者模式
生产者-消费者是一种并发设计模式，其中一个或多个生产者线程参与生产任务，而一个或多个消费者线程参与消费这些任务。所有这些任务都保存在生产者和消费者都可以访问的共享队列中。
生产者-消费者设计模式将生产者和消费者的工作联系起来，允许他们独立地扩展和发展。
在我们的应用程序中，有几种方法可以实现这种设计模式。然而，在本文中，我们将探索如何使用Java的BlockingQueue来实现生产者-消费者模式。
阻塞队列基础知识
Java的BlockingQueue是一个线程安全类，它使用内部锁定来确保所有排队方法本质上都是原子的。类名中的单词Blocking源于这样一个事实，即该类包含几个阻塞方法，如put，take，offer和poll。阻塞方法阻塞线程的执行进程，直到满足某些条件。
put方法阻塞，直到队列中有一些可用的空间来放置元素。类似地，take方法阻塞，直到队列中有空间放置新元素。方法offer和poll对于在特定时间之后解除阻塞的put和take是定时等效的。
当我们对BlockingQueue可以存储的元素的最大数量设置限制时，它被称为有界BlockingQueue。否则，它被称为无限BlockingQueue。
Java提供了几个BlockingQueue实现，如ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue、SynchronousQueue等。
生产者-消费者示例
以一家餐馆为例。顾客到达餐厅，下了一些食物订单。每一份订单都由厨师负责准备食物并将其提供给客户。
我们将使用Order类来表示客户订单。
包生产者消费者;
public class Order {
private final String name;
private final Long tableNumber;
public Order(final String name, final Long tableNumber) {
this.name = name;
this.tableNumber = tableNumber;
}
public String getName() {
return name;
}
public Long getTableNumber() {
return tableNumber;
}
}
现在让我们定义我们的厨师的行为。考虑下面的类。
package ProducerConsumer;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
public class Chef implements Runnable {
private final BlockingQueue<Order> orderBlockingQueue;
private Long ordersServed = 0L;
public Chef(final BlockingQueue<Order> orderBlockingQueue) {
this.orderBlockingQueue = orderBlockingQueue;
}
@Override
public void run() {
while (true) {
// Wait for an order
final Order order;
try {
order = orderBlockingQueue.take();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
// Sleep some random time to simulate the time to prepare the food
int sleepTime = (int) (Math.random() * 20000 + 10000);
try {
Thread.sleep(sleepTime);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
System.out.println("Order for " + order.getName() + " at table " + order.getTableNumber() + " is ready!");
ordersServed++;
}
}
public Long getOrdersServed() {
return ordersServed;
}
}
代码本身非常简单。Chef不断地从orderBlockingQueue中挑选订单。然后，它花一些时间准备订单，并最终服务。
在生产者-消费者世界中，厨师是客户生产的订单的消费者。
现在让我们看看生产者方面的事情。
package ProducerConsumer;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
public class Customer implements Runnable {
private final List<String> menuItems;
private final Long tableNumber;
private final BlockingQueue<Order> orderBlockingQueue;
private final CountDownLatch countDownLatch;
public Customer(final List<String> menuItems,
final Long tableNumber,
final BlockingQueue<Order> orderBlockingQueue,
final CountDownLatch countDownLatch) {
this.menuItems = menuItems;
this.tableNumber = tableNumber;
this.orderBlockingQueue = orderBlockingQueue;
this.countDownLatch = countDownLatch;
}
@Override
public void run() {
try {
countDownLatch.await();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
// Spend some time to simulate the customer's behavior
int sleepTime = (int) (Math.random() * 60000 + 1000);
try {
Thread.sleep(sleepTime);
} catch (InterruptedException e) {
e.printStackTrace();
}
final String foodName = menuItems.get((int) (Math.random() * menuItems.size()));
final Order order = new Order(foodName, tableNumber);
// Place order
try {
orderBlockingQueue.put(order);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
我们使用CountDownLatch来确保在餐厅营业之前没有顾客下订单。一旦餐厅开门，每个顾客都会花一些随机的时间来决定订单。完成订单后，他们将其放置在orderBlockingQueue上。
创建了生产者和消费者之后，我们现在可以处理Restaurant类了。
package ProducerConsumer;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
public class Restaurant {
final static int NUMBER_OF_CUSTOMERS = 100;
final static int NUMBER_OF_CHEFS = 7;
public static void main(String[] args) {
final BlockingQueue<Order> orderBlockingQueue = new LinkedBlockingQueue<>();
final CountDownLatch restaurantLatch = new CountDownLatch(1);
final Thread[] customers = new Thread[NUMBER_OF_CUSTOMERS];
for (int i = 0; i < NUMBER_OF_CUSTOMERS; i++) {
customers[i] = new Thread(
new Customer(List.of("Pizza", "Pasta", "Salad"),
(long) i,
orderBlockingQueue,
restaurantLatch));
customers[i].start();
}
final Thread[] chefs = new Thread[NUMBER_OF_CHEFS];
for (int i = 0; i < NUMBER_OF_CHEFS; i++) {
chefs[i] = new Thread(new Chef(orderBlockingQueue));
chefs[i].start();
}
restaurantLatch.countDown();
for (int i = 0; i < NUMBER_OF_CUSTOMERS; i++) {
try {
customers[i].join();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
for (int i = 0; i < NUMBER_OF_CHEFS; i++) {
chefs[i].interrupt();
}
}
}
餐厅类是所有东西结合在一起的地方。我们初始化一定数量的客户和厨师。一旦初始化，我们打开门闩打开餐厅。在几秒钟内，我们看到客户下订单，厨师准备这些订单。
生产者-消费者模式是软件开发的基石，提供了大量的实现选项。值得注意的是，Java的阻塞队列提供了一种实现这种模式的简单有效的方法。此外，该模式的相关性扩展到分布式系统，在那里它促进了数据生产和消费的解耦。这种解耦反过来又使系统具有可伸缩性、容错性和参与异步通信的能力，使其成为现代软件体系结构中的宝贵资产。
这就把我们带到了这篇文章的结尾。希望你今天学到了新东西！