Java之线程篇七-java黑洞网


// 期望这个类能够有唯一一个实例.
class Singleton {
    private static Singleton instance = new Singleton();
 
    // 通过这个方法来获取到刚才的实例.
    // 后续如果想使用这个类的实例, 都通过 getInstance 方法来获取.
    public static Singleton getInstance() {
        return instance;
    }
 
    // 把构造方法设置为 私有 . 此时类外面的其他代码, 就无法 new 出这个类的对象了.
    private Singleton() { }
}
 
public class Demo17 {
    public static void main(String[] args) {
        Singleton s1 = Singleton.getInstance();
        Singleton s2 = Singleton.getInstance();
        System.out.println(s1 == s2);
    }
}

运行结果

懒汉模式-单线程版

类加载的时候不创建实例 . 第一次使用的时候才创建实例 .


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

懒汉模式-多线程版


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

懒汉模式-多线程版改进

使用双重 if 判定 , 降低锁竞争的频率 .

给 instance 加上了 volatile.


class Singleton {
    private static volatile Singleton instance = null;
    private Singleton() {}
    public static Singleton getInstance() {
        if (instance == null) {
            synchronized (Singleton.class) {
           if (instance == null) {
               instance = new Singleton();
               }
           }
       }
        return instance;
   }
}

上述代码使用volatile的作用

1.防止指令重排序：在多线程环境中，JVM 和现代处理器可能会对代码进行优化，以提高执行效率。这种优化可能包括指令重排序，即改变指令的执行顺序。在 Singleton 的实例化过程中，如果 instance 变量没有被声明为 volatile，那么 JVM 可能会将 instance = new Singleton(); 这行代码分解为三个操作：
分配内存给 instance。
调用 Singleton 的构造函数来初始化对象。
将 instance 指向分配的内存地址。
如果没有 volatile，这三个操作可能会被重排序，导致某个线程在 instance 被正确初始化之前就观察到 instance 不为 null 的情况，但此时 instance 指向的对象可能还没有完全初始化（即构造函数还未完全执行完毕）。这被称为“部分初始化”问题，可能导致程序出现不可预测的行为。通过声明 instance 为 volatile，JVM 会保证 instance 的赋值操作不会被重排序到构造函数调用之前，从而避免这个问题。

2.保证可见性：volatile 关键字还确保了不同线程之间对 instance 变量的可见性。即，当一个线程修改了 instance 的值（从 null 变为指向一个 Singleton 实例的引用），这个修改会立即对其他线程可见。没有 volatile，一个线程可能无法看到另一个线程对 instance 的修改，从而导致它错误地创建另一个 Singleton 实例。

阻塞队列

阻塞队列是一种特殊的队列. 也遵守 "先进先出" 的原则.
阻塞队列能是一种线程安全的数据结构, 并且具有以下特性:

当队列满的时候, 继续入队列就会阻塞, 直到有其他线程从队列中取走元素.
当队列空的时候, 继续出队列也会阻塞, 直到有其他线程往队列中插入元素.

阻塞队列的一个典型应用场景就是 "生产者消费者模型". 这是一种非常典型的开发模型.

生产者消费者模型

生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。
生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取.

1) 阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力.
2) 阻塞队列也能使生产者和消费者之间解耦.

标准库中的阻塞队列

在 Java 标准库中内置了阻塞队列 . 如果我们需要在一些程序中使用阻塞队列 , 直接使用标准库中的即可 .

BlockingQueue 是一个接口 . 真正实现的类是 LinkedBlockingQueue.

put 方法用于阻塞式的入队列 , take 用于阻塞式的出队列 .

BlockingQueue 也有 offer, poll, peek 等方法 , 但是这些方法不带有阻塞特性 .

代码示例


import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
 
public class Demo19 {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> queue = new LinkedBlockingQueue<>();
        queue.put("111");
        queue.put("222");
        queue.put("333");
        queue.put("444");
        String elem = queue.take();
        System.out.println(elem);
        elem = queue.take();
        System.out.println(elem);
        elem = queue.take();
        System.out.println(elem);
        elem = queue.take();
        System.out.println(elem);
        elem = queue.take();
        System.out.println(elem);
    }
}

运行结果

阻塞队列实现

