详解Java中的延时队列 DelayQueue

Z时代
2024-01-10
分类：综合

当用户超时未支付时，给用户发提醒消息。另一种场景是，超时未付款，订单自动取消。通常，订单创建的时候可以向延迟队列种插入一条消息，到时间自动执行。其实，也可以用临时表，把这些未支付的订单放到一个临时表中，或者Redis，然后定时任务去扫描。这里我们用延时队列来做。RocketMQ有延时队列，RibbitMQ也可以实现，Java自带的也有延时队列，接下来就回顾一下各种队列。

Queue

队列是一种集合。除了基本的集合操作以外，队列还提供了额外的插入、提取和检查操作。队列的每个方法都以两种形式存在：一种是当操作失败时抛异常，另一种是返回一个特定的值（null或者false，取决于具体操作）。后一种形式的插入操作是专门设计用于有界队列实现的，在大多情况下，插入操作不会失败。

队列通常（但不一定）以FIFO（先进先出）的方式对元素进行排序。例外情况包括优先级队列（根据提供的比较器对元素进行排序或元素的自然排序）和LIFO队列（或堆栈），对LIFO进行排序（后进先出）。无论使用哪种顺序，队列的开头都是该元素，可以通过调用remove（）或poll（）将其删除。在FIFO队列中，所有新元素都插入队列的尾部。其他种类的队列可能使用不同的放置规则。每个Queue实现必须指定其排序属性。无论使用哪种顺序，都可以通过调用remove()或poll()来删除队列开头的元素。在FIFO队列中，所有新元素都插入到队列的尾部。其他类型的队列可能使用不同的放置规则。每个队列实现都必须指定其排序属性。

offer方法在可以的情况下会向队列种插入一个元素，否则返回false。这不同于Collection.add方法，后者只能通过抛异常来添加元素。offer方法设计用于在正常情况下（而不是在例外情况下）发生故障时，例如在固定容量（或者“有界”）队列种使用。

remove()和poll()方法删除并返回队头元素。当队列为空时，remove()抛出异常，而poll()返回null。

element()和peek()方法返回队头元素。

PriorityQueue

PriorityQueue是一个无界优先级队列是基于优先级堆的。优先级队列种的元素根据自然顺序进行排序，或者通过在队列构建时提供的Comparator进行排序，当然这取决于使用哪种构造函数。优先级队列不允许空（null）元素。一个依赖自然顺序的优先级队列也不允许插入不可比较的对象。

优先级队列的队头元素是最小的元素，如果有多个元素并列最小，那么队头是它们其中之一。

优先级队列是无界的，但是有一个内部容量来控制用于在队列上存储元素的数组的大小。它总是至少与队列大小一样大。将元素添加到优先级队列时，其容量会自动增长

BlockingQueue

这种队列还支持以下操作：在检索元素时等待队列变为非空，并在存储元素时等待队列中的空间变为可用。

BlockingQueue方法有四种形式，它们以不同的方式处理操作，这些操作无法立即满足，但将来可能会满足：一种抛出异常，第二种返回特殊值（null或false，取决于具体操作），第三种阻塞当前线程，直到操作成功为止；第四种阻塞当前线程，超时则放弃。下表总结了这些方法：

阻塞队列不接受空元素，如果你试图add , put 或者 offer 一个null，将会抛NullPointerException。

阻塞队列是线程安全的。所有排队方法都使用内部锁或者其他形式的并发控制来保证以原子方式实现它们的效果。

阻塞队列被设计主要用于生产者-消费者队列。

下面是一个典型的生产者-消费者方案：


package com.example;
import java.text.MessageFormat;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
/**
 * @author ChengJianSheng
 * @date 2020/12/15
 */
public class Setup {
  public static void main(String[] args) {
    BlockingQueue<Bread> queue = new ArrayBlockingQueue<>(5);
    Producer p1 = new Producer(queue);
    Producer p2 = new Producer(queue);
    Consumer c1 = new Consumer(queue);
    Consumer c2 = new Consumer(queue);
    new Thread(p1, "p1").start();
    new Thread(p2, "p2").start();
    new Thread(c1, "c1").start();
    new Thread(c2, "c2").start();
  }
}
class Bread {
}
/**
 * 生产者
 */
class Producer implements Runnable {
  private final BlockingQueue<Bread> queue;
  public Producer(BlockingQueue<Bread> queue) {
    this.queue = queue;
  }
  @Override
  public void run() {
    try {
      while (true) {
        queue.put(produce());
      }
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
  public Bread produce() {
    try {
      Thread.sleep(Math.round(2000));
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    return new Bread();
  }
}
/**
 * 消费者
 */
class Consumer implements Runnable {
  private final BlockingQueue<Bread> queue;
  public Consumer(BlockingQueue<Bread> queue) {
    this.queue = queue;
  }
  @Override
  public void run() {
    try {
      while (true) {
        consume(queue.take());
      }
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
  public void consume(Bread bread) {
    try {
      Thread.sleep(Math.round(2000));
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}

ArrayBlockingQueue

ArrayBlockingQueue是用数组实现的有界阻塞队列。这种队列中的元素按FIFO（先进先出）排序。队头是在队列中停留最长时间的元素。队尾是在队列中停留时间最短的元素。新元素插入到队列的尾部，并且队列检索操作在队列的头部获取元素。

这是一个经典的“有界缓冲区”，其中固定大小的数组包含由生产者插入并由消费者提取的元素。创建后，容量将无法更改。试图将一个元素放入一个已满的队列将导致操作阻塞；试图从空队列中取出一个元素也会阻塞。

这个类支持一个可选的公平性策略，用于对等待的生产者和消费者线程进行排序。默认情况下，不保证这个顺序。然而，将公平性设置为true的队列将按FIFO顺序授予线程访问权。公平性通常会降低吞吐量，但会降低可变性并避免饥饿。

LinkedBlockingQueue

LinkedBlockingQueue是一个基于链表实现的可选边界的阻塞队列。

PriorityBlockingQueue

PriorityBlockingQueue是一个无界阻塞队列，它使用与PriorityQueue相同的排序规则，并提供阻塞检索操作。

DelayQueue

DelayQueue是一种由延迟元素组成的无界阻塞队列，在该队列中，仅当元素的延迟到期时才可以使用该元素。队头是已经过期的延迟元素，它已过期时间最长。如果没有过期的延迟，则队列没有头部，此时调用poll将返回null。当调用元素的getDelay(TimeUnit.NANOSECONDS)方法返回值小于或等于0时，就会发生过期。即使元素没有过期，也不能用take或者poll将其删除。

AbstractQueuedSynchronizer

AbstractQueuedSynchronizer提供了一个框架来实现依赖于先进先出（FIFO）等待队列的阻塞锁和相关的同步器（信号灯，事件等）。该类旨在为大多数依赖单个原子int值表示状态的同步器提供有用的基础。子类必须定义更改此状态的受保护方法，并定义该状态对于获取或释放此对象而言意味着什么。鉴于这些，此类中的其他方法将执行所有排队和阻塞机制。子类可以维护其他状态字段，但是仅跟踪关于同步的使用方法getState（），setState（int）和compareAndSetState（int，int）操作的原子更新的int值。