考虑一个包含任务的队列(task1、task2、task3、task1、task2、task3、...)，如何使用两个线程来删除队列中的任务

Question

案例：考虑到您有一个包含任务的队列(task1、task2、task3、task1、task2、task3、…)，我如何使用两个线程来删除队列中的任务。

要求：

流量图

代码

    public class testRunManager {

    public static void main(String[] args){
        ConcurrentLinkedQueue<BaseTask> tasks = new  ConcurrentLinkedQueue<>();
        for (int index = 0; index < 10; index++) {
            tasks.add(new Task1());
            tasks.add(new Task2());
            tasks.add(new Task3());
        }
        BaseRunManager.getInstance().addTasks(tasks);
        Thread thread1 = BaseRunManager.getInstance().getNewThread(TaskThread.Type.BeforeWards);
        Thread thread2 = BaseRunManager.getInstance().getNewThread(TaskThread.Type.AfterWards);

        //start
        thread1.start();
        thread2.start();


    }

    }

    public class TaskThread extends Thread{
    enum Type{
        BeforeWards,
        AfterWards
    }

    Type type;


    public TaskThread(Type type,Runnable runnable){
        super(runnable);
        this.type = type;

    }

    }

    public interface ShouldRunBeforeWardsJob {

    }

    public interface ShouldRunAfterWardsJob {

    }
    abstract public class RunController {
    public enum Performance {
        BUSYING,
        RUNNING,
        PAUSED,
    }

    protected enum ControlState {
        PAUSING,
        PAUSED,
        STOPING,
        RESUMING,
        RUNNING,
        STEPPING,
    }

    private ControlState state = ControlState.RUNNING;
    private Performance performance = Performance.BUSYING;
    private List<ControlListener> listeners = new ArrayList<>();

    protected ReentrantLock controlLock  = new ReentrantLock();
    protected Condition controlCondition = controlLock.newCondition();

    public Performance getPerformance() {
        return performance;
    }

    protected ControlState getState() {
        return state;
    }

    public void addListener(ControlListener listener) {
        listeners.add(listener);
    }

    public void removeListener(ControlListener listener) {
        listeners.remove(listener);
    }

    public void pause() {
        if (performance != Performance.RUNNING) {
            return;
        }
        setState(ControlState.PAUSING);
    }

    public void step() {
        if (performance != Performance.PAUSED) {
            return;
        }
        setState(ControlState.STEPPING);
    }

    public void stop() {
        if (performance != Performance.RUNNING && performance != Performance.PAUSED) {
            return;
        }
        setState(ControlState.STOPING);
    }

    public void resume() {
        if (performance != Performance.PAUSED) {
            return;
        }
        setState(ControlState.RESUMING);
    }

    private void setPerformance(Performance p) {
        if (performance != p) {
            Performance old = this.performance;
            this.performance = p;
            for (ControlListener cl : listeners) {
                cl.performChanged(old, p);
            }
        }
    }

    protected void setState(ControlState state) {
        controlLock.lock();
        try {
            this.state = state;
            switch (this.state) {
                case RESUMING:
                case STEPPING:
                case PAUSING:
                case STOPING:
                    controlCondition.signal();
                    setPerformance(Performance.BUSYING);
                    break;
                case PAUSED:
                    setPerformance(Performance.PAUSED);
                    break;
                case RUNNING:
                    setPerformance(Performance.RUNNING);
            }
        }finally {
            controlLock.unlock();
        }
    }

    public interface ControlListener {
        void performChanged(Performance oldState, Performance newState);
    }
    }   

    public abstract class BaseTask {
    enum State{
        FINISH,
        NOT
    }

    protected State state;

    public State getState(){
        return state;
    }

    public void setState(State state){
        this.state = state;
    }

    abstract void  runJob();

    abstract void doJob();

    }




    public class BaseRunManager {                
       private static BaseRunManager instance;
       private ConcurrentLinkedQueue<BaseTask> tasks = new 
    ConcurrentLinkedQueue<>();

            public synchronized static BaseRunManager getInstance(){
                if(instance == null){
                    instance = new BaseRunManager();
                }

                return instance;
            }

            public BaseRunManager(){

            }

            public void addTasks(ConcurrentLinkedQueue<BaseTask> tasks){
                this.tasks = tasks;
            }

            public Thread getNewThread(TaskThread.Type type){
                return new TaskThread(type,new BaseRunnable());
            }


            private  class  BaseRunnable extends RunController implements Runnable{
                private BaseTask curTask;
                private final AtomicBoolean afterwardsFinish = new AtomicBoolean(true);
                private final AtomicInteger beforewardsFinishNum = new AtomicInteger(0);
                private final AtomicInteger currentThreadNum = new AtomicInteger(0);
                 private final Condition condition = controlLock.newCondition();
                @Override
                public void run() {
                    currentThreadNum.incrementAndGet();
                    TaskThread curThread = (TaskThread)Thread.currentThread();


                    while (tasks.size()>0) {
                        //get task
                        controlLock.lock();
                        try {

                            curTask = tasks.peek();
                            if ((curTask instanceof ShouldRunBeforeWardsJob && curThread.type == TaskThread.Type.BeforeWards)
                                    || (curTask instanceof ShouldRunAfterWardsJob && curThread.type == TaskThread.Type.AfterWards)) {
                                tasks.poll();                      

                                if (curTask instanceof ShouldRunBeforeWardsJob) {
                                    curTask.runJob();
                                    beforewardsFinishNum.incrementAndGet();
                                    condition.signalAll();

                                } else if (curTask instanceof ShouldRunAfterWardsJob) {

                                    if (beforewardsFinishNum.get() / 2 != 0) {                               
                                        condition.await();
                                        curTask.runJob();
                                    }

                                }
                            } else {
                                condition.awaitNanos(20);
                                continue;
                            }

                        } catch (Exception e) {
                            e.printStackTrace();
                        } finally {
                            controlLock.unlock();
                        }

                    }            

                }

            }

        }

Answer 1

下面是另一种方法：

public static void main(String[] args) {
    final BlockingQueue<BaseTask> tasks = new LinkedBlockingQueue<>();
    final BlockingQueue<BaseTask> queue2 = new LinkedBlockingQueue<>();
    for (int index = 0; index < 10; index++) {
        tasks.add(new BaseTask("Task1"));
        tasks.add(new BaseTask("Task2"));
        tasks.add(new BaseTask("Task3"));
    }
    Thread thread1 = new Thread(new Runnable() {
        @Override
        public void run() {
            while (true) {
                try {
                    BaseTask task = tasks.take();
                    task.run();
                    task = tasks.take();
                    task.run();
                    task = tasks.take();
                    queue2.offer(task);
                } catch (InterruptedException ex) {
                    Logger.getLogger(Main.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
        }
    });
    Thread thread2 = new Thread(new Runnable() {
        @Override
        public void run() {
            while (true) {
                try {
                    BaseTask task = queue2.take();
                    task.run();
                } catch (InterruptedException ex) {
                    Logger.getLogger(Main.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
        }
    });
    thread2.start();
    thread1.start();
}

private static class BaseTask implements Runnable {
    private final String name;

    public BaseTask(String name) {
        this.name = name;
    }

    @Override
    public void run() {
        System.out.println(name + " ["
                + Thread.currentThread().getName() + "]");
    }
}