美文网首页
Java多线程之线程通信生产者消费者模式及等待唤醒机制代码详解

Java多线程之线程通信生产者消费者模式及等待唤醒机制代码详解

作者: Lemonrel | 来源:发表于2019-10-12 18:09 被阅读0次

    前言
    大多情况下,程序中需要不同的线程做不同的事,比如一个线程对共享变量做tickets++操作,另一个线程对共享变量做tickets–操作,这就是生产者和消费者模式。
    正文
    一,生产者-消费者模式也是多线程
    生产者和消费者模式也是多线程的范例。所以其编程需要遵循多线程的规矩。
    首先,既然是多线程,就必然要使用同步。上回说到,synchronized关键字在修饰函数的时候,使用的是“this”锁,所以在同一个类中的函数被synchronized修饰后,使用的是同一把锁。线程调用这些函数时,不管调用的是tickets++操作函数,还是tickets–函数,都会先去判断是否加锁了,得到锁之后再去进行具体的操作。
    我们先用代码把程序中的资源,生产者,消费者表示出来。

    package com.jimmy.ThreadCommunication;
    class Resource{  // 资源类
      private String productName; // 资源名称
      private int count = 1;    // 资源编号
      public void produce(String name){  // 生产资源函数
        this.productName = name + count;
        count ++;  // 资源编号递增,用来模拟资源递增
        System.out.println(Thread.currentThread().getName()+"...生产者.."+this.productName);
      }
      public void consume() { // 消费资源函数
        System.out.println(Thread.currentThread().getName()+"...消费者.."+this.productName);    
      }
    }
    class Producer implements Runnable{ // 生产者类,用于开启生产者线程
      private Resource res;
      //生产者初始化就要分配资源
      public Producer(Resource res) {  
        this.res = res;
      }
      @Override
      public void run() {
        for (int i = 0; i < 10; i++) {     
          res.produce("bread");   // 循环生产10次
        }
      }
    }
    class Comsumer implements Runnable{  // 消费者类,用于开启消费者线程
      private Resource res;
      //同理,消费者一初始化也要分配资源
      public Comsumer(Resource res) {
        this.res = res;
      }
      @Override
      public void run() {
        for (int i = 0; i < 10; i++) {     
          res.consume(); // 循环消费10次
        }
      }
    }
    public class ProducerAndConsumer1 {
      public static void main(String[] args) {
        Resource resource = new Resource(); // 实例化资源
        Producer producer = new Producer(resource); // 实例化生产者和消费者类,它们取得同一个资源
        Comsumer comsumer = new Comsumer(resource);
        Thread threadProducer = new Thread(producer); // 创建1个生产者线程
        Thread threadComsumer = new Thread(comsumer); // 创建1个消费者线程
        threadProducer.start(); // 分别开启线程
        threadComsumer.start();
      }
    }
    

    架子搭好了,就来运行一下,当然会出现错误的结果,如下所示:

    Thread-0...生产者..bread1
    Thread-0...生产者..bread2
    Thread-0...生产者..bread3
    Thread-0...生产者..bread4
    Thread-0...生产者..bread5
    Thread-1...消费者..bread1
    Thread-1...消费者..bread6
    Thread-1...消费者..bread6
    Thread-1...消费者..bread6
    Thread-1...消费者..bread6
    Thread-1...消费者..bread6
    Thread-0...生产者..bread6
    Thread-0...生产者..bread7
    Thread-1...消费者..bread6
    Thread-1...消费者..bread8
    Thread-1...消费者..bread8
    Thread-1...消费者..bread8
    Thread-0...生产者..bread8
    Thread-0...生产者..bread9
    Thread-0...生产者..bread10
    

    很明显,出现了线程安全错误。这时,就需要“同步”来保证对共享变量的互斥访问。上面代码中需要同步的就是Resource资源类中的produce和consume方法,分别使用synchronized来修饰,由于synchronized修饰方法时使用的是“this”锁,所以同一个类中的所有被修饰的方法用的都是同一个锁,那么线程一次只能访问其中一个方法。加锁后的Resource类方法如下:

    class Resource{  // 资源类
      private String productName; // 资源名称
      private int count = 1;    // 资源编号
      public synchronized void produce(String name){  // 生产资源函数
        this.productName = name + count;
        count ++;  // 资源编号递增,用来模拟资源递增
        System.out.println(Thread.currentThread().getName()+"...生产者.."+this.productName);
      }
      public synchronized void consume() { // 消费资源函数
        System.out.println(Thread.currentThread().getName()+"...消费者.."+this.productName);    
      }
    }
    

    再来跑一次代码,又出现问题了:

    Thread-0...生产者..bread1
    Thread-0...生产者..bread2
    Thread-0...生产者..bread3
    Thread-0...生产者..bread4
    Thread-0...生产者..bread5
    Thread-0...生产者..bread6
    Thread-0...生产者..bread7
    Thread-0...生产者..bread8
    Thread-0...生产者..bread9
    Thread-0...生产者..bread10
    Thread-1...消费者..bread10
    Thread-1...消费者..bread10
    Thread-1...消费者..bread10
    Thread-1...消费者..bread10
    Thread-1...消费者..bread10
    Thread-1...消费者..bread10
    Thread-1...消费者..bread10
    Thread-1...消费者..bread10
    Thread-1...消费者..bread10
    Thread-1...消费者..bread10
    

    虽然没有了线程安全错误,但是问题来了,生产者不停的生产,还没等消费者消费呢,就将后面的资源覆盖了前面的资源,导致消费者消费不到前面的资源,这样很容易造成系统资源浪费。理想中的结果应该是,生产者生产一个,消费者消费一个,和谐运行。对此,java为多线程引入了”等待-唤醒”机制。
    二,等待唤醒机制
    与线程做同样的操作不同,不同线程之间的操作需要等待唤醒机制来保证线程间的执行顺序。生产者和消费者模式中,生产者和消费者是两类不同的线程, 这两类中又可以有很多线程来协同工作。通俗来说就是,系统为资源设置一个标志flag,该标志用来标明资源是否存在,所有的线程执行操作前都要判断资源是否存在。举例来说,系统初始化后,资源是空的。接下来要执行的可能是生产者线程,也可能是消费者线程。如果是消费者线程获得执行权,先判断资源,此时为空,就会进入阻塞状态,交出执行权,并唤醒其他线程。如果是生产者线程获得执行权,先判断资源,此时为空,立马进行生产,完了交出执行权并唤醒其他线程。
    注意,上面提到了两点,第一点是标志位flag,也就是等待机制,生产者要判断系统没有资源才进行生产,不然要等待,消费者要判断系统有资源才进行消费,不然也要等待。第二点是唤醒机制,不管是生产者还是消费者,它们在生产完或者消费完后,都要执行一个唤醒操作。java提供的等待唤醒机制是由java.lang.Object类中的wait()和notify()函数组来实现的。其中notify()函数随机唤醒一个被wait()的线程,而notifyAll()唤醒所有被wait()的线程。很遗憾,并没有直接唤醒对方线程的函数。
    notify()适用于单生产者和单消费者模式,而notifyAll()适用于多生产者或多消费者模式。
    下面来看2个生产者和2个消费者线程处理一个共享变量的代码示例:

    package com.jimmy.ThreadCommunication;
    class Resource2{
      private String productName;
      private int count = 1;
      private boolean flag = false; // 资源类增加一个标志位,默认false,也就是没有资源
      public synchronized void produce(String name){
        while (flag == true) { // 如果flag为true,也就是有资源了,生产者线程就去等待。
          try {
            wait(); // wait函数抛出的异常只能被截获
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
        }
        this.productName = name + count;
        count ++;
        System.out.println(Thread.currentThread().getName()+"....生产者.."+this.productName);
        flag = true; // 生产完了就将flag修改为true
        notifyAll(); // 然后唤醒其他线程
      }
      public synchronized void consume() {
        while (flag == false) { // 如果flag为false,也就是没有资源,消费者线程就去等待
          try {        // 判断flag要用while,因为线程被唤醒后会再次判断flag   
            wait();     // 而如果是if来判断,被唤醒后不会再判断flag,那么多个生产者线程就可能死锁
          } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
          }
        }
        System.out.println(Thread.currentThread().getName()+"...消费者.."+this.productName);    
        flag = false; // 消费完了就把标志改为false
        notifyAll();  // 然后唤醒其他线程,因为有多个生产者和消费者线程,所以要用notifyAll,
                // 因为notify只唤醒一个,唤醒到同类型的线程就不好了。
      }
    }
    class Producer2 implements Runnable{
      private Resource2 res;
      //生产者初始化就要分配资源
      public Producer2(Resource2 res) {
        this.res = res;
      }
      @Override
      public void run() {
        for (int i = 0; i < 5; i++) {
          res.produce("bread");
        }
      }
    }
    class Comsumer2 implements Runnable{
      private Resource2 res;
      //同理,消费者一初始化也要分配资源
      public Comsumer2(Resource2 res) {
        this.res = res;
      }
      @Override
      public void run() {
        for (int i = 0; i < 10; i++) {
          res.consume();
        }
      }
    }
    public class ProducerAndConsumer2 {
      public static void main(String[] args) {
        Resource2 resource = new Resource2(); // 实例化资源
        Producer2 producer = new Producer2(resource); // 实例化生产者,并传入资源对象
        Comsumer2 comsumer = new Comsumer2(resource); // 实例化消费者,并传入相同的资源对象
        Thread threadProducer1 = new Thread(producer); // 创建2个生产者线程
        Thread threadProducer2 = new Thread(producer);
        Thread threadComsumer1 = new Thread(comsumer); // 创建2个消费者线程
        Thread threadComsumer2 = new Thread(comsumer);
        threadProducer1.start();
        threadProducer2.start();
        threadComsumer1.start();
        threadComsumer2.start();
      }
    }
    

    上述代码的输出结果如下,是理想中的生产一个,消费一个依次进行。

    Thread-0....生产者..bread1
    Thread-3...消费者..bread1
    Thread-1....生产者..bread2
    Thread-2...消费者..bread2
    Thread-1....生产者..bread3
    Thread-3...消费者..bread3
    Thread-0....生产者..bread4
    Thread-3...消费者..bread4
    Thread-1....生产者..bread5
    Thread-2...消费者..bread5
    Thread-1....生产者..bread6
    Thread-3...消费者..bread6
    Thread-0....生产者..bread7
    Thread-3...消费者..bread7
    Thread-1....生产者..bread8
    Thread-2...消费者..bread8
    Thread-0....生产者..bread9
    Thread-3...消费者..bread9
    Thread-0....生产者..bread10
    Thread-2...消费者..bread10
    

    可以看出,线程0和1是生产者线程,他们每次只有一个进行生产。线程2和3是消费者线程,同样的,每次只有一个进行消费。
    注意,上述代码中的问题有2点需要注意,第一点是用if还是while来判断flag,第二点是用notify还是notifyAll函数。统一来说,while判断在线程唤醒后还会再次判断,如果只有一个生产者和消费者线程的话可以用if,如果有多个生产者或者消费者,就必须用while判断,不然会出现死锁。所以,最终要用while和notifyAll()的组合。
    总结
    多线程编程往往是多个线程执行不同的任务,不同的任务不仅需要“同步”,还需要“等待唤醒机制”。两者结合就可以实现多线程编程,其中的生产者消费者模式就是经典范例。
    然而,使用synchronized修饰同步函数和使用Object类中的wait,notify方法实现等待唤醒是有弊端的。就是效率问题,notifyAll方法唤醒所有被wait的线程,包括本类型的线程,如果本类型的线程被唤醒,还要再次判断并进入wait,这就产生了很大的效率问题。理想状态下,生产者线程要唤醒消费者线程,而消费者线程要唤醒生产者线程。为此,jdk1.5引入了java.util.concurrent.locks包,并提供了Lock和Condition接口及实现类。
    对标阿里P6级架构师
    扫描下方二维码,可以获取更多学习资料。

    相关文章

      网友评论

          本文标题:Java多线程之线程通信生产者消费者模式及等待唤醒机制代码详解

          本文链接:https://www.haomeiwen.com/subject/zghcmctx.html