当前位置:   article > 正文

并发编程(四)---设计模式_并发设计模式

并发设计模式

四、设计模式

4.0同步和异步:

同步异步 , 举个例子来说,一家餐厅吧来了5个客人,同步的意思就是说,来第一个点菜,点了个鱼,好, 厨师去捉鱼杀鱼,过了半小时鱼好了给第一位客人,开始下位一位客人,就这样一个一个来,按顺序

相同, 异步呢,异步的意思就是来第一位客人,点什么,点鱼,给它一个牌子,让他去一边等吧,下一位客人接着点菜,点完接着点让厨师做去吧,哪个的菜先好就先端出来,

同步的优点是:同步是按照顺序一个一个来,不会乱掉,更不会出现上面代码没有执行完就执行下面的代码, 缺点:是解析的速度没有异步的快;

异步的优点是:异步是接取一个任务,直接给后台,在接下一个任务,一直一直这样,谁的先读取完先执行谁的, 缺点:没有顺序 ,谁先读取完先执行谁的 ,会出现上面的代码还没出来下面的就已经出来了,会报错;

4.1同步模式之保护性暂停

4.1.1定义:

即:Guraded Suspension,用在一个线程等待另一个线程的执行结果

注意:

  • 有一个结果需要从一个线程传递到另一个线程,让他们关联同一个GuardedObject
  • 如果有结果不断从一个线程到另一个线程,那么可以使用消息队列(生产者/消费者)
  • JDK中,join的实现、Future的实现、采用的就是此模式
  • 因为要等待另一方的记过,因此回归到同步模式

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-J0DcG6Mu-1662119996481)(C:\Users\admin\AppData\Roaming\Typora\typora-user-images\image-20220628153346533.png)]

4.2.2实现

一个简单的一个线程下载,另一个线程输出结果的例子

package com.example.demo;

import com.example.tools.Downloader;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.List;

/**
 * @author 我见青山多妩媚
 * @date Create on 2022/6/28 16:00
 */
@Slf4j
public class Demo20 {
    //线程1等待线程2的下载结果
    public static void main(String[] args) {
        GuardeObject guardeObject = new GuardeObject();
        new Thread(()->{
            //等待结果
            log.debug("等待结果");
            List<String> list = (List<String>)guardeObject.get();
            log.debug("结果大小为:{}",list.size());
        },"线程1").start();

        new Thread(()->{
            log.debug("执行下载");
            try {
                //执行下载
                List<String> download = Downloader.download();
                guardeObject.complete(download);
            } catch (IOException e) {
                e.printStackTrace();
            }
        },"线程2").start();
    }
}

class GuardeObject{
    //结果
    private Object response;

