Java ブロッキングキュー BlockingQueue インスタンスの分析-＆＃＆チュートリアル-php.cn

キュータイプ

無制限キュー 容量制限なし、ストレージによってのみ変更
境界キュー 最大容量を定義します

無限キューに要素を追加するすべての操作は決してブロックされない (スレッドセーフでもある) ため、非常に大きな容量まで増加する可能性があります。無限ブロッキングキューの使用 BlockingQueue プロデューサー/コンシューマーモデルを設計するときに最も重要なことは、プロデューサーがメッセージをキューに追加するのと同じ速度でコンシューマーがメッセージを消費できる必要があることです。そうしないと、メモリが不足し、OutOfMemory 例外がスローされる可能性があります。

データ構造

1. 通常はリンクリストまたは配列を使用して実装されます
2. 通常は FIFO を使用します (最初は、先出し) 機能、両端キューとして設計することもできます
#3. キューの主な操作:入力とキューからの取り出し

Blocking QueueBlockingQueue

定義: スレッド通信では、同時実行性がどれほど高くても、単一の JVM 上では常に 1 つのスレッドのみが通信できます。キューに入れるか、同時にキューにエンキューします。デキュー操作。 BlockingQueue は、明示的な同期を行わずにスレッド間で共有できます。

ブロッキングキューの種類:

Java ブロッキングキュー BlockingQueue インスタンスの分析

JAVA アプリケーションシナリオの場合: スレッドプール、SpringCloud-Eureka 3 レベルキャッシュ、Nacos、MQ、Netty など

共通のブロッキングキュー

ArrayBlockingQueue: 配列によってサポートされる境界付きキュー

アプリケーションシナリオ: スレッドプールには多くのアプリケーションとプロデューサー/コンシューマーモデルがあります
動作原理: ReentrantLock に基づいてスレッドの安全性を確保し、Condition に基づいてキューがいっぱいになったときにブロックを実現します

LinkedBlockingQueue: リンクリストに基づく無制限のキュー (理論的には制限あり)

#PriorityBlockingQueue: By Unboundedプライオリティヒープによってサポートされるプライオリティキュー
DelayQueue: プライオリティヒープによってサポートされる時間ベースのスケジューリングキュー。無制限キュー PriorityQueue および Array に基づいて内部実装されます。無制限キューのベースの拡張実装

使用説明: キューに追加されるオブジェクトは Delayed インターフェイスを実装する必要があり、Delayed は類似のインターフェイス
アプリケーションシナリオ: 映画チケットの販売など
## 動作原理:
キューは時間に応じて優先順位付けされます。クラススレッドプールサイクルの実行を遅延します。

などのメソッドを備えています。作成方法は次のとおりです:

BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<> (666);

ログイン後にコピー

BlockingQueue API

要素の追加:

#メソッド意味add()挿入が成功した場合は true を返し、それ以外の場合は IllegalStateException 例外がスローされます。 put()Insert指定された要素 Queue (キューがいっぱいの場合、挿入するスペースができるまでブロックされます)offer()挿入が成功した場合は true を返し、それ以外の場合は戻りますfalseoffer(E e、長いタイムアウト、TimeUnit 単位)キューに要素を挿入しようとします。キューがいっぱいの場合は、要素が追加されるまでブロックされます。挿入するスペース、およびブロッキングには時間制御があります要素の取得:

メソッド意味take()キューの先頭要素を取得して削除します。キューが空の場合はブロックして待機します。使用可能になる要素poll (長いタイムアウト、TimeUnit 単位)キューの先頭を取得して削除します。必要に応じて、指定された待機時間要素を待ちます。使用可能にするか、タイムアウトが発生した場合は null を返します

ArrayBlockingQueue 源码简解

实现：同步等待队列（CLH）+ 条件等待队列满足条件的元素在CLH队列中等待锁，不满足条件的队列挪到条件等待队列，满足条件后再从 tail 插入 CLH 队列

线程获取锁的条件： 在 CLH 队列里等待的 Node 节点，并且 Node 节点的前驱节点是 Singal。条件等待队列里的线程是无法获取锁的。

