3

たくさんのスレッドが同時に実行されています。スレッドが他のスレッドに通知して、ジョブが終了するのを待ち、再開するように再度通知する必要がある場合があります。私はJavaの同期に何とか慣れていないので、そのようなことを行う正しい方法は何でしょうか。私のコードは次のようなものです：

private void Concurrent() {
    if (shouldRun()) {
        // notify threads to pause and wait for them
        DoJob();
        // resume threads
    }

    // Normal job...
}

アップデート：

私が書いたコードは、各スレッドによって実行されるクラス内にあることに注意してください。私はそれらのスレッドやそれらがどのように実行されているかにアクセスできません。私はただスレッドの中にいます。

アップデート2：

私のコードはクローラークラスからのものです。クローラークラス（crawler4j）は、並行性を処理する方法を知っています。必要なのは、関数を実行する前に他のクローラーを一時停止し、後で再開することだけです。このコードは私のクローラーの基本です：

   public class TestCrawler extends WebCrawler {
    private SingleThread()
    {
        //When this function is running, no other crawler should do anything
    }

    @Override
    public void visit(Page page) {
        if(SomeCriteria())
        {
            //make all other crawlers stop until I finish
            SingleThread();
            //let them resume
        }

        //Normal Stuff
    }
   }
4

3 に答える 3

2

これは、クールなJava同時実行機能を使用してこれを実現する方法の簡単な例です。

古いコードを切り取るのは、Pauseクラスではもう重要ではありません。

編集：

新しいテストクラスは次のとおりです。

package de.hotware.test;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Test {

    private Pause mPause;

    public Test() {
        this.mPause = new Pause();
    }

    public void concurrent() throws InterruptedException {
        while(true) {
            this.mPause.probe();
            System.out.println("concurrent");
            Thread.sleep(100);
        }
    }

    public void crucial() throws InterruptedException {
        int i = 0;
        while (true) {
            if (i++ % 2 == 0) {
                this.mPause.pause(true);
                System.out.println("crucial: exclusive execution");
                this.mPause.pause(false);
            } else {
                System.out.println("crucial: normal execution");
                Thread.sleep(1000);
            }
        }
    }

    public static void main(String[] args) {
        final Test test = new Test();
        Runnable run = new Runnable() {

            @Override
            public void run() {
                try {
                    test.concurrent();
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }

        };
        Runnable cruc = new Runnable() {

            @Override
            public void run() {
                try {
                    test.crucial();
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }

        };
        ExecutorService serv = Executors.newCachedThreadPool();
        serv.execute(run);
        serv.execute(run);
        serv.execute(cruc);
    }

}

そして、ユーティリティPauseクラス：

package de.hotware.test;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Utility class to pause and unpause threads
 * with Java Concurrency
 * @author Martin Braun
 */
public class Pause {

    private Lock mLock;
    private Condition mCondition;
    private AtomicBoolean mAwait;

    public Pause() {
        this.mLock = new ReentrantLock();
        this.mCondition = this.mLock.newCondition();
        this.mAwait = new AtomicBoolean(false);
    }

    /**
     * waits until the threads until this.mAwait is set to true
     * @throws InterruptedException
     */
    public void probe() throws InterruptedException {
        while(this.mAwait.get()) {
            this.mLock.lock();
            try {
                this.mCondition.await();
            } finally {
                this.mLock.unlock();
            }
        }
    }

    /**
     * pauses or unpauses
     */
    public void pause(boolean pValue) {
        if(!pValue){
            this.mLock.lock();
            try {
                this.mCondition.signalAll();
            } finally {
                this.mLock.unlock();
            }
        }
        this.mAwait.set(pValue);
    }

}

基本的な使用法は、各実行の前にprobe（）を呼び出すことです。これは、pause（false）が呼び出されるまで一時停止されている場合にブロックされます。

クラスは次のようになります。

public class TestCrawler extends WebCrawler {

private Pause mPause;

public TestCrawler(Pause pPause) {
    this.mPause = pPause;
}

private SingleThread()
{
        //When this function is running, no other crawler should do anything
}

@Override
public void visit(Page page) {
    if(SomeCriteria())
    {
        //only enter the crucial part once if it has to be exclusive
        this.mPause.probe();
        //make all other crawlers stop until I finish
        this.mPause.pause(true);
        SingleThread();
        //let them resume
        this.mPause.pause(false);
    }
    this.mPause.probe();
    //Normal Stuff
}
}
于 2013-01-13T14:11:49.510 に答える
1 
public class StockMonitor extends Thread {

    private boolean suspend = false; 
    private volatile Thread thread;

    public StockMonitor() {
        thread = this;
    }

    // Use name with underscore, in order to avoid naming crashing with
    // Thread's.
    private synchronized void _wait() throws InterruptedException {
        while (suspend) {
            wait();
        }
    }

    // Use name with underscore, in order to avoid naming crashing with
    // Thread's.
    public synchronized void _resume() {
        suspend = false;
        notify();
    }

    // Use name with underscore, in order to avoid naming crashing with
    // Thread's.
    public synchronized void _suspend() {
        suspend = true;
    }  

     public void _stop() { 
        thread = null;
        // Wake up from sleep.
        interrupt();     
     }

     @Override
     public void run() {
        final Thread thisThread = Thread.currentThread();
        while (thisThread == thread) {
            _wait();
            // Do whatever you want right here.
        }
     }
}

を呼び出す_resumeと、_suspendを再開して一時停止できるようになりますThread。_stopスレッドを正常に停止できます。Threadを停止すると、再開する方法がないことに注意してください。はThread使用できなくなりました。

コードは実際のオープンソースプロジェクトから選択されています：http：//jstock.hg.sourceforge.net/hgweb/jstock/jstock/file/b17c0fbfe37c/src/org/yccheok/jstock/engine/RealTimeStockMonitor.java#l247

于 2013-01-13T13:52:23.783 に答える
0

wait（）とnotify（）を使用できます

スレッド待機中：

  // define mutex as field
  Object mutex = new Object();

   // later:
   synchronized(mutex) {
        wait();
   }

続行するようにスレッドに通知する

   synchronized (mutex) {
        notify();
   }
于 2013-01-13T13:49:28.683 に答える