阻塞队列介绍

阻塞队列

kafka是目前来说性能最好的消息队列服务器，能处理TB级别的数据

作用:点赞、评论时，服务器会自动给某个用户发送通知

kafka是个框架，如果不用框架还要解决类似问题，就要用到阻塞队列

BlockingQueue

阻塞队列就是一个接口BlockingQueue，java核心API中的接口

作用：解决线程通信的问题。

wait/notify等也可以解决线程通信问题，但更加麻烦，因此如果要在API中选一种方法解决线程通信问题，选择阻塞队列更好
阻塞方法：put、take。
生产者消费者模式
- 生产者：产生数据的线程。
- 消费者：使用数据的线程。
假设线程1为生产者，线程2为消费者，两线程直接打交道可能产生问题：
- 生产速度过快，远快于消费速度。当线程2不再需要数据后，线程1没有被阻塞，仍在占用cpu生产，造成资源浪费
- 生产速度很慢，赶不上消费速度。当线程2把数据使用完后，没有被阻塞，仍占用着cpu获取数据，但取不到，也造成系统资源浪费
BlockingQueue就是在生产者和消费者之间搭建了一个桥梁，构造一个缓冲空间，可避免两线程直接打交道。

线程1用put方法往队列里存数据，线程2用take方法从队列中取数据。故线程1为生产者，线程2为消费者。

线程1将生产的数据放在队列里，若队列未满，则继续生产；若队列满了，线程1就会被阻塞

线程2从队列中取数据，若队列不空，则继续取；若队列为空，线程2被阻塞

阻塞：等着，什么也不做，不会占用cpu资源，对系统性能没有影响。

BlockingQueue的实现类

ArrayBlockingQueue
LinkedBlockingQueue
PriorityBlockingQueue、SynchronousQueue、DelayQueue等

阻塞队列代码实例

public class BlockingQueueTests {

    public static void main(String[] args) {

        // 构造容量为10的阻塞队列
        BlockingQueue queue = new ArrayBlockingQueue(10);
        // 创建一个生产者线程,并启动
        new Thread(new Producer(queue)).start();
        // 创建多个消费者线程,并启动
        new Thread(new Consumer(queue)).start();
        new Thread(new Consumer(queue)).start();
        new Thread(new Consumer(queue)).start();
    }

}

// 生产者类,用于构造生产者线程
class Producer implements Runnable {
    // 要求调用方在调用生产者线程时,要将阻塞队列传入,因为线程是要交给阻塞队列去管理的

    private BlockingQueue<Integer> queue;

    // 阻塞队列:通过泛型声明阻塞队列中存什么
    // 有参构造器:实例化时传入阻塞队列
    public Producer(BlockingQueue<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            // 生产者每隔20s生产一次数据,一共生产100次,将数据交给队列管理
            for (int i = 0; i < 100; i++) {
                Thread.sleep(20);
                queue.put(i);
                // 打印: 哪个线程在生产,生产后阻塞队列中有多少个数
                System.out.println(Thread.currentThread().getName() + "生产:" + queue.size());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

// 消费者
class Consumer implements Runnable {

    private BlockingQueue<Integer> queue;

    public Consumer(BlockingQueue<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            // 消费者只要有数据就一直消费
            while (true) {
                // 消费者消费数据的时间间隔不是确定的,不能写死
                // 大概率生成的随机数比20大,即大概率生产速度快于消费速度
                Thread.sleep(new Random().nextInt(1000));
                // 弹出队首数据
                queue.take();
                // 打印:消费者是谁.消费后阻塞队列里有多少数据
                System.out.println(Thread.currentThread().getName() + "消费:" + queue.size());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}