跳至主要内容

基于队列的负载均衡模式在 Java 中：平衡工作负载以实现可扩展的性能

弹性异步缓冲解耦容错消息传递可扩展性同步线程管理大约 5 分钟

也称为

  • 负载均衡
  • 消息队列

基于队列的负载均衡设计模式的意图

基于队列的负载均衡模式巧妙地利用队列来管理 Java 中的系统负载均衡，通过有效地将工作负载分配到生产者和消费者之间，从而增强系统性能和可扩展性。

基于队列的负载均衡模式的详细解释，包括现实世界中的例子

现实世界中的例子

在实际场景中，类似于繁忙的餐厅，基于队列的负载均衡是一种系统优化策略，它使用队列系统化地管理订单，从而保持高质量和高效的服务。在高峰时间，如果所有顾客都立即得到服务，厨房将会不堪重负，导致长时间的等待时间和潜在的订单错误。为了解决这个问题，餐厅通过使用售票机来实施基于队列的负载均衡系统。

当顾客下单时，他们会收到一个票号，他们的订单会被放入队列中。然后，厨房员工按订单接收顺序逐个处理订单。这确保厨房能够以可控的速度处理工作负载，防止超负荷并保持服务质量。顾客可以安心等待，因为他们知道他们的订单已排队，即使在最繁忙的时候也会得到高效处理。

通俗地说

基于队列的负载均衡是一种设计模式，它使用队列来管理和平衡生产者和消费者之间的工作负载，防止系统过载并确保平稳处理。

维基百科说

消息队列是进程间通信（IPC）和线程间通信的重要组成部分，它使用队列来管理消息传递。它们有助于解耦生产者和消费者，允许异步处理，这是基于队列的负载均衡模式的关键方面。

基于队列的负载均衡模式在 Java 中的编程示例

基于队列的负载均衡模式有助于管理高流量、间歇性的任务突发，这些突发可能会使系统不堪重负。它使用队列作为缓冲区来保存任务，从而将任务生成与任务处理解耦。任务以受控速度处理，确保最佳负载管理和容错能力，这对维护健壮的系统架构至关重要。

首先，让我们看一下MessageQueue和Message类。MessageQueue充当缓冲区，存储消息，直到它们被ServiceExecutor检索。Message表示要处理的任务。

public class Message {
  // Message details
}
public class MessageQueue {
  private Queue<Message> queue;

  public MessageQueue() {
    queue = new LinkedList<>();
  }

  // Method to add a message to the queue
  public void addMessage(Message message) {
    queue.add(message);
  }

  // Method to retrieve a message from the queue
  public Message getMessage() {
    return queue.poll();
  }
}

接下来，我们有TaskGenerator类。此类表示任务生产者。它生成任务并将其提交给MessageQueue。

public class TaskGenerator implements Runnable {
    
  private MessageQueue msgQueue;
  private int taskCount;

  public TaskGenerator(MessageQueue msgQueue, int taskCount) {
    this.msgQueue = msgQueue;
    this.taskCount = taskCount;
  }

  @Override
  public void run() {
    for (int i = 0; i < taskCount; i++) {
      Message message = new Message(); // Create a new message
      msgQueue.addMessage(message); // Add the message to the queue
    }
  }
}

ServiceExecutor类表示任务消费者。它从MessageQueue中检索任务并处理它们。

public class ServiceExecutor implements Runnable {
    
  private MessageQueue msgQueue;

  public ServiceExecutor(MessageQueue msgQueue) {
    this.msgQueue = msgQueue;
  }

  @Override
  public void run() {
    while (true) {
      Message message = msgQueue.getMessage(); // Retrieve a message from the queue
      if (message != null) {
        // Process the message
      } else {
        // No more messages to process
        break;
      }
    }
  }
}

最后，我们有App类，它设置TaskGenerator和ServiceExecutor线程，并将它们提交给ExecutorService。

public class App {
  public static void main(String[] args) {
    var msgQueue = new MessageQueue();

    final var taskRunnable1 = new TaskGenerator(msgQueue, 5);
    final var taskRunnable2 = new TaskGenerator(msgQueue, 1);
    final var taskRunnable3 = new TaskGenerator(msgQueue, 2);

    final var srvRunnable = new ServiceExecutor(msgQueue);

    ExecutorService executor = Executors.newFixedThreadPool(2);
    executor.submit(taskRunnable1);
    executor.submit(taskRunnable2);
    executor.submit(taskRunnable3);
    executor.submit(srvRunnable);

    executor.shutdown();
  }
}

