使用線程將問題郵件複製到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接口：

class MoveMessages implements Callable<Boolean> { 

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

    // Your implementation here 

    return success; 
} 

}

而旁邊提交給執行和檢索Future，在未來調用get()你會等到贖回的計算完成。

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(); 

當然，你可以提交更多的任務獲取所有列表，並等待所有將完成。

編輯：

OK，首先你說你需要等到所有郵件被移動，所以這種情況下，一個方法是使用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; 
} 

根據您的意見，您拆分陣列使用叉子更小的碎片/ join框架可能是一個更好的解決方案。在谷歌查看一些更多的信息。有些鏈接：

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