通过 "循环队列" 的方式来实现.
使用 synchronized 进行加锁控制.
put 插入元素的时候, 判定如果队列满了, 就进行 wait. (注意, 要在循环中进行 wait. 被唤醒时不一定队列就不满了, 因为同时可能是唤醒了多个线程).
take 取出元素的时候, 判定如果队列为空, 就进行 wait.


class MyBlockingQueue{
    private String[] data=new String[1000];
    private volatile int head=0;//队列起始位置
    private volatile int tail = 0;//队列结束位置的下一个位置
    private volatile int size=0;//队列中有效元素的个数
 
    public void put(String elem) throws InterruptedException {
        synchronized (this){
            while(size==data.length){
                this.wait();
            }
            data[tail]=elem;
            tail++;
            if(tail==data.length)
                tail=0;
            size++;
            this.notify();
        }
    }
 
    public String take() throws InterruptedException {
        synchronized (this){
            while(size==0){
                this.wait();
            }
            String ret=data[head];
            head++;
            if(head== data.length)
                head=0;
            size--;
            this.notify();
            return ret;
        }
    }
}
 
public class Demo18 {
    public static void main(String[] args) {
        MyBlockingQueue queue=new MyBlockingQueue();
 
        //消费者
        Thread t1=new Thread(()->{
            while(true){
                try {
                    String result=queue.take();
                    System.out.println("消费元素："+result);
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
 
        //生产者
        Thread t2=new Thread(()->{
            int num=1;
            while(true){
                try {
                    queue.put(num+"");
                    System.out.println("生产元素："+num);
                    num++;
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
 
        t1.start();
        t2.start();
    }
}

运行结果

定时器

定时器也是软件开发中的一个重要组件 . 类似于一个 " 闹钟 ". 达到一个设定的时间之后 , 就执行某个指定好的代码.

标准库中的定时器

标准库中提供了一个 Timer 类 . Timer 类的核心方法为 schedule .

schedule 包含两个参数 . 第一个参数指定即将要执行的任务代码 , 第二个参数指定多长时间之后执行 ( 单位为毫秒 ).

代码示例


import java.util.Timer;
import java.util.TimerTask;
 
public class Demo20 {
    public static void main(String[] args) {
        Timer timer=new Timer();
        //给定时器安排一个任务，预定在某个时间去执行
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                System.out.println("3000");
            }
        },3000);
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                System.out.println("2000");
            }
        },2000);
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                System.out.println("1000");
            }
        },1000);
        System.out.println("程序启动");
    }
}

运行结果

实现定时器

定时器的构成:
1.一个带优先级的阻塞队列
为啥要带优先级呢?
因为阻塞队列中的任务都有各自的执行时刻 (delay). 最先执行的任务一定是 delay 最小的. 使用带
优先级的队列就可以高效的把这个 delay 最小的任务找出来.
2.队列中的每个元素是一个 Task 对象.
3.Task 中带有一个时间属性, 队首元素就是即将要执行的任务.
4.同时有一个 worker 线程一直扫描队首元素, 看队首元素是否需要执行

代码示例


import java.util.Comparator;
import java.util.PriorityQueue;
import java.util.TimerTask;
 
class MyTimerTask implements Comparable<MyTimerTask>{
    private Runnable runnable;
    private long time;
    public MyTimerTask(Runnable runnable,long delay){
        this.runnable=runnable;
        this.time=System.currentTimeMillis()+delay;
    }
 
    @Override
    public int compareTo(MyTimerTask o) {
        return (int)(this.time-o.time);
    }
 
    public long getTime(){
        return time;
    }
 
    public Runnable getRunnable() {
        return runnable;
    }
}
 
class MyTimer{
    private PriorityQueue<MyTimerTask> queue=new PriorityQueue<>();
    private Object locker=new Object();
 
    public void schedule(Runnable runnable,long delay){
        synchronized (locker){
            queue.offer(new MyTimerTask(runnable,delay));
            locker.notify();
        }
    }
 
