Worker-Thread设计模式

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

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

//产品传送带，在传送带上除了负责产品加工的工人之外，还有在传送带上等待加工的产品
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)
                {
                }
            }
            //将产品放到传送带，并且通知工人线程工作
            productionQueue[tail] = production;
            tail = (tail + 1) % productionQueue.length;
            total++;
            this.notifyAll();
        }
    }
    //工人线程（Worker）从传送带上获取产品，并且进行加工
    public Production takeProduction()
{
        synchronized (this)
        {
            //当传送带上没有产品时，工人等待着产品从上游输送到传送带上
            while (total <= 0)
            {
                try
                {
                    this.wait();
                } catch (InterruptedException e)
                {
                }
            }
            //获取产品
            Production prod = productionQueue[head];
            head = (head + 1) % productionQueue.length;
            total--;
            this.notifyAll();
            return prod;
        }
    }
}

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.takeProduction();
                System.out.println(getName() + " process the " + production);
                //对产品进行加工
                production.create();
                TimeUnit.SECONDS.sleep(random.nextInt(10));
            } catch (InterruptedException e)
            {
                e.printStackTrace();
            }
        }
    }
}

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()
        );
    }
}

...省略
execute the 3 second process
Worker-0 process the PROD:5
execute the 5 first process
execute the 5 second process
Worker-4 process the PROD:7
Worker-0 process the PROD:6
execute the 7 first process
execute the 6 first process
execute the 7 second process
execute the 6 second process
...省略