在这个例子中，TaskGenerator线程以可变速率生成任务并将它们提交给MessageQueue。ServiceExecutor从队列中检索任务并按自己的速度处理它们，防止系统因峰值负载而不堪重负。

运行应用程序将产生以下控制台输出

[main] INFO App - Submitting TaskGenerators and ServiceExecutor threads.
[main] INFO App - Initiating shutdown. Executor will shutdown only after all the Threads are completed.
[pool-1-thread-2] INFO TaskGenerator - Message-1 submitted by pool-1-thread-2
[pool-1-thread-1] INFO TaskGenerator - Message-5 submitted by pool-1-thread-1
[pool-1-thread-1] INFO TaskGenerator - Message-4 submitted by pool-1-thread-1
[pool-1-thread-2] INFO TaskGenerator - Message-2 submitted by pool-1-thread-2
[pool-1-thread-1] INFO TaskGenerator - Message-3 submitted by pool-1-thread-1
[pool-1-thread-2] INFO TaskGenerator - Message-1 submitted by pool-1-thread-2
[pool-1-thread-1] INFO TaskGenerator - Message-2 submitted by pool-1-thread-1
[pool-1-thread-2] INFO ServiceExecutor - Message-1 submitted by pool-1-thread-2 is served.
[pool-1-thread-1] INFO TaskGenerator - Message-1 submitted by pool-1-thread-1
[pool-1-thread-2] INFO ServiceExecutor - Message-5 submitted by pool-1-thread-1 is served.
[pool-1-thread-2] INFO ServiceExecutor - Message-4 submitted by pool-1-thread-1 is served.
[pool-1-thread-2] INFO ServiceExecutor - Message-2 submitted by pool-1-thread-2 is served.
[pool-1-thread-2] INFO ServiceExecutor - Message-3 submitted by pool-1-thread-1 is served.
[pool-1-thread-2] INFO ServiceExecutor - Message-1 submitted by pool-1-thread-2 is served.
[pool-1-thread-2] INFO ServiceExecutor - Message-2 submitted by pool-1-thread-1 is served.
[pool-1-thread-2] INFO ServiceExecutor - Message-1 submitted by pool-1-thread-1 is served.
[pool-1-thread-2] INFO ServiceExecutor - Service Executor: Waiting for Messages to serve .. 
[pool-1-thread-2] INFO ServiceExecutor - Service Executor: Waiting for Messages to serve .. 
[pool-1-thread-2] INFO ServiceExecutor - Service Executor: Waiting for Messages to serve .. 
[pool-1-thread-2] INFO ServiceExecutor - Service Executor: Waiting for Messages to serve .. 
[main] INFO App - Executor was shut down and Exiting.
[pool-1-thread-2] ERROR ServiceExecutor - sleep interrupted

何时在 Java 中使用基于队列的负载均衡模式

  • 当有可变的工作负载时，你需要确保峰值负载不会使系统不堪重负
  • 在分布式系统中，任务的产生速率与消费速率不同
  • 在异步消息传递系统中解耦生产者和消费者

基于队列的负载均衡模式在 Java 中的现实世界应用

  • Amazon Web Services (AWS) 简单队列服务 (SQS)
  • RabbitMQ
  • 企业 Java 应用程序中的 Java 消息服务 (JMS)

基于队列的负载均衡模式的优点和权衡

优点

  • 解耦生产者和消费者，使它们能够以自己的速度运行
  • 通过防止过载情况，提高系统弹性和容错能力
  • 通过允许添加更多消费者来处理增加的负载，增强可扩展性

权衡

  • 增加系统架构的复杂性
  • 可能会引入延迟，因为消息需要排队和出队
  • 需要管理和监控额外的组件（队列）
  • 异步消息传递：基于队列的负载均衡使用异步消息传递来解耦生产者和消费者
  • 断路器：通常与基于队列的负载均衡一起使用，通过暂时停止消息处理来防止系统过载
  • 生产者-消费者：基于队列的负载均衡是生产者-消费者模式的特定应用，其中队列充当中间体
  • 重试：与基于队列的负载均衡一起工作，通过重试失败的操作来处理瞬态故障

参考文献和鸣谢

MIT 许可
版权 © 2024 iluwatar