java多线程三之线程协作与通信实例

多线程的难点主要就是多线程通信协作这一块了,前面笔记二中提到了常见的同步方法,这里主要是进行实例学习了,今天总结了一下3个实例:

1、银行存款与提款多线程实现,使用Lock锁和条件Condition。     附加 : 用监视器进行线程间通信

2、生产者消费者实现,使用LinkedList自写缓冲区。

3、多线程之阻塞队列学习,用阻塞队列快速实现生产者消费者模型。    附加:用布尔变量关闭线程

       在三种线程同步方法中,我们这里的实例用Lock锁来实现变量同步,因为它比较灵活直观。

       实现了变量的同步,我们还要让多个线程之间进行“通话”,就是一个线程完成了某个条件之后,告诉其他线程我完成了这个条件,你们可以行动了。下面就是java提供的条件接口Condition定义的同步方法:

       很方便的是,java的Lock锁里面提供了newConditon()方法可以,该方法返回:一个绑定了lock锁的Condition实例,有点抽象,其实把它看作一个可以发信息的锁就可以了,看后面的代码,应该就能理解了。

1、银行存款与提款多线程实现。

我们模拟ATM机器存款与提款,创建一个账户类Account(),该类包含同步方法:

     存款方法:deposit()

     提款方法:withdraw()

     以及一个普通的查询余额的方法getbalance().

