多线程应用实例（上半）

生命因何探索

752人浏览 · 2025-08-06 18:25:45

生命因何探索 · 2025-08-06 18:25:45 发布

一、单例模式

单例模式指的是，在一个进程中，限制某个类只能有唯一的实例。就比如JDBC的DataSource大部分只需要一个。

单例模式有很多种写法，最主流的的有两个，“饿汉模式”和“懒汉模式”。

1.1饿汉模式

“饿”的意思是急切，就是要在类被加载的时候就创建出实例，但饿汉模式又是单例模式，所以它不能被多次实例。

具体实现如下

class Singleton{
//static修饰的成员变量是类成员，在类被加载时初始化
    private static Singleton instance=new Singleton();


    public static Singleton getInstance(){
        return instance;
    }
    //关键
    private Singleton(){

    }
}



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

此代码有以下关键点：

static修饰的instance用来保存唯一实例

通过方法给外部使用实例

private修饰的构造方法无法被外部new创建

1.2懒汉模式

“懒”就指的是在实例要被使用时才去创建实例，节省了饿汉直接初始化的开销

相比于饿汉，懒汉初始化为空，后续在创建即可

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

1.3单例模式的线程安全问题（重点）

线程安全问题最容易出现在对变量的修改上

对于饿汉模式，以下操作只涉及到变量的读操作，是比较安全的

而对于懒汉模式，不仅有对变量的修改操作，还有变量的判断，这就很容易出现问题

如果两线程如此执行，那么右边的线程岂不是白费功夫？

1.3.1饿汉模式的原子性问题

这时我们要解决线程执行的原子性问题就可以利用加锁来解决，加锁可以保障锁内操作的原子性。

但是这样还有一个问题，那就是懒汉实例创建好了就没有修改操作了，变为线程安全了，就不需要加锁了，但调用时还是不断加锁阻塞影响性能

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

完善上述代码，解决无效锁的问题

在原基础上在加上判断是否要加锁，让锁只加一次

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

1.3.2饿汉模式的指令重排序问题

指令重排序时编译器优化的一种方式，类似内存可见性问题也是编译器优化导致的。

举例说明：上述代码在JVM层要进行三步

1.分配内存
2.调用构造方法初始化内存
3.把地址赋值给引用变量
正常来说应该是123如此执行，但编译器优化会导致指令执行顺序变为231，132等等。

试想一下如果两个线程如此执行

在t1先执行3指令使instance有了地址，t2就会判定非空拿到return结果最后拿到一个没有内容的实例，这也触发了安全问题。

解决办法就是volatile关键字，告诉编译器这个变量是易变的，你小心点。

二、阻塞队列

2.1阻塞队列和生产者消费者模型

阻塞队列：
遵守“先入先出”原则
具有线程安全特性
当队列满的时候, 继续入队列就会阻塞, 直到有其他线程从队列中取走元素
当队列空的时候, 继续出队列也会阻塞, 直到有其他线程往队列中插入元素

生产者消费者模型：
通过容器来解决生产者和消费者的强耦合问题
生产者产生数据，消费者处理数据
使用阻塞队列之前：消费者要直接调用才能拿到生产者的数据，也就是说生产者和消费者的关联很大，具体实例就是生产者包含消费者代码，消费者包含生产者代码

使用阻塞队列后，生产者和消费者之间通过阻塞队列来联系，生产者和消费者之间解耦合

同时阻塞队列也充当缓冲区，具有削峰填谷的作用

为何要在二者之间加入阻塞队列

如果消费者是服务器，在没有阻塞队列的情况下，生产者传递大量请求数据，服务器在处理这些操作就会消耗很多资源，一旦超过服务器承载上限，就会崩掉。而缓冲队列就可以让消费者单位时间处理的内容变化没。那么剧烈，保证正常使用（这里生产者和队列不容易崩，是因为它们执行的操作相比服务器要简单很多，消耗资源少）

阻塞队列代价：
需要更多硬件设备来实现
使生产者和消费者之间通讯时间延长

2.2Java标准库提供的阻塞队列

BlockingQueue<String> queue = new LinkedBlockingQueue<>();

其中（）内写入的内容表示容量

BlockingQueue 是⼀个接⼝. 真正实现的类是 LinkedBlockingQueue

put ⽅法⽤于阻塞式的⼊队列, take ⽤于阻塞式的出队列.

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

具体代码示例

试运行一下，可发现容量为3的队列再加入一个元素就会被阻塞

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

public class Demo52 {

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

        queue.put("a");
        System.out.println("a");
        queue.put("b");
        System.out.println("b");
        queue.put("c");
        System.out.println("c");
        queue.put("d");
        System.out.println("d");

    }
}

这时我们拿出一个元素就可以了

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

public class Demo52 {

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

