多线程(代码案例: 单例模式, 阻塞队列, 生产者消费者模型,定时器)-阿里云开发者社区

设计模式是什么

类似于棋谱一样的东西

计算机圈子里的大佬为了能让小菜鸡的代码不要写的太差

针对一些典型的场景, 给出了一些典型的解决方案

这样小菜鸡们可以根据这些方案(ACM里面叫板子, 象棋五子棋里叫棋谱, 咱这里叫设计模式), 略加修改, 这样代码再差也差不到哪里去 …

单例模式

单例模式 => 单个对象(实例)

在有些场景中, 有些特定的类, 只允许创建出一个实例, 不应该创建出多个实例

单例模式就是巧用 Java 的语法规则, 达成了某个类只能被创建出一个实例 (单线程多线程下都只能创建出一个实例)

单例模式的实现 – 饿汉模式

// 单例模式 - 饿汉模式
class Singleton {
  // 此处先创建出一个实例 (类加载阶段就创建出来了)
    private static Singleton singleton = new Singleton();
  
  // 如果使用该唯一实例, 统一通过 Singleton.getSingleton() 方法使用
    public static Singleton getSingleton() {
        return singleton;
    }
    
    // 将构造方法设置为私有, 即不可再创建实例
    private Singleton() {}
}
public class Main{
    public static void main(String[] args) {
        Singleton s = Singleton.getSingleton();
        Singleton ss = Singleton.getSingleton();
        System.out.println("s == ss : " + (s==ss));
    }
}

运行结果

单例模式的实现 – 懒汉模式

// 单例模式 - 懒汉模式
class SingletonLazy {
    volatile private static SingletonLazy singletonLazy = null; //volatile 保证内存可见性, 即

    public static SingletonLazy getInstance() {
        if(singletonLazy == null) { //判断是否要加锁
            synchronized (SingletonLazy.class) {
                if(singletonLazy == null) { //判断是否要创建对象
                    singletonLazy = new SingletonLazy();
                }
            }
        }
        return singletonLazy;
    }

    private SingletonLazy() {}
}

public class Main {
    public static void main(String[] args) {
        SingletonLazy s = SingletonLazy.getInstance();
        SingletonLazy ss = SingletonLazy.getInstance();
        System.out.println("s == ss : " + (s == ss));
    }
}

运行结果

这段代码挺有意思的, 其中值得关注的点挺多

双层 if :

外层 if 判断里面的 sychronized ,加锁操作是否要执行, 如果 singletonLazy 对象已存在, 就不用再进行加锁,创建对象的操作了 (sychronized 操作比 if 操作消耗要多的多, 如果不追求性能, 外层 if 可以不要)

内层 if 判断是否要创建对象, 如果 singletonLazy 未存在, 就创建. 多线程环境下可能会出现同时创建多个对象的情况 (不满足单例模式的要求), 因此我们对内层 if 判断及里面的创建对象进行加锁, 由于是单例模式 (只有一个类对象), 因此直接对该类对象加锁就好

volatile 保证内存可见性以及禁止指令重排序, 这也是对内层 if :if(singletonLazy == null)的限制, 防止多线程环境下出现 “类似脏读的问题”

警告: sychronized 能够保证原子性, 但是 sychronized 能否保证内存可见性, 这里是存疑的 (有的资料说 sychronized 不能保证内存可见性, 因此保险起见, 这里是 volatile 也加上的 …)

懒汉模式和饿汉模式的区别

懒汉模式就是不直接创建对象(实例), 什么时候用到, 才创建对象(实例)

饿汉模式就是直接创建对象, 需要用的时候可以直接用, 不用再等待实例的创建等过程

阻塞队列

阻塞队列, 也是队列, 因此具有特点 – 先进先出, 后进后出

阻塞:

如果队列为空,执行出队列操作, 就会阻塞, 直到其他线程往队列里添加元素 (队列不空)

如果队列已满,执行入队列操作, 也会阻塞, 直到其他线程从队列里取走元素 (队列不满)

Java 标准库实现的阻塞队列

public class Main{
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<>();

        blockingQueue.put("hello");
        System.out.println(blockingQueue.take());
        // 其实在多线程环境下使用阻塞效果更明显, 这里只是单纯当初队列来用了
    }
}

运行结果

手写阻塞队列

不考虑泛型, 单纯考虑队列中元素为 Integer 类型, 使用数组实现循环队列

// 手写阻塞队列 (不考虑泛型, 单纯的 Integer 类型, 循环数组实现)
class MyBlockingQueue {
    private int[] items = new int[1000];
    private int head = 0; //头指针
    private int tail = 0; //尾指针
    private int size = 0; //已有元素的数量

    // 入队列
    public void put(int value) throws InterruptedException {
        synchronized (this) {
            while(size == items.length) { // while 是精髓, 如果 put 操作被唤醒后, 又因某些原因队列又满了, 这里的 while 可以达到多次判断的效果(这也是 Java 标准库阻塞队列的写法)
                this.wait();
            }
            items[tail] = value;
            tail++;
            if(tail >= items.length) tail = 0;
            size++;

            this.notify();
        }
    }

