线程间的通信

发表于 | 分类于
何时使用 在多线程环境下，有时候一个线程的执行，依赖于另外一个线程的某种状态的改变，这个时候，我 们就可以使用wait与notify或者notifyAll...

1.wait、notify、notifyAll

何时使用 在多线程环境下，有时候一个线程的执行，依赖于另外一个线程的某种状态的改变，这个时候，我 们就可以使用wait与notify或者notifyAll wait跟sleep的区别 wait会释放持有的锁，而sleep不会，sleep只是让线程在指定的时间内，不去抢占cpu的资 源 注意点 wait notify必须放在同步代码块中, 且必须拥有当前对象的锁，即不能取得A对象的锁，而调用B对象 的wait 哪个对象wait，就得调哪个对象的notify notify跟notifyAll的区别 nofity随机唤醒一个等待的线程 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
/**
 * notify跟notifyAll的区别
 *
 * nofity随机唤醒一个等待的线程 notifyAll唤醒所有在该对象上等待的线程
 */
public class Demo1 {
    private static boolean flag =false;
    public static void main(String[] args) throws InterruptedException {

        Object obj = new Object();
        new Thread(()->{
            while(!flag){
                //wait的对象必须放在同步代码块中
                synchronized (obj ){
                    try {
                        obj.wait();
                        Thread.sleep(50L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("flag is false!");
                }
            }

            System.out.println("flag is true!");
        }).start();




        new Thread(()->{
            while(!flag){
                //wait的对象必须放在同步代码块中
                synchronized (obj ){
                    try {
                        obj.wait();
                        Thread.sleep(50L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("flag is false!");
                }
            }

            System.out.println("flag is true!");
        }).start();


        Thread.sleep(50L);

        new Thread(()->{
            flag = true;
            synchronized (obj ) {

                obj.notifyAll();
            }
        }).start();
    }
}

2.等待通知经典模型之生产者消费者

生产者消费者模型一般包括：生产者、消费者、中间商

Producer
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
/**
 * 线程通信之生产者
 */
public class Producer implements Runnable {

    //引入中间商
    private Medium medium;

    public Producer (Medium medium){

        this.medium =medium;
    }
    @Override
    public void run() {
        while (true){
            medium.put();
        }
    }
}
Medium
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
/**
 * 线程通信之中间商
 */
public class Medium {

    private int num = 0;
    private static final int TOTAL = 20;

    /**
     * 接收生产者的消费数据
     */
    public synchronized void put() {

        //判断当前的库存,是否已经是最大库存容量
        if (num < TOTAL) {
            //如果不是,生产完成之后,通知消费者进行消费
            System.out.println("新增库存---当前库存:" + ++num);

            notifyAll();
        } else {
            //如果是,这种通知生产者进行等待
            try {
                System.out.println("新增库存-->库存已满:"+num);
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 获取消费数据
     */
    public  synchronized void take() {

        //判断当前库存是否不足
        if(num>0){
            //如果充足,在消费完成之后通知生产者进行生产
            System.out.println("消费库存---->当前库存容量:"+ --num);
            //通知生产者进行生产
            notifyAll();

        }else {
            //如果不足,通知消费者暂停消费
            System.out.println("消费库存-->库存不足:"+num);
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
Consumer
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
/**
 * 线程通信之生产者
 */
public class Consumer  implements Runnable{

    private Medium medium;

    public Consumer(Medium medium) {
        this.medium = medium;
    }

    @Override
    public void run() {

        while (true){
            medium.take();
        }
    }
}
DemoTest
1
2
3
4
5
6
7
8
9
10
11
12
13
14
public class DemoTest {

    public static void main(String[] args) {

        Medium medium = new Medium();

        new Thread(new Consumer(medium)).start();

        new Thread(new Producer(medium)).start();
        new Thread(new Producer(medium)).start();

    }

}

3. 使用管道流进行通信

以内存为媒介，用于线程之间的数据传输。 主要有面向字节：【PipedOutputStream、PipedInputStream】、面向字符【PipedReader、PipedWriter】

Reader
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
/**
 * 使用管道流通信
 */
public class Reader implements Runnable {

    private PipedInputStream pipedInputStream;

    public Reader(PipedInputStream pipedInputStream) {
        this.pipedInputStream = pipedInputStream;
    }

    @Override
    public void run() {

        if(pipedInputStream!=null){
            String collect = new BufferedReader(new InputStreamReader(pipedInputStream))
                    .lines().collect(Collectors.joining("\n"));
            System.out.println(Thread.currentThread().getName()+"--"+collect);
        }

        try {
            pipedInputStream.close();


        }catch (Exception e){

        }

    }
}
MainTest
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
public class MainTest {

    public static void main(String[] args) throws Exception {
        PipedInputStream pipedInputStream = new PipedInputStream();
        PipedOutputStream pipedOutputStream = new PipedOutputStream();
        pipedOutputStream.connect(pipedInputStream);
        BufferedReader bufferedReader =null;
        try {
            new Thread(new Reader(pipedInputStream)).start();

            bufferedReader=  new BufferedReader(new InputStreamReader(System.in));

            pipedOutputStream.write(bufferedReader.readLine().getBytes());

        } finally {

            pipedOutputStream.close();

            if (bufferedReader!=null){

                bufferedReader.close();

            }

        }
    }
}

4.Thread.join通信及其源码浅析

使用场景：线程A执行到一半，需要一个数据，这个数据需要线程B去执行修改，只有B修改完成之后，A才能 继续操作 线程A的run方法里面，调用线程B的join方法，这个时候，线程A会等待线程B运行完成之后，再接着运行

MainTest:
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
public class MainTest {

    public static void main(String[] args) {

 Thread thread =  new Thread(()->{
            System.out.println(Thread.currentThread().getName()+"开始运行了");
            try {
                Thread.sleep(3000L);
            }catch (Exception e){
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName()+"结束运行");
        },"线程1");

        new Thread(()->{

            System.out.println(Thread.currentThread().getName()+"开始运行了");
            thread.start();
            try {
                thread.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName()+"结束运行");

        },"线程2").start();
    }
}
查看join源码:(让线程等待当前线程执行完成)
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
public final synchronized void join(long millis)
    throws InterruptedException {
        long base = System.currentTimeMillis();
        long now = 0;

        if (millis < 0) {
            throw new IllegalArgumentException("timeout value is negative");
        }

        if (millis == 0) {
            while (isAlive()) {
                wait(0);
            }
        } else {
            while (isAlive()) {
                long delay = millis - now;
                if (delay <= 0) {
                    break;
                }
                wait(delay);
                now = System.currentTimeMillis() - base;
            }
        }
    }

5.ThreadLocal的使用

线程变量，是一个以ThreadLocal对象为键、任意对象为值的存储结构。为每个线程单独存放一份变量副本， 也就是说一个线程可以根据一个ThreadLocal对象查询到绑定在这个线程上的一个值。只要线程处于活动状态 并且ThreadLocal实例可访问，那么每个线程都拥有对其本地线程副本的隐式引用变量一个线程消失后，它的 所有副本线程局部实例受垃圾回收（除非其他存在对这些副本的引用） 一般用的比较多的是 ThreadLocal.get: 获取ThreadLocal中当前线程共享变量的值。

ThreadLocal.set: 设置ThreadLocal中当前线程共享变量的值。

ThreadLocal.remove: 移除ThreadLocal中当前线程共享变量的值。 ThreadLocal.initialValue: ThreadLocal没有被当前线程赋值时或当前线程刚调用remove方法后调用get方法， 返回此方法值。

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
/**
 * ThreadLocal
 * 为每个线程单独存放一份变量副本
 */
public class ThreadLocalDemo {
    ThreadLocal<Integer> num = ThreadLocal.withInitial(()->0);
    /**
     * 自增并输出num的值
     */
    public void inCreate(){

        Integer myNum = num.get();
        myNum++;
        System.out.println(Thread.currentThread().getName()
        +"--->"+myNum);

        num.set(myNum);
    }

    public static void main(String[] args) {
        ThreadLocalDemo threadLocalDemo = new ThreadLocalDemo();

        for (int i =1; i <3; i++) {
            int finalI = i;
            new  Thread(()->{

                while (true){
                    threadLocalDemo.inCreate();
                    try {
                        Thread.sleep(finalI *1000L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                }

            }).start();
        }
    }
}

6.Condition的使用

可以在一个锁里面，存在多种等待条件 主要的方法 await signal signalAll

我们可以将之前的中间商的实现改为如下:

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
/**
 * 线程通信之中间商
 */
public class Medium {

    private int num = 0;
    private static final int TOTAL = 20;
    private Lock lock = new ReentrantLock();
    private Condition consumerCondition = lock.newCondition();
    private Condition producerCondition = lock.newCondition();

    /**
     * 接收生产者的消费数据
     */
    public void put() {
        lock.lock();
        try {
            //判断当前的库存,是否已经是最大库存容量
            if (num < TOTAL) {
                //如果不是,生产完成之后,通知消费者进行消费
                System.out.println("新增库存---当前库存:" + ++num);
                Thread.sleep(500L);
                consumerCondition.signalAll();
            } else {
                //如果是,这种通知生产者进行等待
                try {
                    System.out.println("新增库存-->库存已满:" + num);
                    producerCondition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }

    /**
     * 获取消费数据
     */
    public synchronized void take() {
        lock.lock();
        try {

            //判断当前库存是否不足
            if (num > 0) {
                //如果充足,在消费完成之后通知生产者进行生产
                System.out.println("消费库存---->当前库存容量:" + --num);
                //通知生产者进行生产
                Thread.sleep(500L);
                producerCondition.signalAll();

            } else {
                //如果不足,通知消费者暂停消费
                System.out.println("消费库存-->库存不足:" + num);
                try {
                    consumerCondition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }

    }
}
城南少年与猫 wechat
欢迎您扫一扫订阅我的微信公众号！
0%