Java中延迟队列使用指南

喜欢ヅ旅行 2022-09-12 11:51 178阅读 0赞

# 1.介绍 #

本文中，将了解 java.util.concurrent 包中的 DelayQueue 构造。 这是一个阻塞队列，可以在生产者-消费者程序中使用。

它有一个非常有用的特性——当消费者想要从队列中取出一个元素时，**只能在该特定元素的延迟到期时才能取出**。

# 2.为DelayQueue中的元素实现Delayed #

想要放入 DelayQueue 的每个元素都需要实现 Delayed 接口。 假设要创建一个 DelayObject 类。 该类的实例将被放入 DelayQueue。

将 String 数据和 delayInMilliseconds 作为和参数传递给它的构造函数：

public class DelayObject implements Delayed { 
    
        private String data;
        private long startTime;
    
        public DelayObject(String data, long delayInMilliseconds) { 
            this.data = data;
            this.startTime = System.currentTimeMillis() + delayInMilliseconds;
        }
    
        @Override
        public long getDelay(TimeUnit unit) { 
            long diff = startTime - System.currentTimeMillis();
            return unit.convert(diff, TimeUnit.MILLISECONDS);
        }
        @Override
        public int compareTo(Delayed o) { 
            return Ints.saturatedCast(
                    this.startTime - ((DelayObject) o).startTime);
        }
            @Override
        public String toString() { 
            return "DelayObject{" +
                    "data='" + data + '\'' +
                    ", startTime=" + startTime +
                    '}';
        }
    }

上面代码

定义一个 startTime——从队列中消耗元素的时间。

实现 getDelay() 方法——在给定的时间单位内返回与此对象关联的剩余延迟。

使用TimeUnit.convert() 方法以适当的 TimeUnit 返回剩余延迟

**当消费者尝试从队列中取出一个元素时，DelayQueue 将执行 getDelay() 以确定是否允许从队列中返回该元素。 如果 getDelay() 方法将返回零或负数，则表示可以从队列中检索它。**

**还需要实现 compareTo() 方法，因为 DelayQueue 中的元素会根据过期时间进行排序。 最先过期的元素放在队列的头部，过期时间最长的元素放在队列的尾部**

# 3.DelayQueue 消费者和生产者 #

为了能够测试 DelayQueue，需要实现生产者和消费者逻辑。 生产者类将队列、要生产的元素数量以及每条消息的延迟（以毫秒为单位）作为参数。

然后当 run() 方法被调用时，它将元素放入队列中，并在每次放入后休眠 500 毫秒：

public class DelayQueueProducer implements Runnable { 
    
        private final BlockingQueue<DelayObject> queue;
        /** * 要生产的元素数量 */
        private final Integer numberOfElementsToProduce;
        /** * 延迟毫秒数 */
        private final Integer delayOfEachProducedMessageMilliseconds;
    
    
        public DelayQueueProducer(BlockingQueue<DelayObject> queue, Integer numberOfElementsToProduce, Integer delayOfEachProducedMessageMilliseconds) { 
            this.queue = queue;
            this.numberOfElementsToProduce = numberOfElementsToProduce;
            this.delayOfEachProducedMessageMilliseconds = delayOfEachProducedMessageMilliseconds;
        }
    
        @Override
        public void run() { 
            for (int i = 0; i < numberOfElementsToProduce; i++) { 
                DelayObject object
                        = new DelayObject(
                        UUID.randomUUID().toString(), delayOfEachProducedMessageMilliseconds);
                System.out.println("放入对象: " + object);
                try { 
                    queue.put(object);
                    Thread.sleep(500);
                } catch (InterruptedException ie) { 
                    ie.printStackTrace();
                }
            }
        }
    }

消费者实现非常相似

public class DelayQueueConsumer implements Runnable { 
        private BlockingQueue<DelayObject> queue;
        private Integer numberOfElementsToTake;
        public AtomicInteger numberOfConsumedElements = new AtomicInteger();
    
    
        public DelayQueueConsumer(BlockingQueue<DelayObject> queue, Integer numberOfElementsToTake) { 
            this.queue = queue;
            this.numberOfElementsToTake = numberOfElementsToTake;
        }
    
