多线程设计模式：第四篇 - Thread-Per-Message

2018-10-18 本文已影响0人张angang强吖

一，Thread-Per-Message模式

Thread-Per-Message模式是说为每个请求都分配一个线程，由这个线程来执行处理。这里包含两个角色，请求的提交线程和请求的执行线程。

下面的示例代码中，Host 提交一个请求交给另外一个线程来处理。

/**
 * @author koma <komazhang@foxmail.com>
 * @date 2018-10-18
 */
public class Main {
    public static void main(String[] args) {
        System.out.println("Main BEGIN");
        Host host = new Host();
        host.request(10, 'A');
        host.request(20, 'B');
        host.request(30, 'C');
        System.out.println("Main END");
    }
}

public class Host {
    private final Helper helper = new Helper();

    public void request(final int count, final char c) {
        System.out.println("Request BEGIN");
        new Thread() {
            @Override
            public void run() {
                helper.handle(count, c);
            }
        }.start();
        System.out.println("Request END");
    }
}

public class Helper {
    public void handle(int count, char c) {
        System.out.println("Helper BEGIN");
        for (int i = 0; i < count; i++) {
            slowly();
            System.out.print(c);
        }
        System.out.println();
        System.out.println("Helper END");
    }

    private void slowly() {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Thread-Per-Message模式可以提高程序的响应性，但是并不能保证操作的顺序，而且没有返回值。一般用于简单服务器实现，当服务器主线程收到客户端的请求之后就创建一个新的线程去处理客户端的请求，而服务器主线程则返回继续接收客户端请求。

１，Executor 框架

Java 中通常是通过 new Thread 或者 new Runnable 实现类来创建线程，在 juc 包中提供了 Executors，ThreadFactory，ExecutorService 另外三种方式来更加方便的创建线程或者线程池等，在以后的代码中，我们推荐使用这种方式创建线程。

二，Worker-Thread模式

Worker-Thread模式也被称为 Background Thread (背景线程)模式，这种模式不同于 Thread-Per-Message 模式在任务到来时创建线程，而是会先创建一些工作线程等待任务到来，从这个角度来说 Worker-Thread 模式也可被称为线程池模式。

下面的示例程序模拟了Worker-Thread模式的工作过程，代码如下：

/**
 * @author koma <komazhang@foxmail.com>
 * @date 2018-10-18
 */
public class Main {
    public static void main(String[] args) {
        Channel channel = new Channel(5);
        channel.startWorkers(); //启动工作线程
        new ClientThread("ClientA", channel).start();
        new ClientThread("ClientB", channel).start();
        new ClientThread("ClientC", channel).start();
    }
}

public class Channel {
    private static final int MAX_REQUESTS = 100; //定义请求队列最大长度
    private final Queue<Request> queue;
    private final WorkerThread[] threadPool;

    public Channel(int threads) {
        queue = new LinkedList<>();

        threadPool = new WorkerThread[threads];
        for (int i = 0; i < threads; i++) {
            threadPool[i] = new WorkerThread("Worker-"+i, this);
        }
    }

    public void startWorkers() {
        for (int i = 0; i < threadPool.length; i++) {
            threadPool[i].start();
        }
    }

    public synchronized void putRequest(Request request) throws InterruptedException {
        while (queue.size() >= MAX_REQUESTS) {
            wait();
        }
        queue.offer(request);
        notifyAll();
    }

    public synchronized Request takeRequest() throws InterruptedException {
        while (queue.size() <= 0) {
            wait();
        }
        Request request = queue.poll();
        notifyAll();
        return request;
    }
}

public class Request {
    private final String name;
    private final int number;
    private static final Random random = new Random();

    public Request(String name, int number) {
        this.name = name;
        this.number = number;
    }

    public void execute() {
        System.out.println(Thread.currentThread().getName()+" execute "+this);
        try {
            Thread.sleep(random.nextInt(1000));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public String toString() {
        return "[ Request from "+name+", No."+number+" ]";
    }
}

public class ClientThread extends Thread {
    private final Channel channel;
    private static final Random random = new Random();

    public ClientThread(String name, Channel channel) {
        super(name);
        this.channel = channel;
    }

    @Override
    public void run() {
        try {
            for (int i = 0; true; i++) {
                Request request = new Request(getName(), i);
                channel.putRequest(request);
                Thread.sleep(random.nextInt(1000));
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

public class WorkerThread extends Thread {
    private final Channel channel;

    public WorkerThread(String name, Channel channel) {
        super(name);
        this.channel = channel;
    }

    @Override
    public void run() {
        while (true) {
            try {
                Request request = channel.takeRequest();
                request.execute();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

示例中 Channel 类和生产者-消费者模式中一样，我们也可以使用 Java 中的 BlockingQueue 来实现，从而省去我们自己写 wait/notify 逻辑的复杂性。

Worker-Thread模式中应用到了生产者-消费者模式，因为提交请求的线程和处理请求的线程不是同一个线程。Worker-Thread模式的优点在于它的工作线程是预先启动好的，这样节省了线程启动的时间损耗，提高了程序的性能，而且工作线程是可以重复利用的，节省了程序的资源消耗。

１，ThreadPoolExecutor

juc包通过 ThreadPoolExecutor 类轻松的实现了Worker-Thread模式，通过该类可以对工作线程做非常精细的控制，如下：

指定线程池的大小
指定创建线程的方式：提前创建，按需创建
指定线程创建的工厂类：更加细致的控制创建出来的线程
指定多长时间后终止不再需要的线程
指定传递任务的队列的排队方式
指定当队列饱和时拒绝请求的方式
指定线程执行工作前，执行工作后的处理动作

通常我们会通过 Executors 类中的静态方法来创建线程池。利用 Executor 框架改写后的 WorkerThread 类，以及启动 Worker 的代码如下：

public class WorkerThread implements Runnable {
    //代码不变，仅仅把 WorkerThread 改成实现 Runnable 接口
    //为了传递到 Executors 类中
}

public class Channel {
    //创建工作线程池
    private final ExecutorService executorService = Executors.newFixedThreadPool(5);

    //其它代码不变，记得将原来的 threadPool 代码注释掉
    
    //从线程池中启动工作线程
    public void startWorkers() {
        executorService.execute(new WorkerThread(this));
    }
}

多线程设计模式：第四篇 - Thread-Per-Message

一，Thread-Per-Message模式

１，Executor 框架

二，Worker-Thread模式

１，ThreadPoolExecutor

猜你喜欢

热点阅读