java - 使用线程将电子邮件复制到 javamail 中的文件夹时遇到问题

Question

我在使用线程将一些电子邮件复制到其他文件夹时遇到问题，我的问题是，代码不会等待完成作业。

我想通过线程移动消息来加速工作，但我需要等待才能移动所有消息，那么我该怎么做？

private static void moveMessagesToFolders(List<Message> listMessages, Store store, Set<String> setSender) throws MessagingException {

    HashMap<String, List<Message>> mapMessages = separeteMessagesBySender(listMessages, setSender);

    for (Entry<String, List<Message>> mapMessage : mapMessages.entrySet()) {
        Message[] messageArray = mapMessage.getValue().toArray(new Message[mapMessage.getValue().size()]);
        moveMessagesThread(messageArray, mapMessage, store);
    }
}

private static void moveMessagesThread(Message[] messageArray, Entry<String, List<Message>> mapMessage, Store store) {
    Set<Thread> setThread = createMovimentSetThread(messageArray, mapMessage, store);

    for (Thread thread : setThread) {
        thread.start();
    }
}

private static Set<Thread> createMovimentSetThread(Message[] messageArray, Entry<String, List<Message>> mapMessage, Store store) {

    int [] threadIndexs = MathUtil.generateIndex(messageArray);
    Set<Thread> setThread = new HashSet<>(threadIndexs.length);

    for (int i = 0; i < threadIndexs.length; i++) {
        setThread.add(new ThreadMoveMessages(messageArray, mapMessage, store, threadIndexs[i]));
    }

    return setThread;
}

在我更改此实现执行器的方法后。

private static void moveMessagesThread(Message[] messageArray, Entry<String, List<Message>> mapMessage, Store store) {

        int [] threadIndexs = MathUtil.generateIndex(messageArray);
        ExecutorService executor = Executors.newFixedThreadPool(4);

        for (int i = 0; i < 4; i++) {
            executor.execute(new ThreadMoveMessages(messageArray, mapMessage, store, threadIndexs[i]));
        }

        executor.shutdown();
    }

实现类Thread

public class ThreadMoveMessages implements Callable<Boolean> {

    private Entry<String, List<Message>> mapMessage;
    private Store store;
    private Message[] messageArray;
    private static int indexControler;
    private static int indexLimit;

    public ThreadMoveMessages(Message[] messageArray, Entry<String, List<Message>> mapMessage, Store store, int indexEnd) {
        this.messageArray = Arrays.copyOf(messageArray, indexEnd);
        this.indexControler += indexEnd;
        this.indexLimit = indexControler;
        this.mapMessage = mapMessage;
    }

    @Override
    public Boolean call() throws Exception {
        Folder folder = null;
        try {
            folder = this.store.getDefaultFolder().getFolder(this.mapMessage.getKey());
            folder.open(Folder.READ_WRITE);
            folder.appendMessages(this.messageArray);
            EmailUtil.deleteListMessage(this.mapMessage.getValue());
        } catch (MessagingException e) {
            e.printStackTrace();
        }                       
        return true;
    }
}

Answer 1

如果您想以异步方式等待计算并等待结果，则应该使用 Futures 和 Callables。

实现Callable接口(interface):

class MoveMessages implements Callable<Boolean> {

@Override
public Boolean call() throws Exception {
    boolean success = true;

    // Your implementation here

    return success;
}

}

接下来将其提交给Executor并检索Future，在Future上调用get()，您将等待Callable的计算完成。

ExecutorService executor = Executors.newFixedThreadPool(5);

MoveMessages moveMessages = new MoveMessages();
Future<Boolean> submit = executor.submit(moveMessages);

Boolean integer = submit.get(); // Will wait until task is finished

executor.shutdown();

当然，您可以提交更多任务，将所有任务都列出来并等待所有任务完成。

编辑:

好的，首先您说需要等待所有消息都被移动，因此这种情况的一种方法是将 Future 和 callable 与 ExecutorService 一起使用。使用ExecutorService，您不需要创建和启动大量新线程。请记住创建新线程会产生成本。在代码中，您为每个发送者创建 4 个新线程，使用 ExecutorService 只创建固定数量的线程并为每个发送者重用它们。这是使用 Executors 和 Futures 的示例，请注意，ExecutorService 是为调用 moveMessagesToFolders 创建一次:

private static ExecutorService executor 


private static void moveMessagesToFolders(List<Message> listMessages, Store store, Set<String> setSender) throws MessagingException {
    executor = Executors.newFixedThreadPool(4);

    HashMap<String, List<Message>> mapMessages = separeteMessagesBySender(listMessages, setSender);

    for (Map.Entry<String, List<Message>> mapMessage : mapMessages.entrySet()) {
        Message[] messageArray = mapMessage.getValue().toArray(new Message[mapMessage.getValue().size()]);
        moveMessagesThread(messageArray, mapMessage, store);
    }

    executor.shutdown();
}

private static void moveMessagesThread(Message[] messageArray, Map.Entry<String, List<Message>> mapMessage, Store store) {
    List<Future<Boolean>> futures = createMovimentSetThread(messageArray, mapMessage, store);

    for (Future<Boolean> future : futures) {
        try {
            Boolean success = future.get(); // Will wait to accomplished all submited Callables
            if(!success) { // Check if all submited callables end succesulfy 
                throw new RuntimeException("Something goes wrong while moving messages");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

}

private static List<Future<Boolean>> createMovimentSetThread(Message[] messageArray, Map.Entry<String, List<Message>> mapMessage, Store store) {
    int [] threadIndexs = MathUtil.generateIndex(messageArray);
    List<Future<Boolean>> futures = new ArrayList<>();

    for (int i = 0; i < threadIndexs.length; i++) {
        Future<Boolean> submit = executor.submit(new ThreadMoveMessages(messageArray, mapMessage, store, threadIndexs[i]));
        futures.add(submit);
    }

    return futures;
}

根据您的评论，使用 Fork/Join 框架将数组拆分为更小的部分可能是更好的解决方案。在谷歌上查找更多信息。一些链接:

• http://docs.oracle.com/javase/tutorial/essential/concurrency/forkjoin.html
• Java 7: Fork/Join Framework