Java多线程二线程间通信

红太狼 2021-11-01 08:32 363阅读 0赞

线程间通信：

多个线程在处理同一资源，但是

等待唤醒机制

涉及的方法：

1、wait() 让线程处于冻结状态，被wait的线程会被存储到线程池中。

2、notify() 唤醒线程池中的一个线程（任意）

3、notifyAll() 唤醒线程池中的所有线程、、

这些方法都必须定义在同步中，

因为这些方法是用于操作线程状态的方法。

必须明确到底操作的那个锁上的线程。

为什么操作线程的方法wait notify notifyAll定义在了Object中。

因为这些方法是监视器方法，监视器其实就是锁。

锁可以是任意的对象，任意的对象调用的方法一定定义在Object类中。

package com.pzq.thread;
    
    public class SyncDemo3 {
    
        public static void main(String[] args) {
    
            Resource resource = new Resource();
            Input input = new Input(resource);
            OutPut outPut = new OutPut(resource);
            Thread threadInput = new Thread(input);
            Thread threadOuput = new Thread(outPut);
            threadInput.start();
            threadOuput.start();
        }
    
    }
    
    class Input implements Runnable {
    
        Resource r;
    
        public Input(Resource r) {
            this.r = r;
        }
    
        @Override
        public void run() {
            int x = 0;
            while (true) {
                synchronized (r) {
    
                    if (r.flag) {
                        try {
                            r.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    if (x == 0) {
                        r.name = "mike";
                        r.sex = "nan";
                    } else {
                        r.name = "丽丽";
                        r.sex = "女女女女女女女";
                    }
                    r.flag = true;
                    r.notify();
                }
                x = (x + 1) % 2;
            }
        }
    
    }
    
    class OutPut implements Runnable {
    
        Resource r;
    
        public OutPut(Resource r) {
            this.r = r;
        }
    
        @Override
        public void run() {
            while (true) {
                synchronized (r) {
                    if (!r.flag) {
                        try {
                            r.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    System.out.println(r.name + "......" + r.sex);
                    r.flag = false;
                    r.notify();
                }
            }
        }
    
    }
    
    class Resource {
        String name;
        String sex;
        boolean flag = false;
    
        public Resource() {
        }
    
        public Resource(String name, String sex) {
            this.name = name;
            this.sex = sex;
        }
    }

# 生产者消费者 #

## 代码优化，将同步方法封装到属性当中 ##

package com.pzq.thread;
    
    public class SyncDemo3 {
    
        public static void main(String[] args) {
    
            Resource resource = new Resource();
            Input input = new Input(resource);
            OutPut outPut = new OutPut(resource);
            Thread threadInput = new Thread(input);
            Thread threadOuput = new Thread(outPut);
            threadInput.start();
            threadOuput.start();
        }
    
    }
    
    class Input implements Runnable {
    
        Resource r;
    
        public Input(Resource r) {
            this.r = r;
        }
    
        @Override
        public void run() {
            int x = 0;
            while (true) {
                if (x == 0) {
                    r.set("mike", "nan");
                } else {
                    r.set("丽丽", "=========女");
                }
                x = (x + 1) % 2;
            }
        }
    
    }
    
    class OutPut implements Runnable {
    
        Resource r;
    
        public OutPut(Resource r) {
            this.r = r;
        }
    
        @Override
        public void run() {
            while (true) {
                r.out();
            }
        }
    
    }
    
    class Resource {
        private String name;
        private String sex;
        private boolean flag = false;
    
        public Resource() {
        }
    
        public synchronized void set(String name, String sex) {
            if (flag) {
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            this.name = name;
            this.sex = sex;
            flag = true;
            this.notify();
        }
    
        public synchronized void out() {
            if (!flag) {
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println(this.name + "....." + this.sex);
            flag = false;
            this.notify();
        }
    
    }

## 单生产单消费 ##

package com.pzq.thread;
    
    public class SyncDemo4 {
    
        public static void main(String[] args) {
    
            Resource2 resource = new Resource2();
            Producter input = new Producter(resource);
            Customer outPut = new Customer(resource);
            Thread threadInput = new Thread(input);
            Thread threadOuput = new Thread(outPut);
            threadInput.start();
            threadOuput.start();
        }
    
    }
    
    class Producter implements Runnable {
    
        Resource2 r;
    
        public Producter(Resource2 r) {
            this.r = r;
        }
    
        @Override
        public void run() {
            while (true) {
                r.set("烤鸭");
            }
        }
    
    }
    
    class Customer implements Runnable {
    
        Resource2 r;
    
        public Customer(Resource2 r) {
            this.r = r;
        }
    
        @Override
        public void run() {
            while (true) {
                r.out();
            }
        }
    
    }
    
    class Resource2 {
        private String name;
        private boolean flag = false;
        private int count = 1;
    
        public Resource2() {
        }
    
        public synchronized void set(String name) {
            if (flag) {
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            this.name = name + count;
            count++;
            System.out.println(Thread.currentThread().getName() + "生产者+++" + this.name);
            flag = true;
            this.notify();
        }
    
        public synchronized void out() {
            if (!flag) {
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println(Thread.currentThread().getName() + "消费者" + "=================" + this.name);
            flag = false;
            this.notify();
        }
    
    }

## 多生产者多消费者 ##

if判断标记，只有一次，会导致不该运行的线程运行了。出现了数据错误的情况。

while判断标记，解决了线程获取执行权后，是否要运行

notify只能唤醒一个线程，如果本方唤醒了本方，就没有意义了。而且 while判断标记+notify会导致死锁。

notifyAll解决了，本方线程一定会唤醒对方线程。

package com.pzq.thread;
    
    public class SyncDemo4 {
    
        public static void main(String[] args) {
    
            Resource2 resource = new Resource2();
            Producter input = new Producter(resource);
            Customer outPut = new Customer(resource);
            Thread threadInput1 = new Thread(input);
            Thread threadInput2 = new Thread(input);
            Thread threadInput3 = new Thread(input);
            Thread threadOuput1 = new Thread(outPut);
            Thread threadOuput2 = new Thread(outPut);
            Thread threadOuput3 = new Thread(outPut);
            Thread threadOuput4 = new Thread(outPut);
            Thread threadOuput5 = new Thread(outPut);
            threadInput1.start();
            threadInput2.start();
            threadInput3.start();
            threadOuput1.start();
            threadOuput2.start();
            threadOuput3.start();
            threadOuput4.start();
            threadOuput5.start();
        }
    
    }
    
    class Producter implements Runnable {
    
        Resource2 r;
    
        public Producter(Resource2 r) {
            this.r = r;
        }
    
        @Override
        public void run() {
            while (true) {
                r.set("烤鸭");
            }
        }
    
    }
    
    class Customer implements Runnable {
    
        Resource2 r;
    
        public Customer(Resource2 r) {
            this.r = r;
        }
    
        @Override
        public void run() {
            while (true) {
                r.out();
            }
        }
    
    }
    
    class Resource2 {
        private String name;
        private boolean flag = false;
        private int count = 1;
    
        public Resource2() {
        }
    
        public synchronized void set(String name) {
            while (flag) {
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            this.name = name + count;
            count++;
            System.out.println(Thread.currentThread().getName() + "生产者+++" + this.name);
            flag = true;
            this.notifyAll();
        }
    
        public synchronized void out() {
            while (!flag) {
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println(Thread.currentThread().getName() + "消费者" + "=================" + this.name);
            flag = false;
            this.notifyAll();
        }
    
    }

## 多生产者多消费者 1.5新特性 ##

jdk1.5以后将同步和锁封装成了对象。

并将操作锁的隐式方式定义到了该对象中，

将隐式动作变成了显示动作。

Lock接口：出现了替代同步代码块或者同步函数。将同步的隐式锁操作变成现实锁操作。

同时更为灵活，可以一个锁上加上多组监视器。

lock() 获取锁

unlock()：释放锁，通常需要定义finally代码块中

Condition接口：出现替代了Object中的wait notify notifyAll方法。

将这些监视器方法单独进行了封装，变成Condition监视器对象。

可以任意锁进行组合。

await();

signal();

singnalAll

package com.pzq.thread;
    
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    public class SyncDemo4 {
    
        public static void main(String[] args) {
    
            Resource2 resource = new Resource2();
            Producter input = new Producter(resource);
            Customer outPut = new Customer(resource);
            Thread threadInput1 = new Thread(input);
            Thread threadInput2 = new Thread(input);
            Thread threadInput3 = new Thread(input);
            Thread threadOuput1 = new Thread(outPut);
            Thread threadOuput2 = new Thread(outPut);
            Thread threadOuput3 = new Thread(outPut);
            Thread threadOuput4 = new Thread(outPut);
            Thread threadOuput5 = new Thread(outPut);
            threadInput1.start();
            threadInput2.start();
            threadInput3.start();
            threadOuput1.start();
            threadOuput2.start();
            threadOuput3.start();
            threadOuput4.start();
            threadOuput5.start();
        }
    
    }
    
    class Producter implements Runnable {
    
        Resource2 r;
    
        public Producter(Resource2 r) {
            this.r = r;
        }
    
        @Override
        public void run() {
            while (true) {
                r.set("烤鸭");
            }
        }
    
    }
    
    class Customer implements Runnable {
    
        Resource2 r;
    
        public Customer(Resource2 r) {
            this.r = r;
        }
    
        @Override
        public void run() {
            while (true) {
                r.out();
            }
        }
    
    }
    
    class Resource2 {
        private String name;
        private boolean flag = false;
        private int count = 1;
    
        // 创建一个锁对象
        Lock lock = new ReentrantLock();
        // 通过已有的锁获取两组监视器，一组监视生产者，一组监视消费者
        Condition con_producter = lock.newCondition();
        Condition con_customer = lock.newCondition();
    
        public Resource2() {
        }
    
        public void set(String name) {
            lock.lock();
            try {
                while (flag) {
                    try {
                        con_producter.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                this.name = name + count;
                count++;
                System.out.println(Thread.currentThread().getName() + "生产者+++" + this.name);
                flag = true;
                con_customer.signal();
            } finally {
                lock.unlock();
            }
    
        }
    
        public void out() {
            lock.lock();
            try {
                while (!flag) {
                    try {
                        con_customer.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                System.out.println(Thread.currentThread().getName() + "消费者" + "=================" + this.name);
                flag = false;
                con_producter.signal();
    
            } finally {
                lock.unlock();
            }
    
        }
    
    }

## 定义一个容器的生产者消费者 ##

package com.pzq.thread;
    
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    public class BoundBuffer {
        public static void main(String[] args) {
    
            Resources resources = new Resources();
    
            Products products = new Products(resources);
            Customers customers = new Customers(resources);
    
            new Thread(products).start();
            new Thread(products).start();
            new Thread(products).start();
            new Thread(customers).start();
            new Thread(customers).start();
            new Thread(customers).start();
    
        }
    
    }
    
    class Customers implements Runnable {
        Resources rs;
    
        public Customers(Resources rs) {
            this.rs = rs;
        }
    
        @Override
        public void run() {
            while (true) {
                try {
                    rs.take();
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
    }
    
    class Products implements Runnable {
        Resources rs;
    
        int count = 1;
    
        public Products(Resources rs) {
            this.rs = rs;
        }
    
        @Override
        public void run() {
            while (true) {
                try {
                    rs.put("烤鸭");
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
    }
    
    class Resources {
        Lock lock = new ReentrantLock();
        Condition notEmpty = lock.newCondition();
        Condition notFull = lock.newCondition();
        Object[] buffer = new Object[100];
        int putptr, takeptr, count;
    
        public void put(Object x) throws InterruptedException {
            lock.lock();
            try {
                while (count == buffer.length) {
                    notFull.await();
                }
                buffer[putptr] = x;
                if (++putptr == buffer.length) {
                    putptr = 0;
                }
                ++count;
                System.out.println("-------" + Thread.currentThread().getName() + "生产了---" + (x.toString() + putptr));
                notEmpty.signal();
    
            } finally {
                lock.unlock();
            }
        }
    
        public void take() throws InterruptedException {
            lock.lock();
            // Object x = null;
            try {
                while (count == 0) {
                    notEmpty.await();
                }
                Object x = buffer[takeptr];
                if (++takeptr == buffer.length) {
                    takeptr = 0;
                }
                --count;
                System.out.println("#######" + Thread.currentThread().getName() + "消费了####" + (x.toString() + takeptr));
    
                notFull.signal();
    
            } finally {
                lock.unlock();
            }
            // return x;
        }
    }

# wait和sleep区别 #

1、wait可以指定时间也可以不指定。

sleep必须指定时间

2、在同步中时，对CPU的执行权和锁的处理不同

wait：释放执行权，释放锁

sleep:释放执行权，不释放锁

# 停止线程 #

1、run方法结束：线程中的任务结束了，线程就停止了。

2、可以使用interrupt()方法将线程从冻结状态强制回鹘到运行状态中，让线程具备cpu的执行资格

当强制动作发生了InterruputdException,设置标记。

package com.pzq.thread;
    
    public class SyncDemo5 {
    
        public static void main(String[] args) {
            StopRunnable stop = new StopRunnable();
    
            Thread thread1 = new Thread(stop);
            Thread thread2 = new Thread(stop);
    
            thread1.start();
            thread2.start();
            thread1.interrupt();
            thread2.interrupt();
        }
    
    }
    
    class StopRunnable implements Runnable {
    
        boolean flag = true;
    
        @Override
        public synchronized void run() {
            while (flag) {
                try {
                    wait();
                } catch (InterruptedException e) {
                    System.out.println(Thread.currentThread().getName() + " , " + e);
                    setFlagFalse();
                }
                System.out.println(Thread.currentThread().getName() + " ..... ");
            }
        }
    
        public void setFlagFalse() {
            flag = false;
        }
    
    }

# 线程中常见方法 #

setDaemon: 将该线程标记为守护线程（后台线程）或用户线程，当正在运行的线程都是守护线程时java虚拟机退出。该方法必须在启动线程前调用。

该方法首先调用线程的 checkAccess方法，且不带任何参数。这可能抛出SecurityException（当前线程中）

join  调用join时说明线程要申请加入进来运行。主线程会将执行权释放出来。等待thread1执行完。只要thread1执行完主线程就可以执行。

//临时加入一个线程运算时可以用到join方法

package com.pzq.thread;
    
    public class JoinDemo {
    
        public static void main(String[] args) {
            TestRun testRun = new TestRun();
            Thread thread1 = new Thread(testRun);
            Thread thread2 = new Thread(testRun);
    
            thread1.start();
            try {
                thread1.join(); //thread1线程要申请加入进来运行。主线程会将执行权释放出来。等待thread1执行完。只要thread1执行完主线程就可以执行。
                //临时加入一个线程运算时可以用到join方法
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            thread2.start();
    
            for (int x = 0; x < 50; x++) {
                System.out.println(Thread.currentThread().getName() + " ... " + x);
            }
    
        }
    }
    
    class TestRun implements Runnable {
    
        @Override
        public void run() {
            for (int i = 0; i < 50; i++) {
                System.out.println(Thread.currentThread().getName() + " ... " + i);
            }
        }
    
    }

setPriority 设置线程优先级

yield:释放自己的执行权

转载于:https://www.cnblogs.com/qiangge-python/p/11067391.html