Java高并发编程详解：多线程与系统设计笔记 - 第26章 Worker Thread 设计模式 - 《Java》

1.什么是Worker Thread 设计模式
2.Worker Thread模式实现

1.什么是Worker Thread 设计模式

Worker-Thread模式有时也称为流水线设计模式，这种设计模式类似于工厂流水线，上游工作人员完成了某个电子产品的组装之后，将半成品放到流水线传送带上，接下来的加工工作则会交给下游的工人，如图所示。

2.Worker Thread模式实现

根据我们前面的描述可以看出Worker-Thread模式需要有如下几个角色。

流水线工人：流水线工人主要用来对传送带上的产品进行加工。
流水线传送带：用于传送来自上游的产品。
产品组装说明书：用来说明该产品如何组装。

2.1 产品及组装说明书

// 在流水线上需要被加工的产品，create作为一个模版方法，提供了加工产品的说明书
public abstract class InstructionBook {
    public final void create() {
        this.firstProcess();
        this.secondProcess();
    }
    protected abstract void firstProcess();
    protected abstract void secondProcess();
}

抽象类InstructionBook，代表着组装产品的说明书，其中经过流水线传送带的产品将通过create() 方法进行加工，而firstProcess() 和secondProcess() 则代表着加工每个产品的步骤，这就是说明书的作用。
传送带上的产品除了说明书以外还需要有产品自身，产品继承了说明书，每个产品都有产品编号，通用Production的代码如清单所示。

public class Production extends InstructionBook {
    // 产品编号
    private final int proID;
    public Production(int proID) {
        this.proID = proID;
    }
    @Override
    protected void firstProcess() {
        System.out.println("execute the " + proID + " first process");
    }
    @Override
    protected void secondProcess() {
        System.out.println("execute the " + proID + " second process");
    }
}

2.2流水线传送带

流水线的传送带主要用于传送待加工的产品，上游的工作人员将完成的半成品放到传送带上，工作人员从传送带上取下产品进行再次加工，传送带Production Channel的代码所示。

// 产品传送带， 在传送带上除了负责产品加工的工人之外，还有在传送带上等待加工的产品
public class ProductionChannel {
    // 传送带上最多可以有多少个待加工的产品
    private final static int MAX_PROD = 100;
    // 主要用来存放待加工的产品，也就是传送带
    private final Production[] productionQueue;
    // 队列尾
    private int tail;
    // 队列头
    private int head;
    // 当前在流水线上有多少个待加工的产品
    private int total;
    // 在流水线上工作的工人
    private final Worker[] workers;
    // 创建ProductionChannel时应指需要多少个流水线工人
    public ProductionChannel(int workerSize) {
        this.workers = new Worker[workerSize];
        this.productionQueue = new Production[MAX_PROD];
        // 实例化每一个工人（Worker线程）并且启动
        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 >= productionQueue.length ) {
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            // 将产品放到传送带，并且通知工人线程工作
            productionQueue[tail] = production;
            tail = (tail + 1) % productionQueue.length;
            total++;
            this.notifyAll();
        }
    }
    // 工人线程（Worker）从传送带上获取产品，并且进行加工
    public Production takeProductino() {
        synchronized (this) {
            //当传送带上没有产品时，工人等待着产品从上游传送到传送带上
            while(total <= 0 ) {
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            // 获取产品
            Production prod = productionQueue[head];
            head = (head+1) % productionQueue.length;
            total--;
            this.notifyAll();
            return prod;
        }
    }
}

传送带是搁置产品的地方，如果工人们处理比较慢则会导致无限制的产品积压，因此我们需要做的是让上游的流水线阻塞并等待，直至流水线有位置可以用于放置新的产品为止，MAX_PROD的作用就在于此，其用于控制传送带的最大容量，传送带被创建的同时，流水线上的工人们也已经就绪到位，等待着流水线上产品的到来。

2.3流水线工人

流水线工人是Thread的子类，不断地从流水线上提取产品然后进行再次加工，加工的方法是create() (对该产品的加工方法说明书) ，流水线工人示例代码如所示。

import java.util.Random;
import java.util.concurrent.TimeUnit;
public class Worker extends Thread{
    private final ProductionChannel channel;
    //主要用于获取一个随机值，模拟加工一个产品需要耗费一定的时间，当然每个工人操作时所花费的时间可也能不一样
    private final static Random random = new Random(System.currentTimeMillis());
    public Worker(String workerName, ProductionChannel channel) {
        super(workerName);
        this.channel = channel;
    }
    @Override
    public void run() {
        while(true) {
            try {
                // 从传送带上获取产品
                Production production = channel.takeProductino();
                System.out.println(getName() + " process the " + production);
                production.create();
                TimeUnit.SECONDS.sleep(random.nextInt(10));
            }catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

2.4 流水线测试

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;
import static java.util.concurrent.ThreadLocalRandom.current;
public class Test {
    public static void main(String[] args) {
        // 流水线上有5个工人
        final ProductionChannel channel = new ProductionChannel(5);
        AtomicInteger productionNo = new AtomicInteger();
        // 流水线上有8个工作人员往传送带上不断地放置等待加工的半成品
        IntStream.range(1,8).forEach(i -> new Thread(
                () -> {
                    while(true) {
                        channel.offerProduction(new Production(productionNo.getAndIncrement()));
                        try {
                            TimeUnit.SECONDS.sleep(current().nextInt(10));
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
        ).start());
    }
}

2.5 Worker-Thread和Producer-Consumer

(1) Producer-Consumer模式

图是生产者消费者模式关键角色之间的关系图。

首先Producer、Consumer对Queue都是依赖关系，其次Producer要做的就是不断地往Queue中生产数据，而Consumer则是不断地从Queue中获取数据，Queue既不知道Producer的存在也不知道Consumer的存在，最后Consumer对Queue中数据的消费并不依赖于数据本身的方法(使用说明书)。

(2) Worker-Thread模式

图是Worker-Thread模式关键角色之间的关系图。
左侧的线程，也就是传送带上游的线程，同样在不断地往传送带(Queue) 中生产数据，而当Channel被启动的时候，就会同时创建并启动若干数量的Worker线程，因此我们可以看出Worker于Channel来说并不是单纯的依赖关系，而是聚合关系， Channel必须知道Worker的存在。