    public MyTimer() {
        Thread t=new Thread(()->{
            while(true) {
                try {
                    synchronized (locker) {
                        while (queue.isEmpty()) {
                            locker.wait();
                        }
                        MyTimerTask task = queue.peek();
                        long curTime = System.currentTimeMillis();
                        if (curTime >= task.getTime()) {
                            task.getRunnable().run();
                            queue.poll();
                        } else {
                            locker.wait(task.getTime() - curTime);
                        }
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        t.start();
    }
}
 
public class Demo21 {
    public static void main(String[] args) {
        MyTimer timer = new MyTimer();
        timer.schedule(new Runnable() {
            @Override
            public void run() {
                System.out.println("3000");
            }
        }, 3000);
        timer.schedule(new Runnable() {
            @Override
            public void run() {
                System.out.println("2000");
            }
        }, 2000);
        timer.schedule(new Runnable() {
            @Override
            public void run() {
                System.out.println("1000");
            }
        }, 1000);
        System.out.println("程序开始执行");
    }
}

线程池

线程池最大的好处就是减少每次启动、销毁线程的损耗。

标准库中的线程池

使用 Executors.newFixedThreadPool(4) 能创建出固定包含 4 个线程的线程池.
返回值类型为 ExecutorService
通过 ExecutorService.submit 可以注册一个任务到线程池中.

代码示例


import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
public class Demo22 {
    public static void main(String[] args) {
        ExecutorService service= Executors.newFixedThreadPool(4);
        service.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println("hello");
            }
        });
    }
}

运行结果

Executors 创建线程池的几种方式

newFixedThreadPool: 创建固定线程数的线程池
newCachedThreadPool: 创建线程数目动态增长的线程池.
newSingleThreadExecutor: 创建只包含单个线程的线程池.
newScheduledThreadPool: 设定延迟时间后执行命令，或者定期执行命令. 是进阶版的 Timer.

Executors 本质上是 ThreadPoolExecutor 类的封装.

实现线程池


import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
 
class MyThreadPool{
    private BlockingQueue<Runnable> queue=new ArrayBlockingQueue<>(1000);
    public void submit(Runnable runnable) throws InterruptedException {
        queue.put(runnable);
    }
 
    public MyThreadPool(int n){
        for (int i = 0; i < n; i++) {
            Thread t=new Thread(()->{
                Runnable runnable= null;
                try {
                    runnable = queue.take();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                runnable.run();
            });
            t.start();
        }
    }
}
 
public class Demo23 {
    public static void main(String[] args) throws InterruptedException {
        MyThreadPool myThreadPool=new MyThreadPool(4);
        for (int i = 0; i < 1000; i++) {
            int id=i;
            myThreadPool.submit(new Runnable() {
                @Override
                public void run() {
                    System.out.println("执行任务："+id);
                }
            });
        }
    }
}

运行结果

线程池的优点

1.减少资源消耗：通过重用已存在的线程，线程池避免了线程创建和销毁所带来的开销。线程创建和销毁是昂贵的操作，因为它们涉及到系统资源的分配和释放。使用线程池可以显著减少这些开销，提高系统的资源利用率。
2.提高响应速度：由于线程池中的线程是预先创建好的，当有新任务到来时，可以立即分配线程去执行，而不需要等待新线程的创建。这可以显著提高系统的响应速度，尤其是在高并发场景下。
3.提高线程的可管理性：线程池提供了一种集中管理线程的方式，包括线程的创建、销毁、调度等。通过线程池，开发者可以更容易地控制系统中线程的数量，避免创建过多的线程导致系统资源耗尽。
4.提供灵活的配置选项：大多数线程池实现都提供了丰富的配置选项，如线程池的大小、任务的队列类型、拒绝策略等。这些配置选项使得开发者可以根据应用程序的具体需求来优化线程池的性能。
5.简化并发编程：线程池隐藏了线程管理的复杂性，使得开发者可以更加专注于业务逻辑的实现，而不是线程的管理。这简化了并发编程的难度，降低了出错的可能性。
6.支持并发任务的执行：线程池可以同时执行多个任务，提高了系统的并发处理能力。这对于需要处理大量并发请求的应用程序来说是非常重要的。
7.提供任务调度功能：一些高级的线程池实现还提供了任务调度的功能，允许开发者按照特定的策略（如定时、周期性等）来执行任务。这进一步增强了线程池的灵活性和功能。

ThreadPoolExecutor的构造方法

上图中最后一个构造函数功能最多，以这个为介绍对象：

corePoolSize：核心线程数

maximumPoolSize：最大线程数

线程池里面的线程数目：[corePoolSize，maximumPoolSize]

keepAlive：允许线程最大的存活时间

unit：时间单位

workQueue：阻塞队列，用来存放线程池中的任务

threadFactory：工厂模式

handler：拒绝策略