    // 出队列
    public Integer take() throws InterruptedException {
        int val;
        synchronized (this) {
            while (this.size == 0) { //这个 while 的作用同上
                this.wait();
            }
            val = items[head];
            head++;
            if(head >= items.length) {
                head = 0;
            }
            size--;

            this.notify();
        }
        return val;
    }
}

public class Main{
    public static void main(String[] args) throws InterruptedException {
        MyBlockingQueue queue = new MyBlockingQueue();
        Thread t1 = new Thread(() -> {
            try {
                System.out.println(queue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        Thread t2 = new Thread(() -> {
            try {
                queue.put(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        t1.start();
        Thread.sleep(3000); //这里是为了让 t1线程先执行, 让队列阻塞掉 (如果 t2 先执行, 那么 t1 执行的时候, 队列内就会有数据, 就不会产生阻塞的效果了) (不加 Thread.sleep(3000) 的话, t1 和 t2 谁先执行, 是不一定的[该死的随机调度, 抢占式执行 ...])
        t2.start();

        t1.join();
        t2.join();
    }
}

运行结果

生产者消费者模型

作用

实现发送方和接收方的解耦合

削峰填谷

其实就是阻塞队列的简单使用

// 简单实现生产者消费者模型
public class Main {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<>();

        // 消费者
        Thread customer = new Thread(() -> {
            while(true) {
                try {
                    System.out.println("消费元素 " + blockingQueue.take());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        // 生产者
        Thread producer = new Thread(() -> {
            int x = 0;
            while(true) {
                try {
                    blockingQueue.put(x);
                    System.out.println("生产元素 " + x++);
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        customer.start();
        producer.start();

        customer.join();
        producer.join();
    }
}

运行结果

这里代码未结束, 而是一直执行下去, 并且代码逻辑是先生产, 再消费

定时器

作用

让一个任务在指定时间运行

Java 标准库提供了 “定时器”

// 定时器的简单使用
public class Main{
    public static void main(String[] args) {
        System.out.println("程序启动");

        Timer timer = new Timer();

        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                System.out.println("定时器1 任务执行");
            }
        }, 1000);

        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                System.out.println("定时器2 任务执行");
            }
        }, 2000);

        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                System.out.println("定时器3 任务执行");
            }
        }, 3000);
    }
}

运行结果

手写一个定时器

定时器的核心:

有一个扫描线程, 当任务时间到的时候执行任务

有一个数据结构来被注册任务

数据结构使用阻塞优先级队列 (可保证线程安全), 也可根据任务的执行时间进行排序

每个任务包含两部分 (任务的内容, 执行时间)

// 手写定时器

// 任务
class MyTask implements Comparable<MyTask> {
    // 要执行的任务
    private Runnable runnable;
    // 要执行的时间
    private long time;

    public MyTask(Runnable runnable, long time) {
        this.runnable = runnable;
        this.time = time;
    }

    // 获取任务的执行时间
   public long getTime() {
       return time;
   }

   // 执行任务
   public void run() {
       runnable.run();
   }

   // 这里定义了 阻塞优先级队列 的排序规则, 别死记, 当场试一试
    @Override
    public int compareTo(MyTask o) {
        return (int)(this.time - o.time);
    }
}

// 定时器
class MyTimer {
    // 扫描线程
    private Thread t = null;

    // 使用阻塞优先队列, 来保存任务
    private PriorityBlockingQueue<MyTask> queue = new PriorityBlockingQueue<>();

    public MyTimer() {
        // 只要调用了构造方法,即创建了定时器, 扫描器就会一直扫任务队列, 看是否有任务需要执行
        t = new Thread(() -> {
            while(true) {
                try {
                    synchronized (this) {
                        // 取出队首元素, 判断是否任务时间已到
                        MyTask myTask = queue.take();
                        long curTime = System.currentTimeMillis();
                        if(curTime < myTask.getTime()) {
                            // 任务时间未到, 丢回任务队列
                            queue.put(myTask);
                            this.wait(myTask.getTime() - curTime); //这里的设计很巧妙, wait 既保证可以在当前队列最早的任务可以及时执行, 当新的任务来临时 notify 也可以将扫描线程唤醒, 也防止了扫描线程一直扫占用 CPU 资源(忙等)
                        } else {
                            myTask.run();
                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        t.start();
    }

    public void schedule(Runnable runnable, long after) {
        MyTask myTask = new MyTask(runnable, (after + System.currentTimeMillis()) );
        queue.put(myTask);
        synchronized (this) {
            this.notify(); //这里唤醒的作用是, 如果当前塞进去的任务的执行时间要先于当前队列中任务执行时间最近的那个, 即可以优先执行本任务
        }
    }
}

public class Main {
    public static void main(String[] args) {
        MyTimer myTimer = new MyTimer();

        myTimer.schedule(new Runnable() {
            @Override
            public void run() {
                System.out.println("hello MyTimer!");
            }
        }, 2000);
        myTimer.schedule(new Runnable() {
            @Override
            public void run() {
                System.out.println("hello MyTimer2!");
            }
        }, 3000);
    }
}