多线程代码案例

案例一.单例模式

单例模式是一种设计模式;类似于棋谱,有固定套路,针对一些特定场景可以给出一些比较好的解决方案;

只要按照设计模式来写代码,就可以保证代码不会太差,保证了代码的下限;

-------------------------------------------------------------------------------------------------------------------------------

补充:

设计模式是针对编写代码过程中的软性约束: 不是强制的,可以遵守也可以不遵守;

框架是针对编写代码过程中的硬性约束: 针对一些特定的问题场景,大佬们把基本的代码和大部分逻辑已经写好了,留下一些空位,让你在空位上自定义一些逻辑;

库虽然也是别人写好的,但是代码的主体还是由你来完成,你可以决定调用或者不调用;

-------------------------------------------------------------------------------------------------------------------------------

开发过程中,希望有的类在一个进程中不应该存在多个实例(对象),此时就可以使用单例模式,限制某个类只有一个实例;

饿汉模式

饿的意思是"迫切": 在类被加载的时候就会创建出单例的实例;

class Singleton {
    private static Singleton instance = new Singleton();
    //static修饰将instance变成类成员;
    //类成员的初始化就是在Singleton这个类被加载的时候;
    public static Singleton getInstance() {
        return instance;
    }
}

public class demo20 {
    public static void main(String[] args) {
        Singleton s1 = Singleton.getInstance();
        Singleton s2 = Singleton.getInstance();
        System.out.println(s1 == s2);
    }
}
//true

只要不再其他代码中,new这个类,每次需要使用时都通过getInstance()来获取实例,那么此时这个类就是单例的了;

主要要解决的问题:防止别人new这个类的对象

单例模式的核心:将构造方法设为私有的

意味着在类的外面,就无法再构造新的实例了;

private Singleton() {}

通过反射/序列化反序列化等非常规手段还是可以打破单例模式的;

懒汉模式

计算机的"懒"是褒义词: 意思是效率会更高;

推迟了创建实例的时机,第一次使用的时候,才会创建实例;

class SingletonLazy {
    private static SingletonLazy instance = null;
    public static SingletonLazy getInstance() {
        if(instance == null) {
            instance = new SingletonLazy();
        }
        return instance;
    }
    private SingletonLazy() {}
}

public class demo21 {
    public static void main(String[] args) {
        SingletonLazy s1 = SingletonLazy.getInstance();
        SingletonLazy s2 = SingletonLazy.getInstance();
        System.out.println(s1 == s2);
    }
}
//true

思考:如果在多线程环境下,调用getInstance,是否会有问题呢?

饿汉模式没有线程安全问题,但是懒汉模式存在线程安全问题;

原因:多个线程针对一个变量修改可能会产生线程安全问题,但是如果只是读取,则没有问题;

而饿汉模式中的getInstance方法中只有读操作,而懒汉模式中的getInstance方法则有判断和修改赋值操作,故会出现线程安全问题;

解决办法:加锁!

private static Object locker = new Object();
    public static SingletonLazy getInstance() {
        synchronized (locker) {
            if (instance == null) {
                instance = new SingletonLazy();
            }
        }
        return instance;
    }

当前代码的写法,只要调用getInstance,都会触发加锁操作,虽然没有线程安全问题了,但是Instance new出来了之后就都是读操作,此时也会因为加锁,产生阻塞,影响性能;

优化:

public static SingletonLazy getInstance() {
        if(instance == null) {
            synchronized (locker) {
                if (instance == null) {
                    instance = new SingletonLazy();
                }
            }
        }
        return instance;
    }

-------------------------------------------------------------------------------------------------------------------------------

补充:

如果是一个局部变量,每个线程的局部变量都有自己的一份.但是如果是new出来的对象,可以共享;

少数局部变量在线程中不能共用是java自身的限制,在C++,系统原生api中则没有这样的限制;

创建出的局部变量,处于JVM内存的"栈"区域中;

new出来的变量,处于JVM内存的"堆'区域中;

整个JVM进程中,堆只有一份,是线程之间大家共用的;而栈则是每个线程有自己独立的栈;

正因为变量的共享是常态,所以就容易触发多个线程修改同一个变量导致线程安全问题;

-------------------------------------------------------------------------------------------------------------------------------

instance = new SingletonLazy();可以分为三个步骤

1.分配内存空间;(买房)

2.执行构造方法;(装修)

3.内存空间的地址,赋值给引用变量;(收房拿到钥匙)

编译器可能按照123的顺序也可能按照132的顺序来执行;对于单线程来说是没有区别的;

在多线程中按照132的顺序执行可能会出现问题:

