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

前言

前面的例子都是多个线程在做相同的操作,比如4个线程都对共享数据做tickets–操作。大多情况下,程序中需要不同的线程做不同的事,比如一个线程对共享变量做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接口及实现类。

以上就是本文关于Java多线程之线程通信生产者消费者模式及等待唤醒机制代码详解的全部内容,希望对大家有所帮助。感兴趣的朋友可以继续参阅本站:Java编程之多线程死锁与线程间通信简单实现代码、Java多线程编程小实例模拟停车场系统等,如有不足之处,欢迎留言指出。感谢朋友们对本站的支持!

时间: 2017-10-25

详解java中的互斥锁信号量和多线程等待机制

互斥锁和信号量都是操作系统中为并发编程设计基本概念,互斥锁和信号量的概念上的不同在于,对于同一个资源,互斥锁只有0和1 的概念,而信号量不止于此.也就是说,信号量可以使资源同时被多个线程访问,而互斥锁同时只能被一个线程访问 互斥锁在java中的实现就是 ReetranLock , 在访问一个同步资源时,它的对象需要通过方法 tryLock() 获得这个锁,如果失败,返回 false,成功返回true.根据返回的信息来判断是否要访问这个被同步的资源.看下面的例子 public class Reen

Java多线程ForkJoinPool实例详解

引言 java 7提供了另外一个很有用的线程池框架,Fork/Join框架 理论 Fork/Join框架主要有以下两个类组成. * ForkJoinPool 这个类实现了ExecutorService接口和工作窃取算法(Work-Stealing Algorithm).它管理工作者线程,并提供任务的状态信息,以及任务的执行信息 * ForkJoinTask 这个类是一个将在ForkJoinPool执行的任务的基类. Fork/Join框架提供了在一个任务里执行fork()和join()操作的机制

Java多线程之readwritelock读写分离的实现代码

在多线程开发中,经常会出现一种情况,我们希望读写分离.就是对于读取这个动作来说,可以同时有多个线程同时去读取这个资源,但是对于写这个动作来说,只能同时有一个线程来操作,而且同时,当有一个写线程在操作这个资源的时候,其他的读线程是不能来操作这个资源的,这样就极大的发挥了多线程的特点,能很好的将多线程的能力发挥出来. 在Java中,ReadWriteLock这个接口就为我们实现了这个需求,通过他的实现类ReentrantReadWriteLock我们可以很简单的来实现刚才的效果,下面我们使用一个例子

浅谈Java多线程处理中Future的妙用(附源码)

java 中Future是一个未来对象,里面保存这线程处理结果,它像一个提货凭证,拿着它你可以随时去提取结果.在两种情况下,离开Future几乎很难办.一种情况是拆分订单,比如你的应用收到一个批量订单,此时如果要求最快的处理订单,那么需要并发处理,并发的结果如果收集,这个问题如果自己去编程将非常繁琐,此时可以使用CompletionService解决这个问题.CompletionService将Future收集到一个队列里,可以按结果处理完成的先后顺序进队.另外一种情况是,如果你需要并发去查询一

Java利用future及时获取多线程运行结果

Future接口是Java标准API的一部分,在java.util.concurrent包中.Future接口是Java线程Future模式的实现,可以来进行异步计算. 有了Future就可以进行三段式的编程了,1.启动多线程任务2.处理其他事3.收集多线程任务结果.从而实现了非阻塞的任务调用.在途中遇到一个问题,那就是虽然能异步获取结果,但是Future的结果需要通过isdone来判断是否有结果,或者使用get()函数来阻塞式获取执行结果.这样就不能实时跟踪其他线程的结果状态了,所以直接使用g

java 利用反射机制,获取实体所有属性和方法,并对属性赋值