        @Override
        public void run() { 
            for (int i = 0; i < numberOfElementsToTake; i++) { 
                try { 
                    DelayObject object = queue.take();
                    numberOfConsumedElements.incrementAndGet();
                    System.out.println("Consumer take: " + object);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    }

# 4.测试 #

为了测试 DelayQueue 的行为，将创建一个生产者线程和一个消费者线程。

生产者将 put() 两个对象以 5000毫秒的延迟放入队列。 该测试断言消费者消费了两条消息：

@Test
        public void test1() throws InterruptedException { 
            ExecutorService executor = Executors.newFixedThreadPool(2);
    
            BlockingQueue<DelayObject> queue = new DelayQueue<>();
            int numberOfElementsToProduce = 2;
            int delayOfEachProducedMessageMilliseconds = 500;
            DelayQueueConsumer consumer = new DelayQueueConsumer(
                    queue, numberOfElementsToProduce);
            DelayQueueProducer producer = new DelayQueueProducer(
                    queue, numberOfElementsToProduce, delayOfEachProducedMessageMilliseconds);
    
          
            executor.submit(producer);
            executor.submit(consumer);
    
            // 5s 后终止线程池
            executor.awaitTermination(5, TimeUnit.SECONDS);
            executor.shutdown();
            assertEquals(consumer.numberOfConsumedElements.get(), numberOfElementsToProduce);
        }

输出

放入对象: DelayObject{data='277958f5-500d-433c-8b87-38f941d33992', startTime=1631666478597}
    Consumer take: DelayObject{data='277958f5-500d-433c-8b87-38f941d33992', startTime=1631666478597}
    放入对象: DelayObject{data='17dfe66c-4a73-405e-9ac7-e99e986b5734', startTime=1631666479107}
    Consumer take: DelayObject{data='17dfe66c-4a73-405e-9ac7-e99e986b5734', startTime=1631666479107}

生产者放置对象，一段时间后，第一个延迟到期的对象被消耗。

# 5.消费者无法在给定时间内消费 #

假设有一个生产者正在生产一个将在 10 秒后过期的对象：

int numberOfElementsToProduce = 1;
    int delayOfEachProducedMessageMilliseconds = 10_000;
    DelayQueueConsumer consumer = new DelayQueueConsumer(
      queue, numberOfElementsToProduce);
    DelayQueueProducer producer = new DelayQueueProducer(
      queue, numberOfElementsToProduce, delayOfEachProducedMessageMilliseconds);

开始测试，但它会在 5 秒后终止。 由于 DelayQueue 的特性，消费者将无法消费队列中的消息，因为元素尚未过期：

executor.submit(producer);
    executor.submit(consumer);
    
    executor.awaitTermination(5, TimeUnit.SECONDS);
    executor.shutdown();
    assertEquals(consumer.numberOfConsumedElements.get(), 0);

# 6.生产立即到期的对象 #

当 Delayed 消息 getDelay() 方法的实现返回一个负数时，这意味着给定的元素已经过期。 在这种情况下，生产者将立即消耗该元素。

测试产生负延迟对象的情况：

int numberOfElementsToProduce = 1;
    int delayOfEachProducedMessageMilliseconds = -10_000;
    DelayQueueConsumer consumer = new DelayQueueConsumer(queue, numberOfElementsToProduce);
    DelayQueueProducer producer = new DelayQueueProducer(
      queue, numberOfElementsToProduce, delayOfEachProducedMessageMilliseconds);

开始测试用例时，消费者会立即消费该元素，因为它已经过期：

executor.submit(producer);
    executor.submit(consumer);
    
    executor.awaitTermination(1, TimeUnit.SECONDS);
    executor.shutdown();
    assertEquals(consumer.numberOfConsumedElements.get(), 1);