若按照此顺序,线程A执行到步骤3时(此时instance的地址就不为null了),线程B进行了第一个if语句的判断并返回了instance,注意此时的instance指向了一个没有初始化的空的内存,故可能会产生线程安全问题;

解决方法:给instance加上volatile关键字;

private static volatile SingletonLazy instance = null;

加上volatile之后就能防止对instance的赋值操作插入到其他操作之间;

因此java的volatile有两个功能:

(1)保证内存可见性;

(2)禁止指令重排序[针对赋值];

实例二.阻塞队列

基本概念

标准库中原有的队列Queue和其子类,默认都是线程不安全的;

阻塞队列,就是在普通队列的基础上做了扩充~

(1)是线程安全的;

(2)具有阻塞特性:

a.如果队列为空,进行出队列操作,此时就会出现阻塞;一直阻塞到其他线程往队列里添加元素为止;

b.如果队列为满,进行入队列操作,此时也会进行阻塞;一直阻塞到其他线程从队列中取出元素为止;

基于阻塞队列,最大的应用场景就是实现"生产者消费者模型";

-------------------------------------------------------------------------------------------------------------------------------

使用生产者消费者模型主要有两方面好处:

(1)服务器之间的"解耦合"(即降低模块之间的关联/影响程度);

"阻塞"队列是代码中的一种数据结构,由于太好用了以至于会被单独封装成一个服务器程序,并且在单独的服务器机器上进行部署,此时这个阻塞队列就有了一个新的名字:"消息队列" (Message Queue, MQ);

让服务器A通过阻塞队列来和服务器B进行交互,此时虽然服务器A和服务器B有与阻塞队列有了耦合,但是服务器A,B与阻塞队列交互的代码几乎不会更改故可以忽略不计;

(2)通过中间的阻塞队列,可以起到"削峰弱谷"的效果:在遇到请求量激增的情况下,可以有效的保护服务器不会被请求冲垮;

假设原来服务器A的写入速度略小于和服务器B的处理速度,当服务器A往阻塞队列中写入速度徒增时,服务器B可以依然按照原有的速度来进行处理,此时阻塞队列起到一个缓冲的作用;

如果是直接调用A收到多少请求B也会收到多少请求,很容易把服务器B给搞崩;

问:a.为什么服务器收到的请求越多,就可能会挂?

服务器每次收到一个请求,处理这个请求的过程就需要执行一系列代码,在执行这些代码的过程中,就会消耗一定量的硬件资源(CPU,内存,硬盘,网络带宽......);当这些请求消耗的总的硬件资源超过了机器能提供的上限,那么此时机器就会出现问题(卡死,程序直接崩溃等等);

b.在请求激增的时候,为什么是B最容易挂?

A的角色是一个"网关服务器",收到客户端的请求,在把请求转发给其他服务器.这样的服务器中的代码,做的工作比较简单,消耗的硬件资源也会更少;同理,阻塞队列也是比较简单的程序,单位请求消耗的硬件资源,也是比较少的;而B这个服务器是真正干活的服务器,要真正完成一系列的业务逻辑,这一系列的工作,代码量非常的庞大,消耗的硬件资源和时间更多;

生产者消费者模型的代价:

1.需要更多的机器,来部署这样的消息队列.(易解决)

2.A与B之间通信的延迟会变长;若对性能要求很高则不太适合;

-------------------------------------------------------------------------------------------------------------------------------

阻塞队列在Java标准库中也提供了现成的封装:BlockingQueue;

offer,poll这些Queue中的方法在BlockingQueue中也有实现;