一个普通的实体Person: private int id; private String name; private Date createdTime; ... //其它字段 // get set方法 ............... 现在需要把通过webService传过来的实体Person里面的所有字段的null值,换成"" 实现思路: 1.获取实体的所有字段,遍历 2.获取字段类型 3.调用字段的get方法,判断字段值是否为空 4.如果字段值为空,调用字段的set方法,为字段赋值

Java利用Request请求获取IP地址的方法详解

前言 最近在项目中遇到一个需求,是需要将不同省份的用户,展示不同内容,通过查找相关的资料,发现可以通过Request请求获取IP地址,下面我们先来贴代码, 如果你要在生产环境使用就直接拿去用吧,我这边已经上线了. 示例代码 public class IpAdrressUtil { /** * 获取Ip地址 * @param request * @return */ private static String getIpAdrress(HttpServletRequest request) { S

Java利用反射获取object的属性和值代码示例

在看反射顺便做个笔记,目前知道的反射的Object都是要有对象的也就是实体Bean. referance:Java反射简易教程 import java.lang.reflect.Field; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; /** * 反射处理Bean,得到里面的属性值 * * @author liulinsen * */ publ

Java利用Request请求如何获取IP地址对应的省份、城市详解

前言 最近的一个项目中需要将不同省份的用户,展示不同内容,所以需要通过Request请求获取IP地址, 然后通过IP获取IP对应省份. 这里的操作步骤一共有步: 1. 通过Request获取IP 2. 通过IP获取对应省份.城市 3. 通过设置的省份和IP对应省份进行比对,展示内容 通过Request获取IP 可以参考我的另外一篇文章[Java 通过Request请求获取IP地址]下面是代码: public class IpAdrressUtil { /** * 获取Ip地址 * @param

java利用Calendar类打印日历

本文实例为大家分享了java利用Calendar类打印日历的具体代码,供大家参考,具体内容如下 import java.util.Calendar; import java.util.Date; import java.util.Scanner; public class Test { public static void main(String[] args) { System.out.println("请输入日期(如2018-12-07):"); Scanner sc=new Sc

比较java中Future与FutureTask之间的关系

Future与FutureTask都是用于获取线程执行的返回结果.下面我们就对两者之间的关系与使用进行一个大致的介绍与分析 一.Future与FutureTask介绍: Future位于java.util.concurrent包下,它是一个接口 public interface Future<V> { boolean cancel(boolean mayInterruptIfRunning); boolean isCancelled(); boolean isDone(); V get() t

详解Java利用同步块synchronized()保证并发安全

本文实例为大家分享了Java利用同步块synchronized()保证并发安全的具体代码,供大家参考,具体内容如下 package day10; /** * 同步块 * 有效地缩小同步范围 * 可以在保证并发安全的同时尽可能提高并发效率 * * 实例:模拟两个人同时进店买衣服,为提高效率 * 只在试衣服阶段进行同步排队过程,其他阶段无需排队. * @author kaixu * */ public class SyncDemo2 { public static void main(String[

Java实现基于NIO的多线程Web服务器实例

代码地址:https://github.com/iyuanyb/webserver 实现了 静态.动态资源获取: Cookie.Session.HTTP 长连接,及 Session 和 HTTP 长连接的定时清除: 类似 Spring MVC 的注解式编程,如 @RequestMapping @RequestParam 等,方法中可以根据参数名从前台获取数据,可以传递对象,也支持级联属性,如: // GET /page?pageSize=10&pageNum=1 HTTP/1.1 @Reques

Java利用Redis实现消息队列的示例代码

本文介绍了Java利用Redis实现消息队列的示例代码,分享给大家,具体如下: 应用场景 为什么要用redis? 二进制存储.java序列化传输.IO连接数高.连接频繁 一.序列化 这里编写了一个java序列化的工具,主要是将对象转化为byte数组,和根据byte数组反序列化成java对象; 主要是用到了ByteArrayOutputStream和ByteArrayInputStream; 注意:每个需要序列化的对象都要实现Serializable接口; 其代码如下: package Utils