生产者与消费者基本程序模型
-
在多线程开发过程之中最为著名的案例就是生产者与消费者操作，该操作的主要流程如下：
- 生产者负责信息内容的生产；
- 每当生产者生产完成一项完整的信息之后，消费者要从这里面取走信息；
- 如果生产者没有生产完则消费者要等待他生产完成，如果消费者还没有对信息进行消费，则生产者应该等待消费处理完成后再继续生产；
可以将生产者与消费者定义为两个独立的线程类对象，既然生产者与消费者是两个独立的线程，那么这两个独立的线程之间就需要有一个数据保存集中点，那么可以单独定义一个Message类实现数据的保存；
此图来源于李兴华老师
package com.company;
class Producer implements Runnable {
private Message msg;
public Producer(Message msg) {
this.msg = msg;
}
@Override
public void run() {
for (int x = 0; x < 100; x++) {
if (x % 2 == 0) {
this.msg.setTitle("张三");
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
this.msg.setContent("帅哥");
} else {
this.msg.setTitle("李四");
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
this.msg.setContent("丑八怪");
}
}
}
}
class Consumer implements Runnable {
private Message msg;
public Consumer(Message msg) {
this.msg = msg;
}
@Override
public void run() {
for (int x = 0; x < 100; x++) {
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(this.msg.getTitle() + " - " + this.msg.getContent());
}
}
}
class Message {
private String title;
private String content;
public void setTitle(String title) {
this.title = title;
}
public void setContent(String content) {
this.content = content;
}
public String getTitle() {
return title;
}
public String getContent() {
return content;
}
}
public class ThreadDemo {
public static void main(String[] args) throws Exception {
Message msg = new Message();
new Thread(new Producer(msg)).start();
new Thread(new Consumer(msg)).start();
}
}
- 通过上面程序的执行会发现有两个主要的问题：
- 问题一：数据不同步；
- 问题二：生产一个取走一个，但是发现了重复生产重复取出问题；
解决数据同步
- 要想解决数据同步问题，最简单的做法是使用synchronized关键字定义同步代码块或同步方法；
package com.company;
class Producer implements Runnable {
private Message msg;
public Producer(Message msg) {
this.msg = msg;
}
@Override
public void run() {
for (int x = 0; x < 100; x++) {
if (x % 2 == 0) {
this.msg.set("张三", "帅哥");
} else {
this.msg.set("李四", "丑八怪");
}
}
}
}
class Consumer implements Runnable {
private Message msg;
public Consumer(Message msg) {
this.msg = msg;
}
@Override
public void run() {
for (int x = 0; x < 100; x++) {
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(this.msg.get());
}
}
}
class Message {
private String title;
private String content;
public synchronized void set(String title, String content) {
this.title = title;
this.content = content;
}
public synchronized String get() {
return this.title + " - " + this.content;
}
}
public class ThreadDemo {
public static void main(String[] args) throws Exception {
Message msg = new Message();
new Thread(new Producer(msg)).start();
new Thread(new Consumer(msg)).start();
}
}
- 这个时候发现数据已经可以正常的保持一致了，但是对于重复操作的问题依然存在；
线程等待与唤醒
- 要想解决生产者与消费者的问题，那么最好的解决方案就是使用等待与唤醒机制，而对于等待与唤醒的操作机制主要依靠的是Object类中提供的方法处理的：
- 一直等待：public final void wait() throws InterruptedException
- 设置等待时间：public final void wait(long timeout) throws InterruptedException
- 设置等待时间：public final void wait(long timeout, int nanos) throws InterruptedException
- 唤醒第一个等待线程：public final void notify()
- 唤醒所有等待线程：public final void notifyAll()
- 如果有若干个等待线程的话，那么notify()表示唤醒第一个等待的，而其他的线程继续等待，而notifyAll()表示唤醒所有等待的线程，哪个线程的优先级高就有可能先执行；
package com.company;
class Producer implements Runnable {
private Message msg;
public Producer(Message msg) {
this.msg = msg;
}
@Override
public void run() {
for (int x = 0; x < 100; x++) {
if (x % 2 == 0) {
this.msg.set("张三", "帅哥");
} else {
this.msg.set("李四", "丑八怪");
}
}
}
}
class Consumer implements Runnable {
private Message msg;
public Consumer(Message msg) {
this.msg = msg;
}
@Override
public void run() {
for (int x = 0; x < 100; x++) {
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(this.msg.get());
}
}
}
class Message {
private String title;
private String content;
private boolean flag = true;
public synchronized void set(String title, String content) {
if (!this.flag) {
try {
super.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
this.title = title;
this.content = content;
this.flag = false;
super.notify();
}
public synchronized String get() {
if (this.flag) {
try {
super.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
try {
return this.title + " - " + this.content;
} finally {
this.flag = true;
super.notify();
}
}
}
public class ThreadDemo {
public static void main(String[] args) throws Exception {
Message msg = new Message();
new Thread(new Producer(msg)).start();
new Thread(new Consumer(msg)).start();
}
}
- 这种处理形式是多线程开发之中最原始的处理方案，整个的同步、等待、唤醒机制都由开发者自行通过原生代码实现控制；