但是BlockingQueue还有两个专属的方法:put入队列,take出队列;最大的差别在于put和take是可以阻塞的;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class demo22 {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);

        //生产者线程
        Thread t1 = new Thread(() -> {
            int i = 1;
            while(true) {
                try {
                    queue.put(i);
                    System.out.println("生产元素" + i);
                    i++;
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        //消费者线程
        Thread t2 = new Thread(() -> {
            while(true) {
                try {
                    Integer i = queue.take();
                    System.out.println("消费元素" + i);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        t1.start();
        t2.start();
    }
}
/*
消费元素1
生产元素1
生产元素2
消费元素2
生产元素3
消费元素3
生产元素4
消费元素4
生产元素5
消费元素5
生产元素6
消费元素6
......
*/

实际开发中,一般是多个线程生产多个线程消费;

MyBlockingQueue实现

拓展:

//1
head++;
if(head >= this.data.length) {
    head = 0;
}
//2
head++;
head = head % this.data.length;

此时应该优先选择第一种写法,理由如下:

(1)写法1的代码可读性更高;

(2)写法1的效率更高;对第二种写法,CPU计算乘除法是一个比较慢的操作(除非是对2的N次方进行乘除),尤其是除法;对第一种写法是判定,if判断往往是一个非常简单快速的cmp指令;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

class MyBlockingQueue {
    private String[] data = null;

    private volatile int head = 0;
    private volatile int tail = 0;
    private volatile int size = 0;
//避免内存可见性问题,即使加锁后效率低概率小但是还是要预防

    public MyBlockingQueue(int capacity) {
        this.data = new String[capacity];
    }

    public void put(String s) throws InterruptedException {
        synchronized (this) {
            while (size == this.data.length) {
                //return;
                this.wait();
            }
            data[tail] = s;
            tail++;
            while (tail >= data.length) {
                tail = 0;
            }
            size++;

            this.notify();
        }
    }

    public String take() throws InterruptedException {
        String ret = "";
        synchronized (this) {
            while (size == 0) {
                //return null;
                this.wait();
            }
            ret = data[head];
            head++;
            while (head >= data.length) {
                head = 0;
            }
            size--;
            this.notify();
        }
        return ret;
    }
}

public class demo23 {
    public static void main(String[] args) throws InterruptedException {
        MyBlockingQueue queue = new MyBlockingQueue(10);

        //生产者线程
        Thread t1 = new Thread(() -> {
            int i = 1;
            while(true) {
                try {
                    queue.put("" + i);
                    System.out.println("生产元素" + i);
                    i++;
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        //消费者线程
        Thread t2 = new Thread(() -> {
            while(true) {
                try {
                    Integer i = Integer.parseInt(queue.take());
                    System.out.println("消费元素" + i);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        t1.start();
        t2.start();
    }
}

注意:wait()最好被包含在while循环中而不是if判断语句;

原因:wait被唤醒的途径不止notify一种,可能也会由于其他原因唤醒,比如interrupt;

若wait被包含在try-catch语句中,且catch语句中并未结束线程:则使用if语句则会让程序继续往下走,即使队列中还是空着的,也依然会执行下面的代码,此时就会出现bug;故应该将wait语句包含在while的条件循环中;

实例三.线程池

Java标准库中,也提供了现成的线程池(ThreadPoolExecutor)供我们使用;

在java.util.concurrent包中(简称juc);

"池'这种思想,本质上就是能提高线程的效率;

随着业务上对性能的要求越来越高,对应的线程创建/销毁也变得比较频繁,此时的开销就不能忽略不计了;

线程池就是解决上述问题的常见方案,就是把线程提前从系统中申请好,放到一个地方,后面需要使用线程的时候,就直接到这个地方来取,而不是重新从系统申请;线程用完了之后,也是还回到刚才这个地方;

为什么更高效呢?

---------------------------------------------------------------------------------------------------------------------------------

内核态 & 用户态

是操作系统中的概念; 操作系统 = 操作系统内核(核心功能部分,负责完成一个操作系统的核心工作) + 操作系统配套的应用程序;

应用程序都是由内核统一负责管理和服务,内核的工作可能非常繁忙,提交给内核做的任务可能是不可控的;

从系统创建线程,就相当于让银行的人让我复印,这样的逻辑就是调用系统的api,由系统内核执行一系列逻辑来完成这个过程;

直接从线程池里取,就相当于是自助复印,整个过程都是纯用户态代码,都是咱们自己控制的,整个过程更可控高效;

用户态更加高效.

---------------------------------------------------------------------------------------------------------------------------------

java库中线程池的构造方法:

1.int corePoolSize:核心线程数 && int maxinumPoolSize:最大线程数;

说明此线程池可以进行"线程扩容";

在Java标准库中的线程池中,就把里面的线程分为两类:

(1)核心线程[可以理解为最少要有多少个线程];

会始终存在于线程池的内部;

(2)非核心线程[线程扩容的过程中,新增的线程];

繁忙的时候被创建出来,不繁忙了空闲了,就会把这些线程真正的释放掉;

核心线程数和非核心线程数的最大值就叫最大线程数;

2.long keepAliveTime:非核心线程允许摸鱼的最大时间, TimeUnit unit:是一种枚举类型,表示时间的单位;

非核心线程会在线程空闲的时候被销毁;

3.BlockingQueue<Runnable> workQueue:工作队列;

线程池工作的过程就是典型的"生产者消费者模型";

程序员使用的时候,通过形如"submit"这样的方法,把要执行的任务,设定到线程池里;

线程池内部的工作线程,负责执行这些任务;

此处的阻塞队列可以让我们自行指定:

(1)队列的容量 capacity;

(2)队列的类型;

"Runnable"接口本身的含义就是一段可以执行的任务;

4.ThreadFactory threadFactory:线程工厂;

工厂指的是"工厂设计模式"也是一种常见的设计模式;

工厂设计模式,是一种在创建实例时使用的设计模式;

由于构造方法是有"坑"的,通过工厂设计模式来填坑:

构造方法是一种特殊的方法,必须和类名是一样的; 多个版本的构造方法,必须是通过"重载"来实现的;

比如一个类描述一个平面直角坐标系中的一个点,可以用横坐标和纵坐标进行表示,也可以用极坐标进行表示,可此时要调用的构造方法所需传入的参数都是两个double类型,传入的参数个数和类型都相同,无法实现;

为了解决上述问题,就引入了"工厂设计模式",通过"普通方法"(一般是静态方法)来完成对象的构造和初始化操作:

class Point {
}

class PointFactory{
    public static Point makePointByXY(double x, double y) {
        Point p;
        p.setX(x);
        p.setY(y);
        return p;
    }
    public static Point makePointByRA(double r, double a) {
        Point p;
        p.setR(r);
        p.setA(r);
        return p;
    }
}

用来创建对象的static方法就叫"工厂方法",有时候工厂方法也会放到专门的类中实现,用来放工厂方法的类就叫做"工厂类";

ThreadFactory就是Thread类的工厂类.通过这个类,完成Thread的实例创建和初始化;

5.RejectedExecutionHandler handler拒绝策略!!!

如果线程池的任务队列满了,还是要继续给这个队列添加任务,咋办呢??

当队列满了,不要阻塞,而是要明确的拒绝;

Java标准库给出了四种不同的拒绝策略:

(1)AbortPolicy

添加任务的时候,直接抛出异常(RejectedExecutionEception);

(2)CallerRunsPolicy

线程池拒绝执行,但是由调用submit的线程负责执行;

(3)DiscardOldestPolicy

把任务队列中最老的队列踢掉,然后执行添加新的任务;

(4)DiscardPolicy

把任务队列中,最新的任务踢掉;

---------------------------------------------------------------------------------------------------------------------------------

ThreadPoolExcutor功能很强大,但是使用起来很麻烦;

标准库对这个类进一步的封装了一下:Excutors提供了一些工厂方法,可以更方便的构造出线程池;

newCachedThreadPool();

设置了非常大的最大线程数:就可以对线程池进行不断的扩容;

newFixedThreadPool();

把核心线程数和最大线程数设定成了一样的值:固定数量的线程,不会扩容;

newSingleThreadPool();

newScheduledThreadPool();

newWorkStealingPool();

import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class demo25 {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService service = Executors.newFixedThreadPool(4);
        for(int i = 0; i < 100; i++) {
            int id = i;
            service.submit(() -> {
                Thread current = Thread.currentThread();
                System.out.println("hello Thread" + id + ',' + current.getName());
            });
        }
        Thread.sleep(1000);
            //不能直接打印i,lambda表达式中或者匿名类中使用的外部变量要么是常量,要么是未经修改的;
            //此处线程池创建出来的线程都是前台线程,虽然main线程结束了,
            //但是这些线程结束了,但是这些线程池的前台线程依然存在;
        service.shutdown();
            //把所有的线程都终止掉;
        System.out.println("程序退出") ;
    }
}
hello Thread0,pool-1-thread-1
hello Thread4,pool-1-thread-1
hello Thread5,pool-1-thread-1
hello Thread6,pool-1-thread-1
hello Thread7,pool-1-thread-1
hello Thread8,pool-1-thread-1
......
程序退出

线程池需要指定线程个数,多少才是合适的呢?

不能套公式:

(1)一台主机上并不是只运行一个程序;

(2)程序并不会100%跑满CPU,线程工作的过程中,可能会涉及到一些IO操作/阻塞操作主动放弃CPU[sleep,wait,加锁,打印,网络通信,读写硬盘...];

实际开发中,建议通过实验的方式来找到一个合适的线程数量;

---------------------------------------------------------------------------------------------------------------------------------

自己实现一部分功能:

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

class MyThreadPool {
    private BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(1000);
    public MyThreadPool(int n) {
        //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 void submit(Runnable runnable) throws InterruptedException {
        queue.put(runnable);
    }
}

public class demo26 {
    public static void main(String[] args) throws InterruptedException {
        MyThreadPool pool = new MyThreadPool(4);
        for(int i = 0; i < 100; i++) {
            int id = i;
            pool.submit(() -> {
                System.out.println("task" + id + "Thread:" + Thread.currentThread().getName());
            });
        }
    }
}