⑧ Worker Thread 模式 - 《并发编程》

工作没来就一直等待，工作来了就干活。

突然想起了当年在外面去富士康工作的经验，流水线作业。一根皮带，贯穿南北，我负责安装相机皮圈，来一个手机我就安一个，没来，我就等待着，和工友扯皮，当然来了一样可以扯皮。

那么我们就先来个一个车间，车间里可以容纳 N 个工人，外面会一直有工作不定时的发布。然后工人拿到这个工作，开始工作。

其实我们会发现这个和生产者消费者的模式很形似，只不过针对的对象不同，只需要针对客户线程，而工人线程怎么做我们就不必关系。只关心任务的发布即可。消费者和生成者则是两端都要关心。本质是一致的，对外的表现是不同的。

public class Channel {
    private final int MAX_REQUEST = 10;
    private WorkThread[] threadPool;
    private Request[] requests;
    private int count;//工作线程的数量
    private int tail;//下次放的位置
    private int head;//下次拿的位置
    public Channel(int threadNum) {
        threadPool = new WorkThread[threadNum];
        requests = new Request[MAX_REQUEST];
        for (int i = 0; i < threadPool.length; i++) {
            threadPool[i] = new WorkThread("work-" + i, this);
        }
    }
    public synchronized void putRequest(Request request) {
        while (count >= requests.length) {
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        requests[tail] = request;
        tail = (tail + 1) % MAX_REQUEST;
        count++;
        notifyAll();
    }
    public synchronized Request poll() {
        while (count <= 0) {
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        Request request = requests[head];
        head = (head + 1) % MAX_REQUEST;
        count--;
        notifyAll();
        return request;
    }
    public void startWorks() {
        for (int i = 0; i < threadPool.length; i++) {
            threadPool[i].start();
        }
    }
}

public class Request {
    private String name;
    private int number;
    private Random random = new Random();
    public Request(String name, int number) {
        this.name = name;
        this.number = number;
    }
    public void exec() {
        System.out.println(Thread.currentThread().getName()+"  exec:" + name + "--" + number);
        try {
            Thread.sleep(random.nextInt(1000));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

public class ClientThread extends Thread {
    private Channel channel;
    private String name;
    private int number;
    private Random random = new Random();
    public ClientThread(String name, Channel channel) {
        this.channel = channel;
        this.name = name;
    }
    @Override
    public void run() {
        super.run();
        while (true) {
            channel.putRequest(new Request(name, number++));
            try {
                Thread.sleep(random.nextInt(1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

public class WorkThread extends Thread {
    private final Channel channel;
    public WorkThread(String name, Channel channel) {
        super(name);
        this.channel = channel;
    }
    @Override
    public void run() {
        super.run();
        while (true) {
            Request request = channel.poll();
            request.exec();
        }
    }
}

public class Main {
    public static  void main(String[] args){
        Channel channel = new Channel(5);
        channel.startWorks();
       new ClientThread("AAA",channel).start();
       new ClientThread("BBB",channel).start();
       new ClientThread("CCC",channel).start();
    }
}

运行结果：