生產者-消費者算是并發編程中常見的問題，依靠緩沖區我們可以實作生產者與消費者之間的解耦，生產者只管往緩沖區里面放東西，消費者只管往緩沖區里面拿東西，這樣我們避免生產者想要交付資料給消費者，但消費者此時還無法接受資料這樣的情況發生，
wait notify
這個問題其實就是執行緒間的通訊，所以要注意的是不能同時讀寫，生產者在緩沖區滿的時候不生產，等待；消費者在緩沖區為空的時候不消費，等待，比較經典的做法是wait和notify，
生產者執行緒執行15次set操作
public class Producer implements Runnable{
private Channel channel;
public Producer(Channel channel) {
this.channel = channel;
}
@Override
public void run() {
for(int i=0;i<15;i++){
channel.set(Thread.currentThread().getName()+" "+i);
}
}
}
消費者執行緒執行10次get操作
public class Consumer implements Runnable {
private Channel channel;
public Consumer(Channel channel) {
this.channel = channel;
}
@Override
public void run() {
for(int i=0;i<10;i++){
System.out.println("Consumer "+Thread.currentThread().getName()+" get "+channel.get());
}
}
}
現在定義Channel類,并創建兩個生產者執行緒和三個消費者執行緒
public class Channel {
private List<String> buffer=new ArrayList<>();
private final int MAX_SIZE=10;
public synchronized String get(){
while (buffer.size()==0){//不要用if，醒來了也要再次判斷
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
String str=buffer.remove(0);
notifyAll();
return str;
}
public synchronized void set(String str){
while (buffer.size()==MAX_SIZE){
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
buffer.add(str);
notifyAll();
}
public static void main(String[] args) {
Channel channel=new Channel();
Producer producer=new Producer(channel);
Consumer consumer=new Consumer(channel);
for(int i=0;i<2;i++){
new Thread(producer).start();
}
for (int i=0;i<3;i++){
new Thread(consumer).start();
}
}
}
使用notifyAll而不是notify的原因是，notify有可能出現多次喚醒同類的情況，造成“假死”，我們可以使用Condition來實作更精確的喚醒，
Condition
將上面代碼中的Channel類修改一下即可
public class Channel {
private List<String> buffer=new ArrayList<>();
private final int MAX_SIZE=10;
private Lock lock=new ReentrantLock();
private Condition producer=lock.newCondition();
private Condition consumer=lock.newCondition();
public String get(){
String str=null;
try {
lock.lock();
while (buffer.size()==0){
consumer.await();
}
str=buffer.remove(0);
producer.signalAll();
}catch (InterruptedException e) {
e.printStackTrace();
}finally {
lock.unlock();
}
return str;
}
public void set(String str){
try {
lock.lock();
while (buffer.size()==MAX_SIZE){
producer.await();
}
buffer.add(str);
consumer.signalAll();
}catch (InterruptedException e) {
e.printStackTrace();
}finally {
lock.unlock();
}
}
}
雙緩沖與Exchanger
當同步的花銷非常大時，我們可以采用雙緩沖區的辦法，雙緩沖的一個好處就在于：因為生產者和消費者各自擁有一個緩沖區，所以他們不會同時對同一個緩沖區進行操作，那么我們就不需要為讀寫操作加鎖，用空間換了時間，在Java中可以通過Exchanger來交換兩個執行緒之間的資料結構，
public class Producer implements Runnable{
private List<String> buffer;
private Exchanger<List<String>> exchanger;
public Producer(List<String> buffer, Exchanger<List<String>> exchanger){
this.buffer=buffer;
this.exchanger=exchanger;
}
@Override
public void run() {
for(int i=0;i<10;i++){
for (int j=0;j<10;j++)
buffer.add("Thrad "+Thread.currentThread().getName()+" : "+i+" "+j);
try {
buffer=exchanger.exchange(buffer);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public class Consumer implements Runnable {
private Exchanger<List<String>> exchanger;
private List<String> buffer;
public Consumer(List<String> buffer,Exchanger<List<String>> exchanger) {
this.exchanger = exchanger;
this.buffer = buffer;
}
@Override
public void run() {
for(int i=0;i<10;i++){
try {
buffer=exchanger.exchange(buffer);
} catch (InterruptedException e) {
e.printStackTrace();
}
for(int j=0;j<10;j++){
String message=buffer.get(0);
System.out.println(message);
buffer.remove(0);
}
}
}
}
public class Main {
public static void main(String[] args) {
List<String> buffer1=new ArrayList<>();
List<String> buffer2=new ArrayList<>();
Exchanger<List<String>> exchanger=new Exchanger<>();
Producer producer=new Producer(buffer1,exchanger);
Consumer consumer=new Consumer(buffer2,exchanger);
Thread t1=new Thread(producer);
Thread t2=new Thread(consumer);
t1.start();
t2.start();
}
}
BlockingQueue
我們可以使用更為方便安全的阻塞式集合來實作生產消費者模型，
這類集合具有的特點是：當集合已滿或者是為空的時候，被呼叫的方法不會立即執行，該方法將被阻塞，直到可以成功執行為止，
public class Channel {
private BlockingQueue<String> blockingQueue=new ArrayBlockingQueue<>(10);
public String get(){
String str=null;
try {
str=blockingQueue.take();
} catch (InterruptedException e) {
e.printStackTrace();
}
return str;
}
public void set(String str){
try {
blockingQueue.put(str);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
這次的Channel類是不是比之前的簡潔了許多，有了BlockingQueue我們就不用再去寫wait和notify了，
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/142935.html
標籤：Java
上一篇：泛型小記