「blockingqueue」BlockingQueue

blockingqueue

在项目中有一个地址建议的需求，由于给出地址建议需要获取的数据较多，且计算比较复杂，接口的耗时时间比较长，平均超过了1.3s，后期优化将业务接口只从缓存中获取数据，如果缓存中没有，则将请求数据放入到redis队列中，由定时任务10s一次轮询队列，由定时任务线程去计算，将结果放入缓存中。将接口的耗时缩短到30-40ms之间。虽然解决了耗时的问题，但是在用户在第一次获取建议时总是拿不到数据。在10s后再次请求才会有数据！为了解决这个问题，研究了一下blocking queue。

BlockingQueue分为ArrayBlockingQueue和LinkedBlockingQueue，ArrayBlockingQueue是基于数组实现的，而linkedBlockingQueue是基于链表实现的，ArrayBlockingQueue适合做有界队列，队列可容纳的最大元素需要在队列创建时指定，而LinkedBlockingQueue适合做无界队列，或者边界值非常大的队列，不会因为初值设置容量很大，吃掉很大的内存。

BlockingQueue非常适合做线程之间的数据共享通道，它会让服务线程在队列为空时，当有新的消息进入队列后，自动唤醒线程。那么它是如何实现的呢？以ArrayBlockingQueue为例！

ArrayBlockingQueue的内部元素都放在一个对象数组:

final Object[] items

向队列中压入元素可以使用offer()方法和put()方法。对于offer()方法，如果当前队列已经满了，它就会立即返回false。如果没有满，则正常执行入队操作。所以不讨论这个方法，需要关注的是put()方法，put()方法也是将元素压入队列末尾。但是如果队列满了，它会一直等待，直到队列中有空闲的位置。

从队列中弹出元素可以使用poll()方法和take()方法，它们都是从队列的头部获得一个元素，不同之处在于：如果队列为空poll()方法会直接返回null,而take()方法会等待，直到队列中有可用的元素。

因此，put() 方法和take()方法才能体现Blocking的关键，为了做好等待和通知这两件事，在ArrayBlockingQueue中定义了以下一些字段

final ReentrantLock lock;
private final Condition notempty;
private final Condition notFull;

当执行take()操作时，如果队列为空，则让当前线程等待在notEmpty上。新元素入队列时，则进行一次notEmpty通知

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            notEmpty.await();
        return dequeue();
    } finally {
        lock.unlock();
    }
}

private E dequeue() {
    // assert lock.getHoldCount() == 1;
    // assert items[takeIndex] != null;
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length)
        takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    notFull.signal();
    return x;
}

可以看到当count==0时，则要求当前线程在notEmpty上等待通知，否则将队列尾部元素出栈，且唤醒了在notFull信号量上等待的线程，当有新元素插入队列时:

public void put(E e) throws InterruptedException {
    checknotnull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == items.length)
            notFull.await();
        enqueue(e);
    } finally {
        lock.unlock();
    }
}
    
private void enqueue(E x) {
    // assert lock.getHoldCount() == 1;
    // assert items[putIndex] == null;
    final Object[] items = this.items;
    items[putIndex] = x;
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
    notEmpty.signal();
}

当队列满时，将线程放在notFull信号量上等待，否则将新元素入队列，且唤醒了在notEmpty信号量上等待的线程。

使用BlockQueue可以轻松实现消息的订阅和发布，但是对于多工程不在同一个jvm中无法使用，可以使用redis的发布/订阅模式来实现。

BlockingQueue

blockingqueue

相关阅读

栏目导航

推荐阅读

热门阅读