    //获取结果
    public Object get(){
        synchronized (this){
            while (response == null){
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        return response;
    }

    //产生结果
    public void complete(Object response){
        synchronized (this){
            this.response = response;
            this.notifyAll();
        }
    }

}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64

Downloader

package com.example.tools;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

/**
 * @author 我见青山多妩媚
 * @date Create on 2022/6/28 16:06
 */
public class Downloader {
    public static List<String> download() throws IOException {
        HttpURLConnection httpURLConnection = (HttpURLConnection) new URL("https://www.baidu.com/").openConnection();
        List<String> list = new ArrayList<>();
        try (BufferedReader reader =
                    new BufferedReader(new InputStreamReader(httpURLConnection.getInputStream(), StandardCharsets.UTF_8))){
            String line;
            while ((line = reader.readLine())!=null){
                list.add(line);
            }
        }
        return list;
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30

运行结果:

16:17:18.034 [线程1] DEBUG com.example.demo.Demo20 - 等待结果
16:17:18.034 [线程2] DEBUG com.example.demo.Demo20 - 执行下载
16:17:19.466 [线程1] DEBUG com.example.demo.Demo20 - 结果大小为:3

或

16:19:52.222 [线程2] DEBUG com.example.demo.Demo20 - 执行下载
16:19:52.222 [线程1] DEBUG com.example.demo.Demo20 - 等待结果
16:19:53.747 [线程1] DEBUG com.example.demo.Demo20 - 结果大小为:3
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9

具体流程:如果线程1先执行,那么被执行等待,线程2下载完成之后被唤醒,执行线程1(或直接执行线程2,之后执行线程1)

4.2.3功能增强:超时

让一个线程只等待一段时间,超时不等待了

package com.example.demo;

import com.example.tools.Downloader;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.List;

/**
 * @author 我见青山多妩媚
 * @date Create on 2022/6/28 16:00
 */
@Slf4j
public class Demo20 {
    //线程1等待线程2的下载结果
    public static void main(String[] args) {
        GuardeObject guardeObject = new GuardeObject();
        //等待结果
        new Thread(()->{
            //等待结果
            log.debug("等待结果");
            List<String> list = (List<String>) guardeObject.get(2000l);
            log.debug("结果大小为:{}",list.size());
        },"线程1").start();

        //产生结果
        new Thread(()->{
            try {
                //手动睡眠0.5s
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.debug("执行下载");
            try {
                List<String> download = Downloader.download();
                guardeObject.complete(download);
            } catch (IOException e) {
                e.printStackTrace();
            }
        },"线程2").start();
    }
}

class GuardeObject{
    //结果
    private Object response;

    //获取结果
    //设置最大等待时间
    public Object get(Long timeout){
        synchronized (this){
            //开始时间
            long begin = System.currentTimeMillis();
            long end = 0;
            while (response == null){
                //如果超过最大等待时间,那么退出循环 end - timout >= 0
                long waitTime = timeout - end;
                if(waitTime<=0){
                    break;
                }
                try {
                    //不能等待timeout,因为不会退出,仍然会继续等待,还要考虑虚假唤醒的情况,再次进入时等待的时间变长
                    this.wait(waitTime);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                //经历时间
                end = System.currentTimeMillis() - begin;
            }
        }
        return response;
    }

    //产生结果
    public void complete(Object response){
        synchronized (this){
            this.response = response;
            this.notifyAll();
        }
    }

}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84

运行结果:

17:07:38.233 [线程1] DEBUG com.example.demo.Demo20 - 等待结果
17:07:38.744 [线程2] DEBUG com.example.demo.Demo20 - 执行下载
17:07:40.119 [线程1] DEBUG com.example.demo.Demo20 - 结果大小为:3
  • 1
  • 2
  • 3

当让程序sleep(1000)时,因为达到超时时间,运行结果为:

17:12:06.242 [线程1] DEBUG com.example.demo.Demo20 - 等待结果
17:12:07.251 [线程2] DEBUG com.example.demo.Demo20 - 执行下载
Exception in thread "线程1" java.lang.NullPointerException
	at com.example.demo.Demo20.lambda$main$0(Demo20.java:23)
	at java.lang.Thread.run(Thread.java:748)
  • 1
  • 2
  • 3
  • 4
  • 5

因为达到了最大超时时间,所以list结果实际为空,list.size()自然会报错

虚假唤醒的情况

package com.example.demo;

import com.example.tools.Downloader;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.List;

/**
 * @author 我见青山多妩媚
 * @date Create on 2022/6/28 16:00
 */
@Slf4j
public class Demo20 {
    //线程1等待线程2的下载结果
    public static void main(String[] args) {
        GuardeObject guardeObject = new GuardeObject();
        //等待结果
        new Thread(()->{
            //等待结果
            log.debug("等待结果");
            Object o = guardeObject.get(2000l);
            log.debug("结果大小为:{}",o);
        },"线程1").start();

        //产生结果
        new Thread(()->{
            try {
                //手动睡眠1s
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.debug("执行下载");
            guardeObject.complete(null);
        },"线程2").start();
    }
}

class GuardeObject{
    //结果
    private Object response;

    //获取结果
    //设置最大等待时间
    public Object get(Long timeout){
        synchronized (this){
            //开始时间
            long begin = System.currentTimeMillis();
            long end = 0;
            while (response == null){
                //如果超过最大等待时间,那么退出循环 end - timout >= 0
                long waitTime = timeout - end;
                if(waitTime<=0){
                    break;
                }
                try {
                    //虚假唤醒的情况,此时设置为timeout
                    this.wait(timeout);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                //经历时间
                end = System.currentTimeMillis() - begin;
            }
        }
        return response;
    }

    //产生结果
    public void complete(Object response){
        synchronized (this){
            this.response = response;
            this.notifyAll();
        }
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77

运行结果:

17:15:39.040 [线程1] DEBUG com.example.demo.Demo20 - 等待结果
17:15:40.037 [线程2] DEBUG com.example.demo.Demo20 - 执行下载
17:15:42.049 [线程1] DEBUG com.example.demo.Demo20 - 结果大小为:null
  • 1
  • 2
  • 3

对虚假唤醒和正常唤醒的解释:

可以看到,最后的结果比一开始定的2s多了1s,就是因为等待时间为timeout,当程序第一次运动到wait时,开始等待2s,程序运行线程2时,等待1s后唤醒了线程1,所以线程1实际等待了1s就被唤醒了,因为线程2赋值的为null,所以线程1继续等待2s,等待2s之后,再去循环后,当前时间已经等待了3s,比规定的2s大,所以退出循环,线程结束。

当使用watiTime时,线程1运行到wait,等待2s,去运行线程2,线程1等待1s被唤醒,此时waitTime=2-1s=1s,所以又等待1s后退出循环,线程结束

join的底层实际就是使用了保护性暂停模式

4.2.4扩展

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-lfCxlvxU-1662119996482)(C:\Users\admin\AppData\Roaming\Typora\typora-user-images\image-20220629171454412.png)]

邮递员送信

package com.example.demo;

import lombok.extern.slf4j.Slf4j;

import java.util.Hashtable;
import java.util.Map;
import java.util.Set;

/**
 * @author 我见青山多妩媚
 * @date Create on 2022/6/28 16:00
 */
@Slf4j
public class Demo20 {
    public static void main(String[] args) {
        for (int i = 0; i < 3; i++) {
            //居民线程
            new People().start();
        }
        //休息1s
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        //给几个人去送信 三个人要信三个邮递员送信
        for (Integer id : Mailboxes.getIds()) {
            new Postman(id,"内容"+id).start();
        }

    }
}

//每个人或者每个邮递员 都是一个线程
@Slf4j(topic = "people")
class People extends Thread{
    @Override
    public void run() {
        GuardedObject guardeObject = Mailboxes.createGuardedObject();
        log.debug("收信id:{}",guardeObject.getId());
        Object mail = guardeObject.get(5000l);
        log.debug("收信id:{},收到信:{}",guardeObject.getId(),mail);
    }
}

@Slf4j(topic = "postman")
class Postman extends Thread{

    private int mailId;

    private Object mail;

    public Postman(int mailId, Object mail) {
        this.mailId = mailId;
        this.mail = mail;
    }

    @Override
    public void run() {
        //获取邮件内容,返回给收信者
        GuardedObject guardedObject = Mailboxes.getGuardedObject(mailId);
        log.debug("送信id:{},内容:{}",mailId,mail);
        guardedObject.complete(mail);
    }
}

class Mailboxes{
    //hashtable 确保多线程环境下运行  信箱
    private static Map<Integer,GuardedObject> boxes = new Hashtable<>();

    private static int id = 1;

    private static synchronized int generateId(){
        return id++;
    }

    public static GuardedObject createGuardedObject(){
        GuardedObject guardeObject = new GuardedObject(generateId());
        boxes.put(guardeObject.getId(),guardeObject);
        return guardeObject;
    }

    //返回所有的id
    public static Set<Integer> getIds(){
        return boxes.keySet();
    }

    //根据id获取对于的object,人拿到信之后,移除信箱
    public static GuardedObject getGuardedObject(Integer id){
        return boxes.remove(id);
    }
}

class GuardedObject{

    private Integer id;

    public Integer getId() {
        return id;
    }

    public void setId(Integer id) {
        this.id = id;
    }

    public GuardedObject() {
    }

    public GuardedObject(Integer id) {
        this.id = id;
    }

    //结果
    private Object response;

    //获取结果
    //设置最大等待时间
    public Object get(Long timeout){
        synchronized (this){
            //开始时间
            long begin = System.currentTimeMillis();
            long end = 0;
            while (response == null){
                //如果超过最大等待时间,那么退出循环 end - timout >= 0
                long waitTime = timeout - end;
                if(waitTime<=0){
                    break;
                }
                try {
                    //不能等待timeout,因为不会退出,仍然会继续等待,还要考虑虚假唤醒的情况,再次进入时等待的时间变长
                    this.wait(waitTime);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                //经历时间
                end = System.currentTimeMillis() - begin;
            }
        }
        return response;
    }

    //产生结果
    public void complete(Object response){
        synchronized (this){
            this.response = response;
            this.notifyAll();
        }
    }

}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115
  • 116
  • 117
  • 118
  • 119
  • 120
  • 121
  • 122
  • 123
  • 124
  • 125
  • 126
  • 127
  • 128
  • 129
  • 130
  • 131
  • 132
  • 133
  • 134
  • 135
  • 136
  • 137
  • 138
  • 139
  • 140
  • 141
  • 142
  • 143
  • 144
  • 145
  • 146
  • 147
  • 148
  • 149
  • 150
  • 151

执行结果:

17:18:03.942 [Thread-0] DEBUG people - 收信id:1
17:18:03.942 [Thread-1] DEBUG people - 收信id:2
17:18:03.942 [Thread-2] DEBUG people - 收信id:3
17:18:04.950 [Thread-5] DEBUG postman - 送信id:1,内容:内容1
17:18:04.950 [Thread-3] DEBUG postman - 送信id:3,内容:内容3
17:18:04.950 [Thread-4] DEBUG postman - 送信id:2,内容:内容2
17:18:04.950 [Thread-0] DEBUG people - 收信id:1,收到信:内容1
17:18:04.950 [Thread-2] DEBUG people - 收信id:3,收到信:内容3
17:18:04.950 [Thread-1] DEBUG people - 收信id:2,收到信:内容2
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9

4.2 异步模式:生产者—消费者

4.2.1 定义
  • 与前面的保护性暂停中的GuardObject不同,不需要产生处理结果和消费结果的线程一一对应
  • 消费队列可以用来平衡生产和消费的线程资源
  • 生产者仅负责产生结果数据,不关心数据该如何处理,而消费者专心处理结果数据
  • 消息队列是有容量限制的,满时不会再加入数据,空时不会在消耗数据
  • JDK中各种阻塞队列,采用的就是这种模式

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Zh2noWAO-1662119996483)(C:\Users\admin\AppData\Roaming\Typora\typora-user-images\image-20220702153810429.png)]

4.2.2 实现

生产者生产,消费者消费的一个简单例子:

生产者不断生产,生产数量小于等于消息队列的长度(capacity),消费者不断消费,直到生产者不在生产

package com.example.MessageQueue;

import lombok.extern.slf4j.Slf4j;

import java.util.LinkedList;

/**
 * @author 我见青山多妩媚
 * @date Create on 2022/7/2 15:52
 */
@Slf4j(topic = "demo")
public class Demo21 {
    public static void main(String[] args) {
        MessageQueue messageQueue = new MessageQueue(2);
        //生产者
        for (int i = 0; i < 3; i++) {
            int id = i;
            new Thread(()->{
                messageQueue.put(new Message(id,"值"+id));
            },"生产者"+i+"号").start();

        }

        new Thread(()->{
            while (true){
                try {
                    //等待1s
                    Thread.sleep(1000);
                    Message take = messageQueue.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"消费者").start();
    }
}

@Slf4j(topic = "MessageQueue")
class MessageQueue{
    //消息集合
    private final LinkedList<Message> list = new LinkedList<>();
    //队列容量
    private final int capacity;

    public MessageQueue(int capacity) {
        this.capacity = capacity;
    }

    //获取消息
    public Message take(){
        //检查队列是否为空
        synchronized (list){
            //如果为空,持续等待
            while (list.isEmpty()){
                try {
                    log.debug("队列为空,消费者线程等待");
                    list.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("消费者消费");
            //从队列头部获取消息返回
            Message message = list.removeFirst();
            //取走之后,唤醒等待的加入线程
            list.notifyAll();
            return message;
        }
    }

    //存入消息
    public void put(Message message){
        synchronized (list){
            //如果队列满了
            while (list.size() == capacity){
                try {
                    log.debug("队列已满,生产者线程等待");
                    list.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("生产者生产{}",message.toString());
            //如果有空位,加入队列尾部
            list.addLast(message);
            //加入之后,唤醒等待的线程
            list.notifyAll();
        }
    }
}

@Slf4j(topic = "message")
final class Message{
    //确保不线程安全,不加入set方法
    private final int id;
    private final Object message;

    public int getId() {
        return id;
    }

    public Object getMessage() {
        return message;
    }


    public Message(int id, Object message) {
        this.id = id;
        this.message = message;
    }

    @Override
    public String toString() {
        return "Message{" +
                "id=" + id +
                ", message=" + message +
                '}';
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115
  • 116
  • 117
  • 118
  • 119

运行结果:

16:22:58.446 [生产者0号] DEBUG MessageQueue - 生产者生产Message{id=0, message=值0}
16:22:58.449 [生产者2号] DEBUG MessageQueue - 生产者生产Message{id=2, message=值2}
16:22:58.449 [生产者1号] DEBUG MessageQueue - 队列已满,生产者线程等待
16:22:59.453 [消费者] DEBUG MessageQueue - 消费者消费
16:22:59.453 [生产者1号] DEBUG MessageQueue - 生产者生产Message{id=1, message=值1}
16:23:00.468 [消费者] DEBUG MessageQueue - 消费者消费
16:23:01.481 [消费者] DEBUG MessageQueue - 消费者消费
16:23:02.488 [消费者] DEBUG MessageQueue - 队列为空,消费者线程等待
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8

可以看到,定义的capacity=2,生产者线程数为3,所以生产两个消息之后生产者线程等待,消费者线程进行消费,消费之后唤醒等待的生产者线程,生产者继续生产,消费者继续消费,直到队列为空

4.3同步模式之顺序控制

4.3.1固定运行顺序

比如先打印2之后才打印1

wait notify版

无论怎样,都是先打印出2,再打印出1

package com.example.MessageQueue;

import lombok.extern.slf4j.Slf4j;

/**
 * @author 我见青山多妩媚
 * @date Create on 2022/7/5 18:05
 */
@Slf4j
public class Demo31 {
    static final Object lock = new Object();

    //表示2线程是否打印过
    static boolean isLock = false;
    public static void main(String[] args) {
        Thread t1 = new Thread(()->{
            synchronized (lock){
                while (!isLock){
                    try {
                        lock.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                log.debug("1");
            }
        },"t1");

        Thread t2 = new Thread(()->{
            synchronized (lock){
                log.debug("2");
                isLock = true;
                lock.notify();
            }
        },"t2");

        t1.start();
        t2.start();
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40

运行结果:

18:12:44.364 [t2] DEBUG com.example.MessageQueue.Demo31 - 2
18:12:44.370 [t1] DEBUG com.example.MessageQueue.Demo31 - 1
  • 1
  • 2

ReentrantLock await & signal

package com.example.MessageQueue;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author 我见青山多妩媚
 * @date Create on 2022/7/5 18:13
 */
@Slf4j
public class Demo32 {
    static ReentrantLock lock = new ReentrantLock();
    static Condition condition = lock.newCondition();
    //表示2线程是否打印过
    static boolean isLock = false;
    public static void main(String[] args) {
        Thread t1 = new Thread(()->{
            lock.lock();
            try {
                while (!isLock){
                    try {
                        condition.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                log.debug("1");
            }finally {
                lock.unlock();
            }
        },"t1");

        Thread t2 = new Thread(()->{
            lock.lock();
            try {
                log.debug("2");
                isLock = true;
                condition.signal();
            } finally {
                lock.unlock();
            }
        },"t2");

        t1.start();
        t2.start();
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49

运行结果:

18:16:35.735 [t2] DEBUG com.example.MessageQueue.Demo32 - 2
18:16:35.738 [t1] DEBUG com.example.MessageQueue.Demo32 - 1
  • 1
  • 2

park & unpark

package com.example.MessageQueue;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.locks.LockSupport;

/**
 * @author 我见青山多妩媚
 * @date Create on 2022/7/5 18:17
 */
@Slf4j
public class Demo33 {
    public static void main(String[] args) {
        Thread t1 = new Thread(()->{
            //锁住
            LockSupport.park();
            log.debug("1");
        },"t1");

        Thread t2 = new Thread(()->{
          log.debug("2");
          //唤醒阻塞的t1线程
          LockSupport.unpark(t1);
        },"t2");

        t1.start();
        t2.start();
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29

运行结果:

18:19:09.582 [t2] DEBUG com.example.MessageQueue.Demo33 - 2
18:19:09.585 [t1] DEBUG com.example.MessageQueue.Demo33 - 1
  • 1
  • 2
4.3.2交替输出

要求:

线程1输出’a’五次,线程2输出’b’五次,线程3输出’c’五次。现在要求输出abcabcabcabcabc该怎么实现?

wait & notify

package com.example.DesignModule;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

/**
 * @author 我见青山多妩媚
 * @date Create on 2022/7/5 18:21
 */
@Slf4j
public class Demo34 {
    /**
     * 输出内容  等待标记  下一个标记
     * a        1           2
     * b        2           3
     * c        3           1
     */
    public static void main(String[] args) {
        WaitNotify waitNotify = new WaitNotify(1,5);
        new Thread(()->{
            waitNotify.print("a",1,2);
        }).start();

        new Thread(()->{
            waitNotify.print("b",2,3);
        }).start();

        new Thread(()->{
            waitNotify.print("c",3,1);
        }).start();
    }
}

@Data
@AllArgsConstructor
class WaitNotify {
    //等待标记
    private int flag; //1、2、3  1-->2 , 2-->3 , 3-->1
    //循环次数
    private int loopNumber;

    //打印  [a  1  2]   [b  2  3]  [c  3  1]
    public void print(String str,int waitFlag,int nextFlag){
        for (int i = 0; i < loopNumber; i++) {
            synchronized (this){
                //当前线程和应该执行的比较
                while (flag != waitFlag){
                    try {
                        this.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                System.out.print(str);
                //给下一个线程
                flag = nextFlag;
                //叫醒其他线程
                this.notifyAll();
            }
        }
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63

运行结果:

abcabcabcabcabc
  • 1

await & signal

有些不严谨,因为没用while去解决虚假唤醒,但是仅仅针对于本例子来说没问题

package com.example.DesignModule;

import com.example.tools.Sleep;
import lombok.AllArgsConstructor;
import lombok.Data;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author 我见青山多妩媚
 * @date Create on 2022/7/5 18:36
 */
public class Demo35 {
    public static void main(String[] args) {
        AwaitSignal awaitSignal = new AwaitSignal(5);
        //三间休息室
        Condition a = awaitSignal.newCondition();
        Condition b = awaitSignal.newCondition();
        Condition c = awaitSignal.newCondition();

        new Thread(()->{
            awaitSignal.print("a",a,b);
        }).start();

        new Thread(()->{
            awaitSignal.print("b",b,c);
        }).start();

        new Thread(()->{
            awaitSignal.print("c",c,a);
        }).start();

        Sleep.sleep(1);
        //因为进去之后都休息了,所以需要一个线程去唤醒a休息室的线程
        awaitSignal.lock();
        try {
            System.out.println("主线程发起开始命令");
            a.signal();
        } finally {
            awaitSignal.unlock();
        }
    }
}

@AllArgsConstructor
@Data
class AwaitSignal extends ReentrantLock {
    //循环次数
    private int loopNumber;

    /**
     *
     * @param str 内容
     * @param condition  进入那一间去打印
     * @param nextCondition 下一个休息室
     */
    public void print(String str,Condition condition,Condition nextCondition){
        for (int i = 0; i < loopNumber; i++) {
            lock(); //相当于this.lock();
            try{
                try {
                    condition.await();
                    System.out.print(str);
                    //唤醒下一间休息室
                    nextCondition.signal();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }finally {
                unlock();
            }
        }
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75

运行结果:

主线程发起开始命令
abcabcabcabcabc
  • 1
  • 2

park * unpark

package com.example.DesignModule;

import com.example.tools.Sleep;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.locks.LockSupport;

/**
 * @author 我见青山多妩媚
 * @date Create on 2022/7/5 18:46
 */
@Slf4j
public class Demo36 {
    static Thread t1,t2,t3;
    public static void main(String[] args) {
        ParkUnPark parkUnPark = new ParkUnPark(5);

        t1 = new Thread(()->{
            parkUnPark.print("a",t2);
        });

        t2 = new Thread(()->{
            parkUnPark.print("b",t3);
        });

        t3 = new Thread(()->{
            parkUnPark.print("c",t1);
        });

//        Sleep.sleep(1);
        t1.start();
        t2.start();
        t3.start();

        //主线程先唤醒t1
        log.debug("主线程唤醒t1");
        LockSupport.unpark(t1);
    }
}


@Data
@AllArgsConstructor
class ParkUnPark{
    private int loopNumber;

    public void print(String str,Thread next){
        for (int i = 0; i < loopNumber; i++) {
            LockSupport.park();
            System.out.print(str);
            //唤醒下一个线程
            LockSupport.unpark(next);
        }
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57

运行结果:

19:15:36.449 [main] DEBUG com.example.DesignModule.Demo36 - 主线程唤醒t1
abcabcabcabcabc
  • 1
  • 2

4.4终止模式之两阶段终止模式

使用volatile改进

在一个线程t1中如何优雅的终止t2?

package com.example.DesignModule;

import com.example.tools.Sleep;
import lombok.extern.slf4j.Slf4j;

/**
 * @author 我见青山多妩媚
 * @date Create on 2022/7/6 19:30
 */
@Slf4j
public class Demo37 {
    public static void main(String[] args) {
        TwoPhaseTermination twoPhaseTermination = new TwoPhaseTermination();
        //开始线程
        twoPhaseTermination.start();

        Sleep.sleep(3.5);
        log.debug("停止监控");

        //停止线程
        twoPhaseTermination.stop();
    }

}
@Slf4j
class TwoPhaseTermination{
    //监控线程
    private  Thread monitorThread;

    //停止线程
    private volatile boolean stop = false;

    public void start(){
        monitorThread = new Thread(()->{
            while (true){
                //是否被打断
                if(stop){
                    log.debug("料理后事");
                    break;
                }

                try {
                    Sleep.sleep(1);
                    log.debug("执行监控记录");
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        },"monitor");
        monitorThread.start();
    }

    public void stop(){
        stop = true;
        //目的是如果线程正在睡眠,可以在睡眠时就打断停止,不用等到下次while循环之后再打断
        monitorThread.interrupt();
    }

}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60

运行结果:

19:42:59.777 [monitor] DEBUG com.example.DesignModule.TwoPhaseTermination - 执行监控记录
19:43:00.792 [monitor] DEBUG com.example.DesignModule.TwoPhaseTermination - 执行监控记录
19:43:01.807 [monitor] DEBUG com.example.DesignModule.TwoPhaseTermination - 执行监控记录
19:43:02.279 [main] DEBUG com.example.DesignModule.Demo37 - 停止监控
19:43:02.280 [monitor] DEBUG com.example.DesignModule.TwoPhaseTermination - 执行监控记录
19:43:02.280 [monitor] DEBUG com.example.DesignModule.TwoPhaseTermination - 料理后事
java.lang.InterruptedException: sleep interrupted  //睡眠时被打断
	at java.lang.Thread.sleep(Native Method)
	at com.example.tools.Sleep.sleep(Sleep.java:10)
	at com.example.DesignModule.TwoPhaseTermination.lambda$start$0(Demo37.java:43)
	at java.lang.Thread.run(Thread.java:748)
	
//如果不使用 monitorThread.interrupt(); 不会看到异常信息,时间间隔也不会是3.5s,而是4s
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13

4.5同步模式之Balking

BalKing(犹豫)模式在一个线程发现另一个线程或本线程已经做了某一件相同的事,那么本线程就无需再做了,直接结束返回

package com.example.DesignModule;

import lombok.extern.slf4j.Slf4j;

/**
 * @author 我见青山多妩媚
 * @date Create on 2022/7/7 15:16
 */
@Slf4j
public class Demo38 {
    private volatile boolean starting;

    public void start(){
      log.info("尝试启动线程");
      synchronized (this){
          if(starting){
              return;
          }
          starting = true;
      }

      //真正启动线程
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24

他也经常用来实现线程安全的单例

public final class Singleton{
    private Singleton(){
        
    }
    
    private static Singleton INSTANCE = null;
    
    public static synchronized Singleton getInstance(){
        if(INSTANCE != null){
            return INSTANCE;
        }
        INSTANCE = new Singleton();
        return INSTANCE;
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15

对比一下保护性暂停模式:保护性暂停模式用在一个线程等待另一个线程的执行结果,当条件不满足时线程等待

4.6享元模式

4.6.1简介

享元模式(Flyweight pattern)当需要重用数量有限的同一类对象时

4.6.2体现
4.6.2.1包装类

在JDK中,Boolean、Byte、Short、Integer、Long、Character等包装类提供了一个valueOf方法,例如Long的valueOf

会缓存-128~127之间的Long对象,在这个范围之间会重用对象,大于这个范围,才会创建新Long对象:

public static Long valueOf(long l) {
        final int offset = 128;
        if (l >= -128 && l <= 127) { // will cache
            return LongCache.cache[(int)l + offset];
        }
        return new Long(l);
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

注意:

  • Byte、Short、Long缓存的范围为-128~127
  • Character缓存的范围是0~127
  • Integer的默认范围是-128~127,最小值不能变,但最大值可以通过调整虚拟机参数Djava.lang.Integer.IntegerCache.hign来改变
  • Boolean缓存了TRUE和FALSE
4.6.2Stirng串池
4.6.3BigDecimal 、BigInteger
4.6.3DIY

比如一个线上商城应用,QPS达到数千,如果没吃都重新创建和关闭数据库连接,那么性能会受到很大的影响,这是预先创建好一批连接,放入连接池,一次请求达到后从连接池中获取连接,使用完毕之后再还会连接池,这样既节约了连接的创建和关闭,也实现了连接的复用,不至于让庞大的连接数压垮数据库

package com.example.Seven;

import com.example.tools.Sleep;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.sql.*;
import java.util.Map;
import java.util.Properties;
import java.util.Random;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicIntegerArray;

/**
 * @author 我见青山多妩媚
 * @date Create on 2022/7/14 15:39
 */
@Slf4j
public class Demo03 {
    public static void main(String[] args) {
        Pool pool = new Pool(2);
        for (int i = 0; i < 5; i++) {
            new Thread(()->{
                //使用
                Connection connection = pool.borrow();
                Sleep.sleep(new Random().nextInt(1000));
                //释放
                pool.free(connection);
            }).start();
        }
    }
}

@Data
@Slf4j
class Pool{
    //1、连接池大小  固定大小的连接池
    private final int poolSize;

    //2、连接对象的数组
    private Connection[] connections;

    //3、连接状态数组 0 空闲   1繁忙  防止使用int出现线程不安全问题
    private AtomicIntegerArray states;

    //初始化
    public Pool(int poolSize) {
        this.poolSize = poolSize;
        this.connections = new Connection[poolSize];
        this.states = new AtomicIntegerArray(new int[poolSize]);
        for (int i = 0; i < poolSize; i++) {
            connections[i] = new MockConnection("连接"+(i+1));
        }
    }

    //5、借连接
    public Connection borrow(){
        while (true){
            for (int i = 0; i < poolSize; i++) {
                if (states.get(i) == 0) {
                    //防止线程安全问题,不使用set方法  0--->1
                    if (states.compareAndSet(i,0,1)) {
                        log.debug("borrow  {}",connections[i]);
                        return connections[i];
                    }
                }
            }
            //如果没有空闲连接  当前线程进入等待
            synchronized (this){
                try {
                    log.debug("wait...");
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    //6、还连接
    public void free(Connection conn){
        for (int i = 0; i < poolSize; i++) {
            if (connections[i] == conn) {
                states.set(i,0);
                synchronized (this){
                    log.debug("free  {}",conn);
                    this.notifyAll();
                }
                break;
            }
        }
    }
}

//获取数据库连接
class MockConnection implements Connection{

    private String  name;

    public MockConnection(String name) {
        this.name = name;
    }

    @Override
    public String toString() {
        return "MockConnection{" +
                "name='" + name + '\'' +
                '}';
    }
	//...继承Connection的子类
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111

运行结果:

16:08:35.404 [Thread-2] DEBUG com.example.Seven.Pool - wait...
16:08:35.404 [Thread-1] DEBUG com.example.Seven.Pool - borrow  MockConnection{name='连接2'}
16:08:35.407 [Thread-4] DEBUG com.example.Seven.Pool - wait...
16:08:35.407 [Thread-3] DEBUG com.example.Seven.Pool - wait...
16:08:35.404 [Thread-0] DEBUG com.example.Seven.Pool - borrow  MockConnection{name='连接1'}
  • 1
  • 2
  • 3
  • 4
  • 5

不过以上的代码有些问题没有考虑:

  • 连接的动态增长与收缩(高峰期来时增加,高峰期走后收缩)
  • 连接保活(可用性检测)
  • 等待超时处理(超时之后不等待,饥饿问题)
  • 分布式hash

对于关系型数据库,有比较成熟的连接池可以使用,例如c3p0,druid等

对于更通用的对象池,可以考虑使用Apache commons pool ,例如redis连接池参考jedis中关于连接池的实现

4.7异步模式之工作线程

4.7.1定义

让有限的工作线程(work thread)来轮流异步处理无限多的线程,也可以将其归类为分工模式,他的典型是实现其他线程,也体现了几点设计模式中的享元模式。

例如海底捞的服务员(线程),轮流处理每位客人的点餐(任务),如果为每位客人都配一名专属的服务员,那么成本会很高(对比另一种多线程设计模式:Thread-Per-Message)

注意,不同任务类型应该使用不同的线程池,这样能够避免饥饿,并能提升效率

例如,如果一个餐馆的工人既要招呼客人(任务类型A),又要到后厨做菜(任务类型B)显然效率不咋地,分成服务员(线程池A)与厨师(线程池B)更为合理,当然你能想到更细致的分工

4.7.2饥饿

固定大小的线程池有饥饿现象

  • 两个工人是同一个线程池中的两个线程
  • 他们要做的事情是:为客人点餐和到后厨做菜,这是两个阶段的工作
    • 客人点餐:必须先点完餐,等菜做好,上菜,在此期间处理点餐的工人必须等待
    • 后厨做菜:只有做菜
  • 比如工人A处理了点餐任务,接下里他要等着工人B把菜做好,然后上菜,他俩也配合的挺好
  • 但现在同时来了两个客人,这个时候工人A和工人B都去处理点餐了,这时没人做菜了,死锁
package com.example.Eight;

import lombok.extern.slf4j.Slf4j;

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * @author 我见青山多妩媚
 * @date Create on 2022/7/18 15:43
 */
@Slf4j
public class Demo06 {

    static final List<String> ENUM = Arrays.asList("地三鲜","宫保鸡丁","辣子鸡丁","烤鸡腿");

    static Random RANDOM = new Random();

    static String cooking(){return ENUM.get(RANDOM.nextInt(ENUM.size()));}

    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(2);

        pool.execute(()->{
            //一个线程处理点餐
            log.debug("处理点餐");
            //另一个线程处理做菜
            Future<String> f = pool.submit(() -> {
                log.debug("做菜");
                return cooking();
            });

            try {
                log.debug("上菜:{}",f.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });


        pool.execute(()->{
            log.debug("处理点餐");
            Future<String> f = pool.submit(() -> {
                log.debug("做菜");
                return cooking();
            });

            try {
                log.debug("上菜:{}",f.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60

运行结果:

09:04:58.352 [pool-1-thread-1] DEBUG com.example.Eight.Demo06 - 处理点餐
09:04:58.352 [pool-1-thread-2] DEBUG com.example.Eight.Demo06 - 处理点餐
  • 1
  • 2

可以看到,两个线程都去处理点餐了,没有多的线程去处理做菜的操作,当线程池内再添加一个线程变为三个线程时,运行结果为:

//ExecutorService pool = Executors.newFixedThreadPool(3);
09:06:14.509 [pool-1-thread-2] DEBUG com.example.Eight.Demo06 - 处理点餐
09:06:14.509 [pool-1-thread-1] DEBUG com.example.Eight.Demo06 - 处理点餐
09:06:14.526 [pool-1-thread-3] DEBUG com.example.Eight.Demo06 - 做菜
09:06:14.526 [pool-1-thread-3] DEBUG com.example.Eight.Demo06 - 做菜
09:06:14.526 [pool-1-thread-1] DEBUG com.example.Eight.Demo06 - 上菜:烤鸡腿
09:06:14.526 [pool-1-thread-2] DEBUG com.example.Eight.Demo06 - 上菜:辣子鸡丁
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

这样做确实解决了饥饿的问题,但是总不能多来一个客人就多招一名服务员吧?所以最好的方法就是将不同的任务放在不同的线程池内,服务员放在服务员线程,厨师放在厨师线程,互不干扰,这个例子是吧两个角色全都放在一个线程了,execute()需要一个线程去执行,submit()也需要一个线程去执行,所以当数量为2的时候,自然就不够了

4.7.3饥饿—解决
package com.example.Eight;

import lombok.extern.slf4j.Slf4j;

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * @author 我见青山多妩媚
 * @date Create on 2022/7/18 15:43
 */
@Slf4j
public class Demo06 {

    static final List<String> ENUM = Arrays.asList("地三鲜","宫保鸡丁","辣子鸡丁","烤鸡腿");

    static Random RANDOM = new Random();

    static String cooking(){return ENUM.get(RANDOM.nextInt(ENUM.size()));}

    public static void main(String[] args) {
        //服务员线程池,专门负责点餐
        ExecutorService waiterPool = Executors.newFixedThreadPool(1);
        
        //厨师线程池,专门负责做菜
        ExecutorService cookPool = Executors.newFixedThreadPool(1);

        //一个线程处理点餐
        //同一个线程池内逻辑互不干扰
        waiterPool.execute(()->{
            log.debug("处理点餐");
            //另一个线程处理做菜
            Future<String> f = cookPool.submit(() -> {
                log.debug("做菜");
                return cooking();
            });

            try {
                log.debug("上菜:{}",f.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });


        waiterPool.execute(()->{
            log.debug("处理点餐");
            Future<String> f = cookPool.submit(() -> {
                log.debug("做菜");
                return cooking();
            });

            try {
                log.debug("上菜:{}",f.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65

使用两个不同的线程池,一个专门负责点餐waiterPool,一个专门负责做菜cookPool

运行结果:

09:10:44.395 [pool-1-thread-1] DEBUG com.example.Eight.Demo06 - 处理点餐
09:10:44.402 [pool-2-thread-1] DEBUG com.example.Eight.Demo06 - 做菜
09:10:44.402 [pool-1-thread-1] DEBUG com.example.Eight.Demo06 - 上菜:宫保鸡丁
09:10:44.404 [pool-1-thread-1] DEBUG com.example.Eight.Demo06 - 处理点餐
09:10:44.405 [pool-2-thread-1] DEBUG com.example.Eight.Demo06 - 做菜
09:10:44.405 [pool-1-thread-1] DEBUG com.example.Eight.Demo06 - 上菜:辣子鸡丁
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
4.7.4创建多少线程池合适
  • 过小会导致程序不能充分利用系统资源,容易导致饥饿
  • 过大会导致更多的线程上下文切换,占用更多内存
4.7.4.1CPU密集型运算

通常采用cpu核数+1能够实现最优的CPU利用率,+1是保证当线程中由于缺页故障(操作系统)或其他原因导致暂停时,额外的这个线程就能顶上去,保证CPU时钟周期不被浪费

4.7.4.2I/O密集型运算

CPU不总是处于繁忙状态,例如,当你执行业务计算时,这是会使用CPU资源,但当你执行IO操作时,远程RPC调用时,包括进行数据库操作时,这时候CPU闲下来了,你可以利用多线程提高它的利用率。

经验公式:

线程数 = 核数 * 期望CPU利用率 * 总共时间(CPU计算时间 + 等待时间)/ CPU计算时间

例如4核CPU计算时间是50%,其等待时间是50%,期望CPU被100%利用:

4 * 100% * (50% + 50%)/50% = 8

例如4核CPU计算时间是10%,其等待时间是90%,期望CPU被100%利用:

4 * 100% * (10% + 90%)/10% = 40

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/不正经/article/detail/177835
推荐阅读
相关标签
  

闽ICP备14008679号