/**
 * 构造方法
 * 还有两个构造函数，一个无fair参数，一个可传入集合，创建时插入队列
 * @param capacity 固定容量
 * @param fair 默认是false:访问顺序未指定; true:按照FIFO顺序处理
 */
public ArrayBlockingQueue(int capacity, boolean fair) {
   if (capacity <= 0)
        throw new IllegalArgumentException();
    this.items = new Object[capacity];
    lock = new ReentrantLock(fair); // 根据fair创建对应的锁
    // 条件对象，配合容器能满足业务
    notEmpty = lock.newCondition(); // 出队条件对象
    notFull =  lock.newCondition(); // 入队条件对象
}
/**
 * 入队方法
 * 在队列的尾部插入指定的元素，如果队列已满，则等待空间可用
 */
public void put(E e) throws InterruptedException {
    checkNotNull(e); // 检查put对象是否为空，空抛出异常
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly(); // 若未被中断尝试获取锁，详见下文
    try {
        // 队列中元素的数量 等于 排队元素的长度
        while (count == items.length)
            notFull.await(); // 见下文
        enqueue(e); // 元素入队
    } finally {
        lock.unlock();
    }
}
/**
 * 出队方法
 * 获取队列的头部元素并将其删除，如果队列为空，则阻塞并等待元素变为可用
 */
public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly(); // 见下文
    try {
        while (count == 0)
            notEmpty.await(); // 见下文
        return dequeue(); // 元素出队
    } finally {
        lock.unlock();
    }
}

ログイン後にコピー

令当前线程等待，直到收到信号或被中断详：与此 Condition 关联的锁被自动释放，进入等待，并且处于休眠状态，直到发生以下四种情况之一:

①其他线程调用这个Condition的 signal 方法，当前线程恰好被选为要被唤醒的线程;
②其他线程调用这个条件的 signalAll 方法
③其他线程中断当前线程，支持中断线程挂起;
④一个“虚假的唤醒”发生了。

在这些情况下，在此方法返回之前，当前线程必须重新获得与此条件相关联的锁。当线程返回时，保证它持有这个锁。

如果当前线程有以下两种情况之一:

①在进入该方法时设置中断状态;
②在等待时被中断，支持线程挂起的中断抛出InterruptedException

生产者消费者模式

BlockingQueue 可以在线程之间共享而无需任何显式同步，在生产者消费者之间，只需要将阻塞队列以参数的形式进行传递即可。它内部的机制会自动保证线程的安全性。

生产者：实现了 Runnable 接口，每个生产者生产100种商品和1个中断标记后完成线程任务

