线程池,定时器以及阻塞队列(生产者/消费者模型)

class MyThreadPool {
    private BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
    // 通过这个⽅法, 来把任务添加到线程池中.
    public void submit(Runnable runnable) throws InterruptedException {
        queue.put(runnable);
    }
    // n 表⽰线程池⾥有⼏个线程.
// 创建了⼀个固定数量的线程池.
    public MyThreadPool(int n) {
        for (int i = 0; i < n; i++) {
            Thread t = new Thread(() -> {
                while (true) {
                    try {
                            // 取出任务, 并执⾏~~
                        Runnable runnable = queue.take();
                        runnable.run();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
            t.start();
        }
    }
}
// 线程池
public class Demo {
    public static void main(String[] args) throws InterruptedException {
        MyThreadPool pool = new MyThreadPool(4);
        for (int i = 0; i < 1000; i++) {
            pool.submit(new Runnable() {
                @Override
                public void run() {
// 要执⾏的⼯作
                    System.out.println(Thread.currentThread().getName() + " hell
                }
            });
        }
    }
}

 Timer timer = new Timer();
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                System.out.println("hello");
            }
        }, 3000);

public class MyTimer {
        public void schedule(Runnable command, long after) {
        }
    }

  class MyTask implements Comparable<MyTask> {
                public Runnable runnable;
                // 为了⽅便后续判定, 使⽤绝对的时间戳.
                public long time;
                public MyTask(Runnable runnable, long delay) {
                    this.runnable = runnable;
                   // 取当前时刻的时间戳 + delay, 作为该任务实际执⾏的时间戳
                    this.time = System.currentTimeMillis() + delay;
                }
                @Override
                public int compareTo(MyTask o) {
// 这样的写法意味着每次取出的是时间最⼩的元素.
// 到底是谁减谁?? 俺也记不住!!! 随便写⼀个, 执⾏下, 看看效果~~
                    return (int)(this.time - o.time);
                }
             }

class MyTimer {
    // 核⼼结构
    private PriorityQueue<MyTask> queue = new PriorityQueue<>();
    // 创建⼀个锁对象
    private Object locker = new Object();
    public void schedule(Runnable command, long after) {
// 根据参数, 构造 MyTask, 插⼊队列即可.
        synchronized (locker) {
            MyTask myTask = new MyTask(runnable, delay);
            queue.offer(myTask);
            locker.notify();
        }
    }
}

 // 在这⾥构造线程, 负责执⾏具体任务了.
    public MyTimer() {
        Thread t = new Thread(() -> {
            while (true) {
                try {
                   synchronized (locker) {
// 阻塞队列, 只有阻塞的⼊队列和阻塞的出队列, 没有阻塞的查看队⾸元素.
                        while (queue.isEmpty()) {
                            locker.wait();
                        }
                        MyTask myTask = queue.peek();
                        long curTime = System.currentTimeMillis();
                        if (curTime >= myTask.time) {
// 时间到了, 可以执⾏任务了
                            queue.poll();
                            myTask.runnable.run();
                        } else {
// 时间还没到
                            locker.wait(myTask.time - curTime);
                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        t.start();
    }

例如在"秒杀"场景下,服务器同⼀时刻可能会收到⼤量的⽀付请求.如果直接处理这些⽀付请求, 服务
   器可能扛不住(每个⽀付请求的处理都需要⽐较复杂的流程).这个时候就可以把这些请求都放到⼀个
   阻塞队列中,然后再由消费者线程慢慢的来处理每个⽀付请求. 这样做可以有效进行"削峰",防止服务器被突然到来的⼀波请求直接冲垮.

⽐如过年⼀家⼈⼀起包饺⼦.⼀般都是有明确分⼯,⽐如⼀个⼈负责擀饺⼦⽪,其他⼈负责包.擀饺⼦
⽪的⼈就是"⽣产者",包饺⼦的⼈就是"消费者".
擀饺⼦⽪的⼈不关⼼包饺⼦的⼈是谁(能包就⾏,⽆论是⼿⼯包,借助⼯具,还是机器包),包饺⼦的⼈也
不关⼼擀饺⼦⽪的⼈是谁(有饺⼦⽪就⾏,⽆论是⽤擀⾯杖擀的,还是拿罐头瓶擀,还是直接从超市买
的).

import java.util.Random;

public class BlockingQueue {
    private int[] items = new int[1000];
    private volatile int size = 0;
    private volatile int head = 0;
    private volatile int tail = 0;
    public void put(int value) throws InterruptedException {
        synchronized (this) {
// 此处最好使⽤ while.
// 否则 notifyAll 的时候, 该线程从 wait 中被唤醒,
// 但是紧接着并未抢占到锁. 当锁被抢占的时候, 可能⼜已经队列满了
// 就只能继续等待
            while (size == items.length) {
                wait();
            }
            items[tail] = value;
            tail = (tail + 1) % items.length;
            size++;
           notifyAll();
        }
    }
    public int take() throws InterruptedException {
        int ret = 0;
        synchronized (this) {
            while (size == 0) {
                wait();
            }
            ret = items[head];
            head = (head + 1) % items.length;
            size--;
            notifyAll();
        }
        return ret;
    }
    public synchronized int size() {
        return size;
    }
    // 测试代码
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue blockingQueue = new BlockingQueue();
        Thread customer = new Thread(() -> {
            while (true) {
                try {
                    int value = blockingQueue.take();
                    System.out.println(value);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "消费者");
        customer.start();
        Thread producer = new Thread(() -> {
            Random random = new Random();
            while (true) {
                try {
                    blockingQueue.put(random.nextInt(10000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
           }, "⽣产者");
        producer.start();
        customer.join();
        producer.join();
    }
}