        queue.put("a");
        System.out.println("a");
        queue.put("b");
        System.out.println("b");
        queue.put("c");
        System.out.println("c");
        queue.take();
        queue.put("d");
        System.out.println("d");

    }
}

2.3生产者消费者模型示例

模拟生产>消费，由于队列容量有限而阻塞

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

public class Demo52 {

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

       Thread t1 = new Thread(() -> {
          int i = 1;
          while (true) {
              try {
                  queue.put(String.valueOf(i));
                  System.out.println("生产"+i);
                  i++;

                  //Thread.sleep(1000);
              } catch (InterruptedException e) {
                  throw new RuntimeException(e);
              }
          }
       });
       Thread t2 = new Thread(() -> {
          while (true) {
              try {
                  Thread.sleep(1000);
                  Integer i = Integer.valueOf(queue.take());
                  System.out.println("消费"+i);
              } catch (InterruptedException e) {
                  throw new RuntimeException(e);
              }
          }
       });
       t1.start();
       t2.start();
       t1.join();
       t2.join();

    }
}

模拟消费>生产，平缓消费速率

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

public class Demo52 {

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

       Thread t1 = new Thread(() -> {
          int i = 1;
          while (true) {
              try {
                  queue.put(String.valueOf(i));
                  System.out.println("生产"+i);
                  i++;

                  Thread.sleep(1000);
              } catch (InterruptedException e) {
                  throw new RuntimeException(e);
              }
          }
       });
       Thread t2 = new Thread(() -> {
          while (true) {
              try {
                  //Thread.sleep(1000);
                  Integer i = Integer.valueOf(queue.take());
                  System.out.println("消费"+i);
              } catch (InterruptedException e) {
                  throw new RuntimeException(e);
              }
          }
       });
       t1.start();
       t2.start();
       t1.join();
       t2.join();

    }
}

2.4阻塞队列实现示例（基于数组）

先实现一个队列

class MyBlockingQuene{
    private String[] arr =null;

    private int head=0;
    private int tail=0;
    private int size=0;

    public MyBlockingQuene(int capacity){
        arr = new String[capacity];

    }
    public void put(String s){
        if(size==arr.length){
            return;
        }
        arr[tail] = s;
        tail++;
        if(tail>=arr.length){
            tail=0;
        }
        size++;
    }
    public String take(){
        if(size==0){
            return null;
        }
        String ret= arr[head];
        head++;
        if(head>=arr.length){
            head=0;
        }
        size--;
        return ret;

    }

}

接下来进行线程安全问题的解决

1.原子性问题，加锁解决，加锁的范围就是进行修改操作的地方

class MyBlockingQuene{
    private String[] arr =null;

    private int head=0;
    private int tail=0;
    private int size=0;

    public MyBlockingQuene(int capacity){
        arr = new String[capacity];

    }
    public void put(String s){
        synchronized(this){
            if(size==arr.length){
                return;
            }
            arr[tail] = s;
            tail++;
            if(tail>=arr.length){
                tail=0;
            }
            size++;
        }


    }
    public String take(){
        String ret=null;
        synchronized(this){
            if(size==0){
                return null;
            }
            ret= arr[head];
            head++;
            if(head>=arr.length){
                head=0;
            }
            size--;

        }
        return ret;

    }

}

2.实现阻塞，如果队列满了就通过take唤醒put，如果队列为空就通过put唤醒take

class MyBlockingQuene{
    private String[] arr =null;

    private int head=0;
    private int tail=0;
    private int size=0;

    public MyBlockingQuene(int capacity){
        arr = new String[capacity];

    }
    public void put(String s) throws InterruptedException {
        synchronized(this){
            if(size==arr.length){
                //队列为满
               this.wait();
            }
            arr[tail] = s;
            tail++;
            if(tail>=arr.length){
                tail=0;
            }
            size++;
            //队列为空的唤醒
            this.notify();

        }

    }

    public String take() throws InterruptedException {
        String ret="";
        synchronized(this){
            //队列为空进行阻塞
            if(size==0){
                this.wait();
            }
            ret= arr[head];
            head++;
            if(head>=arr.length){
                head=0;
            }
            size--;
            //队列为满的唤醒
            this.notify();

        }
        return ret;

    }

}

public class Demo53 {
    public static void main(String[] args) {
        MyBlockingQuene myQuene = new MyBlockingQuene(10);
        Thread t1 = new Thread(()->{
            int i = 1;
            while (true) {
                try {
                    myQuene.put(String.valueOf(i));
                    System.out.println("生产"+i);
                    i++;

                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        Thread t2 = new Thread(()->{
            while (true) {
                try {
                    Thread.sleep(1000);
                    int i = Integer.parseInt(myQuene.take());
                    System.out.println("消费"+i);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        t1.start();
        t2.start();

    }
}