Worker-Thread设计模式

? ? Worker-Thread模式类似于工厂流水线，有时也称为流水线设计模式。线程池在某种意义上也算是Worker-Thread模式的一种实现，线程池初始化时创建线程类似于在流水线等待工作的工人，提交给线程池的Runnable接口类似于需要加工的产品，Runnable的run方法相当于组装该产品的说明书。Worker-Thread模式需要如下几个角色：

• 流水线工人：对传送带上的产品进行加工

• 流水线传送带：用于传送来自上游的产品

• 产品组装说明书：用于说明该产品如何组装

? ? ?Worker-Thread模式中生产线保存了在处理中的产品，并且是启动生产线的线程后，生产线启动若干数量的流水线工人线程 ，生产线聚合了产品和工人。生产者消费者模式是单纯的依赖关系，生产者和消费者都依赖产品队列，生产者和消费者是相互不知道。??

示例代码如下：

public abstract class InstructionBook {
protected abstract void firstProcess();
protected abstract void secondProcess();

public final void create() {
this.firstProcess();
this.secondProcess();
}

}

public class Production extends InstructionBook{
private final int productId;

public Production(int productionId) {
this.productId=productionId;
}

public int getProductionId() {
return this.productId;
}

@Override
protected void firstProcess() {
System.out.println("execute the "+this.productId+" first process");
}

@Override
protected void secondProcess() {
System.out.println("execute the "+this.productId+" second process");
}

}

import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;


public class Worker extends Thread{
private final ProductionChannel channel;

public Worker(String workerName, ProductionChannel channel) {
super(workerName);
this.channel=channel;
}

public void run() {
while(true) {
try {
Production production=this.channel.takeProduction();
System.out.println(getName()+ " process the "+production.getProductionId());
production.create();
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5));
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

}

import java.util.ArrayList;


public class ProductionChannel {
private final ArrayList<Production> productionQueue=new ArrayList<>();
private int total=20;
private final Worker[] workers;

public ProductionChannel(int workerSize) {
this.workers=new Worker[workerSize];
for(int i=0;i<workerSize;i++) {
workers[i]=new Worker("Worker-"+i,this);
workers[i].start();
}
}

public void offerProduction(Production production) {
synchronized(this) {
while(total<=this.productionQueue.size()) {
try {
System.out.println("processing production id="+production.getProductionId()+" , in waiting state");
this.wait();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
System.out.println("processing production id="+production.getProductionId());
this.productionQueue.add(production);
this.notifyAll();
}
}

public Production takeProduction() {
synchronized(this) {
while(this.productionQueue.size()<=0) {
try {
System.out.println("processing to fetch production, while in waiting state");
this.wait();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
this.notifyAll();
Production p=this.productionQueue.get(0);
this.productionQueue.remove(0);
return p;
}
}

}

import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;

public class WTtest {

public static void main(String[] args) {
ProductionChannel channel=new ProductionChannel(5);
AtomicInteger pid=new AtomicInteger();
IntStream.range(1, 8).forEach(i->new Thread(()->{
// while(true) {
channel.offerProduction(new Production(pid.getAndIncrement()));
try {
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5));
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
// }
}).start());
}

}