我们创建两个任务线程,分别调用两个同步方法,进行模拟操作,看代码:

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class ThreadCooperation {
	private static Account account = new Account();
	public static void main(String[] args)
	{
		//创建线程池
		ExecutorService executor = Executors.newFixedThreadPool(2);
		executor.execute(new DepositTask());
		executor.execute(new WithdrawTask());
	}
	//存钱
	public static class DepositTask implements Runnable
	{
		@Override
		public void run() {
			try {
				while(true)
				{
					account.deposit((int)(Math.random()*1000)+1);
					Thread.sleep(1000);
				}
				} catch (InterruptedException e) {
					e.printStackTrace();
			}
		}
	}
	public static class WithdrawTask implements Runnable
	{
		@Override
		public void run() {
			try{
				while(true)
				{
					account.withdraw((int)(Math.random()*1000)+1);
					Thread.sleep(500);
				}
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	public static class Account
	{
		//一个锁是一个Lock接口的实例   它定义了加锁和释放锁的方法     ReentrantLock是为创建相互排斥的锁的Lock的具体实现
		private static Lock lock = new ReentrantLock();
		//创建一个condition,具有发通知功能的锁,前提是要实现了lock接口
		private static Condition  newDeposit = lock.newCondition();
		private int balance = 0;
		public int getBalance()
		{
			return balance;
		}
		public void withdraw(int amount)
		{
			lock.lock();
			try {
				while(balance < amount)
				{
				System.out.println("\t\t钱不够,等待存钱");
					newDeposit.await();
				}
				balance -= amount;
				System.out.println("\t\t取出"+amount+"块钱\t剩余"+getBalance());
			} catch (InterruptedException e) {
				e.printStackTrace();
			}finally{
				lock.unlock();
			}
		}

		public void deposit(int amount)
		{
			lock.lock();
			try{
				balance+=amount;
				System.out.println("存入"+amount+"块钱");
				newDeposit.signalAll();   //发信息唤醒所有的线程
			}
			finally{
				lock.unlock();
			}
		}
	}
}

运行截图

分析:

      1、程序中需要注意的:创建一个condition,具有发通知功能的锁,前提是要实现了lock接口。

      2、while(balance < amount)不能改用if判断,用if会使得线程不安全,使用if会不会进行循环验证,而while会,我们经常看到while(true),但是不会经常看到if(true).

      3、调用了await方法后,要记得使用signalAll()或者signal()将线程唤醒,否则线程永久等待。

       最后再来分析一下这个类的结构,有3个类,两个静态任务类实现了Runnable接口,是线程类,而另外一个类则是普通的任务类,包含了线程类所用到的方法。我们的主类在main方法前面就实例化一个Account类,以供线程类调用该类里面的同步方法。

      这种构造方式是多线程常用到的一种构造方式吧。不难发现后面要手写的生产者消费者模型也是这样子构造的。这相当于是一个多线程模板。也是我们学习这个例子最重要的收获吧。

 

 

用监视器进行线程之间的通信

       还有一点,接口Lock与Condition都是在java5之后出现的,在这之前,线程通信是通过内置的监视器(monitor)实现的。

      监视器是一个相互排斥且具有同步能力的对象,任意对象都有可能成为一个monitor。监视器是通过synchronized关键字来对自己加锁(加锁解锁是解决线程同步最基本的思想),使用wait()方法时线程暂停并 等待条件发生,发通知则是通过notify()和notifyAll()方法。大体的模板是这样子的:

不难看出await()、signal()、signally()是wait()、notify()、notifyAll()的进化形态,所以不建议使用监视器。

 

2、生产者消费者实现,使用LinkedList自写缓冲区

         这个模型一直很经典,学操作系统的时候还学过,记得linux还用PV操作去实现它,不过这东西是跨学科的。

考虑缓存区buffer的使用者,生产者和消费者,他们都能识别缓冲区是否满的,且两种各只能发出一种信号:

       生产者:它能发出notEmpty()信号,即缓冲区非空信号,当它看到缓冲区满的时候,它就调用await等待。

       消费者:它能发出notFull()信号,即缓冲区未满的信号,当它看到缓冲区空的时候,它也调用await等待。

看代码:

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

//生产者消费者
public class ConsumerProducer {
	private static Buffer  buffer= new Buffer();
	public static void main(String[] args)
	{
		ExecutorService executor = Executors.newFixedThreadPool(2);
		executor.execute(new ProducerTask());
		executor.execute(new ConsumerTask());
		executor.shutdown();
	}

	public static class ProducerTask implements Runnable
	{
		@Override
		public void run() {
			int i=1;
			try {
				while(true)
				{
					System.out.println("生产者写入数据"+i);
					buffer.write(i++);
					Thread.sleep((int)(Math.random()*80));

				}
			}catch (InterruptedException e) {
					e.printStackTrace();
			}
		}
	}

	public static class ConsumerTask implements Runnable
	{
		public void run() {
			try {
				while(true){
					System.out.println("\t\t消费读出数据"+buffer.read());
					Thread.sleep((int)(Math.random()*100));
					}
				} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	public static class Buffer
	{
		private static final int CAPACTIY = 4;  //缓冲区容量
		private java.util.LinkedList<Integer> queue = new java.util.LinkedList<Integer>();

		private static Lock lock = new ReentrantLock();

		private static Condition  notEmpty = lock.newCondition();
		private static Condition notFull = lock.newCondition();

		public void write(int value)
		{
			lock.lock();
			try{
				while(queue.size()==CAPACTIY)
				{
					System.out.println("缓冲区爆满");
					notFull.await();
				}
				queue.offer(value);
				notEmpty.signalAll();  //通知所有的缓冲区未空的情况
			}catch(InterruptedException ex){
				ex.printStackTrace();
			}finally{
				lock.unlock();
			}
		}

		@SuppressWarnings("finally")
		public int read()
		{
			int value = 0;
			lock.lock();
			try{
				while(queue.isEmpty())
				{
					System.out.println("\t\t缓冲区是空的,等待缓冲区非空的情况");
					notEmpty.await();
				}
				value = queue.remove();
				notFull.signal();
			}catch(InterruptedException ex){
				ex.printStackTrace();
			}finally{
				lock.unlock();
				return value;
			}
		}
	}
}

 

运行截图

程序运行正常,不过稍微延长一下读取时间,就会出现这样的情况

     

        程序里面设置的容量是4,可是这里却可以存入最多5个数据,而且更合理的情况应该是初始缓冲区是空的,后面找了下这个小bug,原来是调用offer()函数应该放在检测语句之前,如果希望一开始就调用ConsumerTask,在main方法里面调换两者的顺序即可。

3、用阻塞队列快速实现生产者消费者模型

        java的强大之处是它有着丰富的类库,我们学习java在某种程度上就是学习这些类库。

 

        阻塞队列是这样的一种队列:当试图向一个满队列里添加元素  或者 从空队列里删除元素时,队列会让线程自动阻塞,且当队列满时,队列会继续存储元素,供唤醒后的线程使用。这应该说是专门为消费者生产者模型而设计的一种队列吧,它实现了Queue接口,主要方法是put()和take()方法。

                                                                        

java支持三个具体的阻塞队列ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue。都在java.util.concurrent包中。

      

简单描述上面三个阻塞队列:

          ArrayBlockingQueue: 该阻塞用数组实现,按照FIFO,即先进先出的原则对数据进行排序,和数组的使用有点相似,它事先需要指定一个容量,不过即便队列超出这个容量,也是不会报错滴。

          LinkeddBlockingQueue:用链表实现,默认队列大小是Integer.MAX_VALUE,也是按照先进先出的方法对数据排序,性能可能比ArrayBlockingQueue,有待研究。

          PriorityBlockingQueue:用优先队列实现的阻塞队列,会对元素按照大小进行排序,也可以创建不受限制的队列,put方法永不阻塞。

 

ok,看代码:

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ConsumerProducerUsingBlockQueue {
	private static ArrayBlockingQueue<Integer>  buffer = new ArrayBlockingQueue<Integer>(2);
	public static void main(String[] args)
	{
		ExecutorService executor = Executors.newFixedThreadPool(2);
		executor.execute(new Consumer());
		executor.execute(new Producer());

		try {
			Thread.sleep(100);
			executor.shutdownNow();     //暴力关闭,会报错,不推荐
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

	public static class Consumer implements Runnable
	{
		@Override
		public void run() {
			try{
				int i=1;
				while(true){
					System.out.println("生成者写入:"+i);
					buffer.put(i++);
					Thread.sleep((int)(Math.random())*1000);
				}
			}catch(InterruptedException ex){
				ex.printStackTrace();
			}
		}
	}
	public static class Producer implements Runnable
	{
		@Override
		public void run() {
			try{
				while(true)
				{
					System.out.println("\t\t消费者取出"+buffer.take());
					Thread.sleep((int)(Math.random())*10000);
				}
			}catch(InterruptedException ex){
				ex.printStackTrace();
			}
		}
	}
}

 

运行截图:

        没啥大的问题,就是在关闭线程的时候太过暴力了,会报错,线程里面的每一个函数都似乎值得研究,之前想通过Interrupt暂停,不过失败了,就直接使用线程池执行器的shoutdownNow方法来的。后面自己又用了另外一种关闭线程的方法,见下面代码

 

使用LinkedBlockingQueue实现消费者生产者且使用布尔变量控制线程关闭

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
public class A_Control_stop {
	private static LinkedBlockingQueue<String>  buffer = new LinkedBlockingQueue<String>();

	public static void main(String[] args)
	{
		ExecutorService executor = Executors.newFixedThreadPool(2);
		executor.execute(new Consumer());
		executor.execute(new Producer());
		executor.shutdown();

		while(!executor.isTerminated()){}
		System.out.println("所有的的线程都正常结束");
	}

	public static class Consumer implements Runnable
	{
		private volatile boolean exit = false;
		@Override
		public void run() {
			try{
				int i=0;
				String[] str ={"as","d","sd","ew","sdfg","esfr"};
				while(!exit){
					System.out.println("生成者写入:"+str[i]);
					buffer.put(str[i++]);
					Thread.sleep((int)(Math.random())*10);
					if(5==i)
					{
						exit=true;
					}
				}
			}catch(InterruptedException ex){
				ex.printStackTrace();
			}
		}
	}
	public static class Producer implements Runnable
	{
		private volatile boolean exit = false;
		@Override
		public void run() {
			try{
				int i=0;
				while(!exit)
				{
					System.out.println("\t\t消费者取出"+buffer.take());
					i++;
					Thread.sleep((int)(Math.random())*10);
					if(5==i)
					{
						exit=true;
					}
				}
			}catch(InterruptedException ex){
				ex.printStackTrace();
			}
		}
	}
}

 

截图

         关于阻塞队列,觉得这篇文章讲的不错,推荐大家看看  聊聊并发----Java中的阻塞队列

        用了几天,多线程算是学了点皮毛,附注一下:这几天文章主要是参考了《java程序语言设计进阶篇第8版》,老外写的书讲的真心不错,只不过现在java都已经更新到java8了。在其他一些网站上看到自己的文章,没有说明转载什么的,估计是直接“被采集”过去了。

时间: 2024-10-16 05:07:35

java多线程三之线程协作与通信实例的相关文章

Java学习笔记46(多线程三:线程之间的通信)

多个线程在处理同一个资源,但是线程的任务却不相同,通过一定的手段使各个线程能有效地利用资源, 这种手段即:等待唤醒机制,又称作线程之间的通信 涉及到的方法:wait(),notify() 示例: 两个线程一个输入,一个输出 package demo; public class Resource { public String name; public String sex; } 输入线程: package demo; public class Input implements Runnable

Java多线程——&lt;三&gt;让线程有返回值

一.概述 到目前为止,我们已经能够声明并使一个线程任务运行起来了.但是遇到一个问题:现在定义的任务都没有任何返回值,那么加入我们希望一个任务运行结束后告诉我一个结果,该结果表名任务执行成功或失败,此时该怎么办呢? 答案是使用Callable.之前定义的任务都直接实现了Runnable,该接口的run方法并无返回值.而Callable的call方法可以根据你传入的泛型参数返回对应类型的数据. 二.实现 1.实现Callable接口,定义可返回结果的线程任务 public class TaskCal

Java多线程(三)

本文主要接着前面多线程的两篇文章总结Java多线程中的线程安全问题. 一.一个典型的Java线程安全例子 1 public class ThreadTest { 2 3 public static void main(String[] args) { 4 Account account = new Account("123456", 1000); 5 DrawMoneyRunnable drawMoneyRunnable = new DrawMoneyRunnable(account,

VC中利用多线程技术实现线程之间的通信

文章来源:[url]http://www.programfan.com/article/showarticle.asp?id=2951[/url] 当前流行的Windows操作系统能同时运行几个程序(独立运行的程序又称之为进程),对于同一个程序,它又可以分成若干个独立的执行流,我们称之为线程,线程提供了多任务处理的能力.用进程和线程的观点来研究软件是当今普遍采用的方法,进程和线程的概念的出现,对提高软件的并行性有着重要的意义.现在的大型应用软件无一不是多线程多任务处理,单线程的软件是不可想象的.

Java多线程之后台线程不执行finally

后台线程不执行finally package wzh.daemon; import java.util.concurrent.TimeUnit; class ADaemon implements Runnable { @Override public void run() { try { System.out.println("Starting ADaemon"); TimeUnit.SECONDS.sleep(1); } catch (Exception e) { System.ou

Java多线程之后台线程

将线程设置成后台线程Daemons 主线程结果后,后台线程将自动结果. package wzh.test; import java.util.concurrent.TimeUnit; class SimpleDaemons implements Runnable{ @Override public void run() { try { while (true) { TimeUnit.MILLISECONDS.sleep(100); System.out.println(Thread.curren

java多线程二之线程同步的三种方法

java多线程的难点是在:处理多个线程同步与并发运行时线程间的通信问题.java在处理线程同步时,常用方法有: 1.synchronized关键字. 2.Lock显示加锁. 3.信号量Semaphore. 线程同步问题引入: 创建一个银行账户Account类,在创建并启动100个线程往同一个Account类实例里面添加一块钱.在没有使用上面三种方法的情况下: 代码: 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

java多线程同步以及线程间通信详解&amp;消费者生产者模式&amp;死锁&amp;Thread.join()(多线程编程之二)

本篇我们将讨论以下知识点: 1.线程同步问题的产生 什么是线程同步问题,我们先来看一段卖票系统的代码,然后再分析这个问题: [java] view plain copy print? package com.zejian.test; /** * @author zejian * @time 2016年3月12日 下午2:55:42 * @decrition 模拟卖票线程 */ public class Ticket implements Runnable { //当前拥有的票数 private 

Java多线程——&lt;三&gt;简单的线程执行:Executor

一.概述 按照<Java多线程——<一><二>>中所讲,我们要使用线程,目前都是显示的声明Thread,并调用其start()方法.多线程并行,明显我们需要声明多个线程然后都调用他的start方法,这么一看,似乎有些问题:第一.线程一旦多了,声明势必是个问题:第二.多线程启动如果通过手动执行的话,那可能一个线程已经跑完了,另外一个还没起来(我推测可能会出现这个问题).所以,我们在想,如果有个管家,能够帮我们管理这么多线程,只需要把我们定义的任务交给管家,管家就能够帮我们