综合实战：“生产者-消费者模型”
在多线程开发中最著名的案例就是生产者-消费者模型，该操作的主要流程如下：
- 生产者负责信息内容的生产
- 每当生产者完成一项完整的信息之后消费者要从这里面取走信息
- 如果生产者没有生产完没有生产完成则消费者要等待它生产完成，如果消费者还没有对信息进行消费，则生产者应等待消费者完成信息消费后在进行生产。
生产者与消费者基本程序模型
可将生产者和消费者定义为两个独立的线程类对象，但是对于现在生产的数据可以使用如下组成：
- 数据一：title = lyz、content=无敌；
- 数据二：title = zky、content=菜的；
既然消费者与生产者是两个独立的线程，那么这两个独立的线程之间就需要有一个数据的保存集中点，那么可以定义一个Message类进行保存。
基本模型
程序基本结构
class Producer implements Runnable{
private Message msg;
public Producer(Message msg) {
this.msg = msg;
}
public void run() {
for(int x =0; x<100; x++) {
if(x%2 == 0) {
this.msg.setTitle("lyz");
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
this.msg.setContent("无敌");
} else {
this.msg.setTitle("zky");
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
this.msg.setContent("菜鸡");
}
}
}
}
class Consumers implements Runnable{
private Message msg;
public Consumers(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 String getTitle() {
return title;
}
public void setTitle(String title) {
this.title = title;
}
public String getContent() {
return content;
}
public void setContent(String content) {
this.content = content;
}
}
public class Consumer {
public static void main(String[] args) {
Message msg = new Message();
new Thread(new Producer(msg)).start(); //启动生产者线程
new Thread(new Consumers(msg)).start(); //启动消费者线程
}
}
通过代码执行发现此时有两个主要问题：
- 数据不同步;
- 生产一个取走一个，但是发现有了重复生产和重复取出问题;
解决生产者-消费者同步问题
如果要解决问题，首先解决的就是数据同步处理问题，如果要想解决数据同步，最简单的做法时使用synchronized关键字定义同步代码块或同步方法，于是此时对于同步的处理就可以直接在Message类中完成。
解决同步操作
package per.lyz.thread_study;
class Producer implements Runnable{
private Message msg;
public Producer(Message msg) {
this.msg = msg;
}
public void run() {
for(int x =0; x<100; x++) {
if(x%2 == 0) {
this.msg.set("lyz","无敌");
} else {
this.msg.set("zky","菜鸡");
}
}
}
}
class Consumers implements Runnable{
private Message msg;
public Consumers(Message msg) {
this.msg = msg;
}
@Override
public void run() {
for (int x=0;x<100;x++) {
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;
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
this.content = content;
}
public synchronized String get() {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
return this.title +" - " + this.content;
}
}
public class Consumer {
public static void main(String[] args) {
Message msg = new Message();
new Thread(new Producer(msg)).start(); //启动生产者线程
new Thread(new Consumers(msg)).start(); //启动消费者线程
}
}
在进行同步处理的时候肯定要有一个同步处理的对象，那么此时肯定要将同步操作交由Message类处理时最合适的。此时数据正常保持一致，但是发现有了重复生产和重复问题依然存在。
利用Object类解决重复问题
要想解决消费者生产者问题，最好的解决方案是使用等待和唤醒机制。而对于等待和唤醒机制，主要依靠的是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();（优先级高的线程有可能被优先执行）
对于当前问题应该通过Message类完成处理。
解决重复问题
class Producer implements Runnable{
private Message msg;
public Producer(Message msg) {
this.msg = msg;
}
public void run() {
for(int x =0; x<100; x++) {
if(x%2 == 0) {
this.msg.set("lyz","无敌");
} else {
this.msg.set("zky","菜鸡");
}
}
}
}
class Consumers implements Runnable{
private Message msg;
public Consumers(Message msg) {
this.msg = msg;
}
@Override
public void run() {
for (int x=0;x<100;x++) {
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;
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
this.content = content;
this.flag = false; //已经生产完成
super.notify(); //唤醒可能的等到线程
}
public synchronized String get() {
if(this.flag) {
//还未生产，需要等待
try {
super.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
return this.title +" - " + this.content;
}finally {
//不管如何都要执行
this.flag = true;
super.notify(); //唤醒可能的等待线程
}
}
}
public class Consumer {
public static void main(String[] args) {
Message msg = new Message();
new Thread(new Producer(msg)).start(); //启动生产者线程
new Thread(new Consumers(msg)).start(); //启动消费者线程
}
}
这种处理形式就是在进行多线程开发中最原始的处理方案，整个的等待、同步、唤醒机制都有开发者自行通过原生代码实现控制。