10个线程同步处理1000行消息

多线程，并发，是经常遇到的问题，平时解决的方案也想过很多，比如说现在有1000行消息，需要开10个线程同时处理。

之前想过两个方案：

方案一： 一次开10个线程，每个线程处理一条消息，等10个线程全部处理结束之后，再开启下10个线程，直到全部处理完毕

缺陷：需要等待其他n - 1个线程结束后，才能同时启动下n个线程

方案二： 将1000行消息分割为10份，每100行用一个线程处理。

优点：无等待

缺陷： 分割不均，无法充分利用所有的线程

现在想想，上面两个缺点挺多，就又想了两种方案：

方案三：使用ConcurrentLinkedQueue<Task>存储所有的Task，然后同时开启n个线程读取Queue.

优点：充分利用所有线程，无等待

缺点：需要将所有的task转移到Queue中，消耗一倍内存

方案四：使用java.util.concurrent包，固定数量线程池。

优点：完美解决

缺点：当单个task执行时间很短的时候，线程池中的线程并不会被全部使用，这样很多task就会block在一个线程中，降低执行速率

下面贴出每个方案的代码实现，备忘吧，如果有更好的想法，或者更简单的方式，再继续补充~

public class Task {

    private int id;

    public Task(int id) {
        this.id = id;
    }

    public void start() {
        System.out.println(Thread.currentThread().getName() + ": start to handle task " + id);

        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

import java.util.LinkedList;
import java.util.List;

public class TaskProducer {

    public static List<Task> produce(int count) {
        List<Task> tasks = new LinkedList<Task>();

        for(int i = 0; i < count; i ++) {
            tasks.add(new Task(i + 1));
        }

        return tasks;
    }

}

import java.util.LinkedList;
import java.util.List;

/**
 * 策略1: 每次开启n个线程，等待n个线程全部结束之后，再开启下n个线程，每个线程处理一个task.
 * 缺陷：需要等待其他n - 1个线程结束后，才能同时启动下n个线程
 */
public class Strategy1 {

    public static void main(String[] args) {
        List<Task> tasks = TaskProducer.produce(1000);
        handleTasks(tasks, 10);
        System.out.println("All finished");
    }

    public static void handleTasks(List<Task> tasks, int threadCount) {
        int taskCount = tasks.size();

        List<Thread> threadHolder = new LinkedList<Thread>();
        for(int i = 0; i < taskCount; i += threadCount) {
            for(int j = 0; j < threadCount && (i + j) < taskCount; j ++) {
                Thread thread = new Thread(new TaskHandler(tasks.get(i + j)));
                threadHolder.add(thread);
                thread.start();
            }

            waitToFinish(threadHolder);
            threadHolder.clear();
        }
    }

    public static void waitToFinish(List<Thread> threadHolder) {
        while(true) {
            boolean allFinished = true;
            for(Thread thread : threadHolder) {
                allFinished = allFinished && !thread.isAlive();
            }

            if(allFinished) {
                break;
            }
        }
    }

    public static class TaskHandler implements Runnable {
        private Task task;

        public TaskHandler(Task task) {
            this.task = task;
        }

        @Override
        public void run() {
            task.start();
        }
    }

}

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;

/**
 * 策略2: 将所有的task分割成n个子task列表，然后开启n个线程，每个线程处理一个子列表
 * 优点：无等待
 * 缺陷： 分割不均，无法充分利用所有的线程
 */
public class Strategy2 {

    public static void main(String[] args) {
        List<Task> tasks = TaskProducer.produce(1000);
        handleTasks(tasks, 10);
        System.out.println("All finished");
    }

    public static void handleTasks(List<Task> tasks, int threadCount) {
        List<List<Task>> splitTasks = splitTasksToNThreads(tasks, threadCount);

        List<Thread> threadHolder = new LinkedList<Thread>();
        for (List<Task> segment : splitTasks) {
            Thread thread = new Thread(new TaskHandler(segment));
            threadHolder.add(thread);
            thread.start();
        }

        waitToFinish(threadHolder);
    }

    public static void waitToFinish(List<Thread> threadHolder) {
        while(true) {
            boolean allFinished = true;
            for(Thread thread : threadHolder) {
                allFinished = allFinished && !thread.isAlive();
            }

            if(allFinished) {
                break;
            }
        }
    }

    public static List<List<Task>> splitTasksToNThreads(List<Task> tasks, int threadCount) {
        List<List<Task>> splitTasks = new ArrayList<List<Task>>(threadCount);

        int taskCount = tasks.size();
        int taskPerThread = new BigDecimal(taskCount).divide(new BigDecimal(threadCount), RoundingMode.CEILING).intValue();

        for (int i = 0; i < taskCount; i += taskPerThread) {
            List<Task> segment = new LinkedList<Task>();
            for (int j = 0; j < taskPerThread && (i + j) < taskCount; j++) {
                segment.add(tasks.get(i + j));
            }

            splitTasks.add(segment);
        }

        tasks.clear();

        return splitTasks;
    }

    public static class TaskHandler implements Runnable {
        private List<Task> tasks;

        public TaskHandler(List<Task> tasks) {
            this.tasks = tasks;
        }

        @Override
        public void run() {
            for (Task task : tasks) {
                task.start();
            }
        }
    }

}

import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * 策略3: 使用ConcurrentLinkedQueue<Task>存储所有的Task，然后同时开启n个线程读取Queue.
 * 优点：充分利用所有线程，无等待
 * 缺点：需要将所有的task转移到Queue中，消耗一倍内存
 */
public class Strategy3 {

    public static void main(String[] args) {
        List<Task> tasks = TaskProducer.produce(1000);
        handleTasks(tasks, 10);
        System.out.println("All finished");
    }

    public static void handleTasks(List<Task> tasks, int threadCount) {
        Queue<Task> taskQueue = new ConcurrentLinkedQueue<Task>();
        taskQueue.addAll(tasks);

        List<Thread> threadHolder = new LinkedList<Thread>();
        for(int i = 0; i < threadCount; i ++) {
            Thread thread = new Thread(new TaskHandler(taskQueue));
            threadHolder.add(thread);
            thread.start();
        }

        waitToFinish(threadHolder);
    }

    public static void waitToFinish(List<Thread> threadHolder) {
        while(true) {
            boolean allFinished = true;
            for(Thread thread : threadHolder) {
                allFinished = allFinished && !thread.isAlive();
            }

            if(allFinished) {
                break;
            }
        }
    }

    public static class TaskHandler implements Runnable {

        private final Queue<Task> tasks;

        public TaskHandler(Queue<Task> tasks) {
            this.tasks = tasks;
        }

        public void run() {
            while(!tasks.isEmpty()) {
                Task task = tasks.poll();
                if(task != null) {
                    task.start();
                }
            }
        }

    }

}

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 策略4: 使用java.util.concurrent包，线程池。
 * 优点：完美解决。
 */
public class Strategy4 {

    public static void main(String[] args) {
        List<Task> tasks = TaskProducer.produce(1000);
        handleTasks(tasks, 10);
        System.out.println("All finished");
    }

    public static void handleTasks(List<Task> tasks, int threadCount) {
        try {
            ExecutorService executor = Executors.newFixedThreadPool(threadCount);

            for(Task task : tasks) {
                executor.submit(new TaskHandler(task));
            }

            executor.shutdown();
            executor.awaitTermination(60, TimeUnit.SECONDS);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static class TaskHandler implements Runnable {

        private Task task;

        public TaskHandler(Task task) {
            this.task = task;
        }

        public void run() {
            task.start();
        }

    }

}