@Slf4j
@Slf4j
public class Producer implements Runnable{
    // 作为参数的阻塞队列
    private BlockingQueue<Integer> blockingQueue;
    private final int stopTag;
    /**
     * 构造方法
     * @param blockingQueue
     * @param stopTag
     */
    public Producer(BlockingQueue<Integer> blockingQueue,int stopTag) {
        this.blockingQueue = blockingQueue;
        this.stopTag = stopTag;
    }
    @Override
    public void run() {
        try {
            generateNumbers();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
   private void generateNumbers() throws InterruptedException {
        // 每个生产者都随机生产10种商品
        for (int i = 0; i < 10; i++) {
            int product = ThreadLocalRandom.current().nextInt(1000,1100);
            log.info("生产者{}号,生产了商品,编号为{}",Thread.currentThread().getId(),product);
            blockingQueue.put(product);
        }
        // 生产终止标记
        blockingQueue.put(stopTag);
        log.info("生产者{}号,生产了第终止标记编号{}",Thread.currentThread().getId(),Thread.currentThread().getId());
    }
}

ログイン後にコピー

消费者：消费者拿到终止消费标记终止消费，否则消费商品，拿到终止标记后完成线程任务

@Slf4j
public class Consumer implements Runnable{
    // 作为参数的阻塞队列
    private BlockingQueue<Integer> queue;
    private final int stopTage;
    public Consumer(BlockingQueue<Integer> queue, int stopTage) {
        this.queue = queue;
        this.stopTage = stopTage;
    }
    @Override
    public void run() {
        try {
            while (true) {
                Integer product = queue.take();
                if (product.equals(stopTage)) {
                    log.info("{}号消费者，停止消费,因为拿到了停止消费标记",Thread.currentThread().getId());
                    return;
                }
                log.info("{}号消费者,拿到的商品编号:{}",Thread.currentThread().getId(),product);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

ログイン後にコピー

客户端类： 创建与计算机 CPU 核数相同的线程数，与 16个生产者

public class ProductConsumerTest {
    public static void main(String[] args) {
        // 阻塞队列容量
        int blockingQueueSize = 10;
        // 生产者数量
        int producerSize = 16;
        // 消费者数量 = 计算机线程核数 8
        int consumerSize = Runtime.getRuntime().availableProcessors();
        // 终止消费标记
        int stopTag = Integer.MAX_VALUE;
        BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<>(blockingQueueSize);
        // 创建16个生产者线程
        for (int i = 0; i < producerSize; i++) {
            new Thread(new Producer(blockingQueue, stopTag)).start();
        }
        // 创建8个消费者线程
        for (int j = 0; j < consumerSize; j++) {
            new Thread(new Consumer(blockingQueue, stopTag)).start();
        }
    }
}

ログイン後にコピー

延迟队列 DelayQueue

定义： Java 延迟队列提供了在指定时间才能获取队列元素的功能，队列头元素是最接近过期的元素。没有过期元素的话，使用 poll() 方法会返回 null 值，超时判定是通过getDelay(TimeUnit.NANOSECONDS)方法的返回值小于等于 0 来判断。延时队列不能存放空元素。

/**
 * 电影票类，实现了Delayed接口，重写 compareTo 和 getDelay方法
 */
public class MovieTicket implements Delayed {
    //延迟时间
    private final long delay;
    //到期时间
    private final long expire;
    //数据
    private final String msg;
    //创建时间
    private final long now;
    public long getDelay() {
        return delay;
    }
    public long getExpire() {
        return expire;
    }
    public String getMsg() {
        return msg;
    }
    public long getNow() {
        return now;
    }
    /**
     * @param msg 消息
     * @param delay 延期时间
     */
    public MovieTicket(String msg , long delay) {
        this.delay = delay;
        this.msg = msg;
        expire = System.currentTimeMillis() + delay;    //到期时间 = 当前时间+延迟时间
        now = System.currentTimeMillis();
    }
    /**
     * @param msg
     */
    public MovieTicket(String msg){
        this(msg,1000);
    }
    public MovieTicket(){
        this(null,1000);
    }
    /**
     * 获得延迟时间   用过期时间-当前时间,时间单位毫秒
     * @param unit
     * @return
     */
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(this.expire
                - System.currentTimeMillis() , TimeUnit.MILLISECONDS);
    }
    /**
     * 用于延迟队列内部比较排序  当前时间的延迟时间 - 比较对象的延迟时间
     * 越早过期的时间在队列中越靠前
     * @param delayed
     * @return
     */
    @Override
    public int compareTo(Delayed delayed) {
        return (int) (this.getDelay(TimeUnit.MILLISECONDS)
                - delayed.getDelay(TimeUnit.MILLISECONDS));
    }
}

ログイン後にコピー

测试类：

public static void main(String[] args) {
    DelayQueue<MovieTicket> delayQueue = new DelayQueue<MovieTicket>();
    MovieTicket ticket = new MovieTicket("电影票1",10000);
    delayQueue.put(ticket);
    MovieTicket ticket1 = new MovieTicket("电影票2",5000);
    delayQueue.put(ticket1);
    MovieTicket ticket2 = new MovieTicket("电影票3",8000);
    delayQueue.put(ticket2);
    log.info("message:--->入队完毕");
    while( delayQueue.size() > 0 ){
        try {
            ticket = delayQueue.take();
            log.info("电影票出队:{}",ticket.getMsg());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

ログイン後にコピー

从运行结果可以看出队列是延迟出队，间隔和我们所设置的时间相同

以上がJava ブロッキングキュー BlockingQueue インスタンスの分析の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。