java - 按顺序处理异步事件和发布结果

Question

我的目标是按顺序发布异步事件，这些事件也按顺序到达并需要任意时间进行处理。因此，下面是我当前仅使用 wait 和 notify 的实现。 MyThread 处理事件，将结果按 id 放入哈希表中，如果在按顺序发布此事件之前被阻塞，则通知 Scheduler 线程。

使用 java.util.concurrent 包实现此功能的更好、更简洁的方法是什么？

import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;


public class AsyncHandler {
    private final Map<Integer, Object> locks = new ConcurrentHashMap<Integer, Object>();
    private final Map<Integer, Result> results = new ConcurrentHashMap<Integer, Result>();
    private static final Random rand = new Random();

    public AsyncHandler () {
        new Scheduler(this, locks, results).start();
    }

    public void handleEvent(Event event) {
        System.out.println("handleEvent(" + event.id + ")");
        new MyThread(this, event, locks, results).start();
    }

    public Result processEvent (Event event) {
        System.out.println("processEvent(" + event.id + ")");
        locks.put(event.id, new Object());

        try {
            Thread.sleep(rand.nextInt(10000));
        } catch (InterruptedException e) {
            System.out.println(e);
        }

        return new Result(event.id);
    }

    public void postProcessEvent (Result result) {
        System.out.println(result.id);
    }

    public static void main (String[] args) {
        AsyncHandler async = new AsyncHandler();

        for (int i = 0; i < 100; i++) {
            async.handleEvent(new Event(i));
        }
    }
}

class Event {
    int id;

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

class Result {
    int id;

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

class MyThread extends Thread {
    private final Event event;
    private final Map<Integer, Object> locks;
    private final Map<Integer, Result> results;
    private final AsyncHandler async;

    public MyThread (AsyncHandler async, Event event, Map<Integer, Object> locks, Map<Integer, Result> results) {
        this.async = async;
        this.event = event;
        this.locks = locks;
        this.results = results;
    }

    @Override
    public void run () {
        Result res = async.processEvent(event);
        results.put(event.id, res);

        Object lock = locks.get(event.id);

        synchronized (lock) {
            lock.notifyAll();
        }
    }
}

class Scheduler extends Thread {
    private int curId = 0;
    private final AsyncHandler async;
    private final Map<Integer, Object> locks;
    private final Map<Integer, Result> results;

    public Scheduler (AsyncHandler async, Map<Integer, Object> locks, Map<Integer, Result> results) {
        this.async = async;
        this.locks = locks;
        this.results = results;
    }

    @Override
    public void run () {
        while (true) {
            Result res = results.get(curId);
            if (res == null) {
                Object lock = locks.get(curId);

                //TODO: eliminate busy waiting
                if (lock == null) {
                    continue;
                }

                synchronized (lock) {
                    try {
                        lock.wait();
                    } catch (InterruptedException e) {
                        System.out.println(e);
                        System.exit(1);
                    }
                }
                res = results.get(curId);
            }

            async.postProcessEvent(res);
            results.remove(curId);
            locks.remove(curId);
            curId++;
        }
    }
}

Answer 1

是的，并发库会简单得多。

ExecutorService 被设计为包装一个线程池和一个队列来为每个任务返回一个 Future 并提供任何等待结果的线程。

如果你想按顺序处理结果，有一个线程按顺序处理 future 的结果。

按照你能做的顺序处理异步结果

public class Main {
    public static void main(String[] args) {
        Main main = new Main();
        for (int i = 0; i < 1000; i++) {
            final int finalI = i;
            main.submitTask(new Callable<Long>() {
                @Override
                public Long call() throws Exception {
                    long millis = (long) (Math.pow(2000, Math.random()));
                    Thread.sleep(millis);
                    return millis;
                }
            }, new ResultHandler<Long>() {
                @Override
                public void onFuture(Future<Long> future) throws ExecutionException, InterruptedException {
                    System.out.println(new Date() + ": " + finalI + " - Slept for " + future.get() + " millis");
                }
            });
        }
        main.shutdown();
    }


    public interface ResultHandler<T> {
        void onFuture(Future<T> future) throws Exception;
    }

    private final ExecutorService pool = Executors.newFixedThreadPool(10);
    private final ExecutorService result = Executors.newSingleThreadExecutor();

    public synchronized <T> void submitTask(Callable<T> callable, final ResultHandler<T> resultHandler) {
        final Future<T> future = pool.submit(callable);
        result.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    resultHandler.onFuture(future);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

    public void shutdown() {
        pool.shutdown();
        result.shutdown();
    }
}

打印

Wed Oct 02 16:32:07 CEST 2013: 0 - Slept for 1 millis
Wed Oct 02 16:32:07 CEST 2013: 1 - Slept for 1899 millis
Wed Oct 02 16:32:09 CEST 2013: 2 - Slept for 32 millis
Wed Oct 02 16:32:09 CEST 2013: 3 - Slept for 32 millis
Wed Oct 02 16:32:09 CEST 2013: 4 - Slept for 214 millis
Wed Oct 02 16:32:09 CEST 2013: 5 - Slept for 366 millis
... many deleted ...
Wed Oct 02 16:32:09 CEST 2013: 82 - Slept for 6 millis
Wed Oct 02 16:32:09 CEST 2013: 83 - Slept for 1636 millis
Wed Oct 02 16:32:10 CEST 2013: 84 - Slept for 44 millis
Wed Oct 02 16:32:10 CEST 2013: 85 - Slept for 1 millis

您可以看到，虽然有些任务比其他任务花费的时间长得多，但输出的顺序是任务添加的顺序。您还可以看到它正在同一秒内(并发)处理许多任务