10件のメールアカウントの未読メールをマルチスレッドで読みたいです。
ただし、スレッド プール サイズが 5 の場合、スレッド プールから 5 つのスレッドが使用されます。各スレッドは、1 つのメール アカウントを読み取ります。そのため、Thread_1 が最初のメール ボックスを読み取ると、mailbox_6 を読み取る必要があります。次に、スレッド 2 が mailbox_7 を読み取ります。
すべてのメール アカウントが一度読み込まれると、最初のメール アカウントからサイクルが開始されます。
Javaでこれを行うにはどうすればよいですか?
これはかなり簡単なはずです。5 つのスレッドで固定スレッド プールを作成し、10 個のジョブをプールに送信します (ユーザーの電子メール アカウントごとに 1 つ)。
// create a thread pool with 5 workers
ExecutorService threadPool = Executors.newFixedThreadPool(5);
// submit all 10 user email accounts to the pool to be processed in turn
for (UserEmail userEmail : userEmailsToProcess) {
threadPool.submit(new EmailProcessor(userEmail));
}
// once we have submitted all jobs to the thread pool, it should be shutdown
threadPool.shutdown();
...
// here's our runnable class which does the actual work
public class EmailProcessor implements Runnable {
private UserEmail userEmail;
public MyJobProcessor(UserEmail userEmail) {
this.userEmail = userEmail;
}
public void run() {
// read the user email
...
}
}
クラスは、UserEmail「読む」電子メールのファイル名、またはアカウント名などを保持できます。メールアカウントと読まれるメールをどのように表現するかはあなた次第です。
[[ コメントから: ]]
メールボックス1...mailbox10のような10個のメールボックスがあり、スレッドプールから5つのスレッドがあるため、スレッド1はメールボックスをフェッチするため、メールボックス1を選択すると仮定し、スレッド2はメールボックス2を選択し、スレッド3はメールボックス3を選択し、スレッド4は選択しますメールボックス 4 とスレッド 5 はメールボックス 5 を選択し、スレッド 1 (既に定義されている特定の期間) が解放されると、メールボックス 6 - メールボックス 10 を選択する必要があります。まだ読んでいないすべてのメールボックス。
ああなるほど。1 つの解決策は、メールボックスにメールがあるかどうかを確認するために、頻繁にスリープおよびウェイクアップするディスパッチ スレッドを用意することです。その場合、そのメールボックスが読み取られるようにジョブをスレッドプールにサブジェクトします。メールボックスが読み取られると、スレッドは戻り、次のメールボックスの処理を要求します。ディスパッチ スレッドは、停止するように指示されるまで、メールボックスをスレッド プールに追加し続けます。
に多くのコンテキストがある場合EmailProcessor、スレッドプールを使用できますが、BlockingQueue<File>どのメールボックスに注意が必要かを伝える何かから消費する可能性があります。
ここではエグゼキューターは最適なソリューションではないかもしれませんが、簡単にするためにグレイのコードのバリエーションを使用します。10 個のメールボックスを継続的にスキャンするには、次の手順を実行できますが、クリーンな終了を処理するコードを追加する必要があります。
// Create one semaphore per mailbox
Semaphore semaphores[] = new Semaphore[10]
for (int s = 0; s < semaphores.length; s ++) {
semaphores[s] = new Semaphore(1);
}
// create a thread pool with 5 workers
ExecutorService threadPool = Executors.newFixedThreadPool(5);
// submit all 10 user email accounts to the pool to be processed in turn
for (int i = 0; i < 5; i ++) {
threadPool.submit(userEmailsToProcess, semaphores);
}
// once we have submitted all jobs to the thread pool, it should be shutdown
threadPool.shutdown();
...
// here's our runnable class which does the actual work
public class EmailProcessor implements Runnable {
private UserEmail userEmailToProcess[];
private Semaphore semaphores[];
public MyJobProcessor(UserEmail userEmailToProcess[], Semaphore semaphores[]) {
this.userEmailsToProcess = userEmailToProcess;
this.semaphores = semaphores;
}
public void run() {
while (true) { // you could use a semaphore here to test program termination instead
for (int s = 0; s < semaphores.size; s ++) {
if (semaphores[s].tryAcquire()) {
UserEmail email = userEmailToProcess[s];
// read the user email
…
semaphores[s].release();
}
}
}
}
}
これは 100% 公正ではない簡単で汚いソリューションですが、任意の数のスレッドとメールボックスで機能します。10 個の電子メールと 5 個のワーカーを使用する特殊なケースでは、各スレッドにメールボックスのサブセットを継続的にスキャンさせることができます。競合はありません
既読の電子メール アカウントを追跡します。たとえば、次のように定義します。
//total number of email accounts that need to be read.
private int noOfEmails=10;
//the thread pool that is used to read the emails
private ExecutorService threadPool = Executors.newFixedThreadPool(5);
//the tracker array that keeps track of the emails that
//are already read. This array is cleared only after all 10
//emails are read.
private ArrayList<String> emailTracker=new ArrayList<String>(noOfEmails);
//any changes to emailTracker should be synchronized as
//this is a common data shared between all 5 threads.
private Object syncObject=new Object();
Runnable 実装では、emailTracker にメール アカウント識別子が含まれているかどうかを確認します。含まれている場合は、既に読み取られていることを意味するため、戻って emailTracker がクリアされるまで待ちます。10 個のメール アカウントがすべて読み取られると、クリアされます。
if(emailTracker.contains(email.identifier))
{
return;
}
//read email.
email.read();
//simple synchronization.
synchronized (syncObject)
{
//read email
emailTracker.add(email.identifier);
//if all emails are read, clear the tracker
//This will allow reading of all the emails
//once again.
if(emailTracker.size()==noOfEmails)
{
emailTracker.clear();
}
}
タスクの同時実行 (または同期、この場合はあまり重要ではない) コレクション (おそらくキュー) を作成するだけではどうですか。各タスクは、電子メール アカウントから必要なデータを読み込みます。次に、コレクションが空になるまで、各スレッドにコレクションからタスクを取得させます。
各スレッドは、このコレクションへの参照を持ち、ループを実行します。コレクションが空でない間に、そこからタスクを取得して処理します。
Sanju 私はあなたが次のラウンドロビン方式でそれらを実行したいと思っていると思います:
スレッド 1: 1,6 など スレッド 2: 2,7 スレッド 3: 3,8 スレッド 4: 4,9 スレッド 5: 5,10
まず第一に、スレッドをそのようなシーケンスで実行したい場合、これはスレッドの意図したものではありません。次に、スレッドプールでそれが可能だとは思いません。それでも必要な場合は、必要に応じて変更できるラウンドロビン方式でスレッドを実行するための私のソリューションを次に示します。
private static class EmailProcessor implements Runnable {
private final Object currentLock;
private final Object nextLock;
private UserEmail userEmail;
public EmailProcessor (UserEmail userEmail,, Object currentLock, Object nextLock) {
this.userEmail = userEmail;
this.currentLock = currentLock;
this.nextLock = nextLock;
}
@Override
public void run() {
try {
work = //reading email
while ( work != null) {
try {
currentLock.wait();
// Do your work here.
}
catch(InterruptedException e) {}
synchronized(nextLock) {
nextLock.notify();
}
}//while ends
} catch (IOException e) {
e.printStackTrace();
}
synchronized(nextLock) {
nextLock.notify(); /// Ensures all threads exit at the end
}
}
public EmailRoundRobin(int numberOfAccountsToRead) {
locks = new Object[numberOfAccountsToRead];
//Initialize lock instances in array.
for(i = 0; i < numberOfAccountsToRead; ++i) locks[i] = new Object();
//Create threads
int j;
for(j=0; j<(numberOfAccountsToRead-1); j++ ){
Thread linePrinterThread = new Thread(new EmailProcessor(emailInfo + "Temp" + j,locks[j],locks[j+1]));
linePrinterThread.start();
}
Thread lastLinePrinterThread = new Thread(new EmailProcessor(emailInfo + "Temp" + j,locks[numberOfFilesToRead-1],locks[0]));
lastLinePrinterThread.start();
}
public void startProcessing() {
synchronized (locks[0]) {
locks[0].notify();
}
}
public static void main(String[] args) {
EmailRoundRobin emailRoundRobin = new EmailRoundRobin(4);
emailRoundRobin.startPrinting();
}
}
[この質問] ( Java でラウンド ロビン方式でスレッドを実行する) に対する回答からこれを取り上げました。同様の要件がありました。その回答にも記載されている Phaser を使用するオプションがあります。
私は自分自身で答えを得ました。それは非常に簡単で、ExecutorService を使用し、それによってのみ管理されていました。