notify() 和 wait() 在 Java 中不工作
notify() and wait() not working in Java
我有 2 个线程要与 wait() 和 notify() 同步。但是,当我通知等待的线程永远不会恢复时。这是我的代码片段。
在 Lib60870 中,我启动了两个线程,线程 HandShake 与 SerialReader 同步。
public Lib60870(){ //Here I start threads
try {
myConnection=new Connection(LOCALHOST,port);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
mySerialReader.start();
myHandshake.start();}
}
Class SerialReader
public class SerialReader extends Thread {
private static boolean isPaused=true;
@Override
public void run() {
synchronized(this){
if(Lib60870.myConnection!=null){
while(true){
if(!isPaused){
byte inByte=Lib60870.myConnection.getByte();
if(inByte==0x68){
...
}
notify();
}
else if(inByte==0x10){
...
}
notify();
}
}
}
}
}
}
public void setPause(boolean pause){
isPaused=pause;
}
Class握手
public class HandShake extends Thread {
public void run() {
synchronized(Lib60870.mySerialReader){
Lib60870.mySerialReader.setPause(false);
...
try {
Lib60870.mySerialReader.wait();
} catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
Lib60870.mySerialReader.setPause(true);
...
Lib60870.mySerialReader.setPause(false);
try {
Lib60870.mySerialReader.wait();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}
提前致谢
你的方法有很多问题:
- 扩展
Thread
被认为是不好的做法。
- 将
wait
/notify
用于 java.util.concurrent
中可以完成的事情不是一个好主意。
- 陷入一个紧密旋转的循环并不是暂停。
这是一个可暂停的线程 class。给自己写一个 Stepper
对象,并使用其中一个对象在不暂停的情况下连续执行 step()
方法。使用其 pause
/resume
方法干净地暂停它。
/**
* PauseableThread is a Thread with pause/resume and cancel methods.
*
* The meat of the process must implement `step`.
*
* You can either extend this and implement `step` or use the factory.
*
* I cannot extend Thread because my resume will clash.
*
*/
public abstract class PauseableThread implements Runnable {
// The lock.
private final ReadWriteLock pause = new ReentrantReadWriteLock();
private final Lock readLock = pause.readLock();
private final Lock writeLock = pause.writeLock();
// Flag to cancel the whole process.
private volatile boolean cancelled = false;
// The exception that cause it to finish.
private Exception thrown = null;
// The thread that is me.
private Thread me = null;
@Override
// The core run mechanism.
public void run() {
// Track my current thread.
me = Thread.currentThread();
try {
while (!finished()) {
// Block here if we're paused.
blockIfPaused();
// Don't do any more work if we've been asked to stop.
if (!finished()) {
// Do my work.
step();
}
}
} catch (Exception ex) {
// Just fall out when exception is thrown.
thrown = ex;
}
}
// Have we finished yet?
private boolean finished() {
return cancelled || me.isInterrupted();
}
// Block if pause has been called without a matching resume.
private void blockIfPaused() throws InterruptedException {
try {
// Grab a write lock. Will block if a read lock has been taken - i.e. we've been paused.
writeLock.lockInterruptibly();
} finally {
// Release the lock immediately to avoid blocking when pause is called.
writeLock.unlock();
}
}
// Pause the work. NB: MUST be balanced by a resume.
public void pause() {
// We can wait for a lock here.
readLock.lock();
}
// Resume the work. NB: MUST be balanced by a pause.
public void resume() {
// Release the lock.
readLock.unlock();
}
// Stop.
public void cancel() {
// Stop everything.
cancelled = true;
}
// Stop immediately (if param is true).
public void cancel(boolean interrupt) {
if (interrupt) {
// Interrupt me.
me.interrupt();
} else {
// Or cancel me.
cancel();
}
}
// Wait for completion.
public void await() throws InterruptedException {
// Wait 'till we've finished. NB: Will wait forever if you haven't instigated a cancel of some kind.
while (me.isAlive()) {
Thread.sleep(0);
}
}
// Start - like a thread.
public void start() {
// Wrap me in a thread and fire the sucker up!
new Thread(this).start();
}
// Get the exception that was thrown to stop the thread or null if the thread was cancelled.
public Exception getThrown() {
return thrown;
}
// Expose my Thread.
public Thread getThread() {
return me;
}
// Create this method to do stuff.
// Calls to this method will stop when pause is called.
// Any thrown exception stops the whole process.
public abstract void step() throws Exception;
// Factory to wrap a Stepper in a PauseableThread
public static PauseableThread make(Stepper stepper) {
StepperThread pauseableStepper = new StepperThread(stepper);
// That's the thread they can pause/resume.
return pauseableStepper;
}
// One of these must be used.
public interface Stepper {
// A Stepper has a step method.
// Any exception thrown causes the enclosing thread to stop.
public void step() throws Exception;
}
// Holder for a Stepper.
private static class StepperThread extends PauseableThread {
// The actual stepper I am proxying.
private final Stepper stepper;
StepperThread(Stepper stepper) {
this.stepper = stepper;
}
@Override
public void step() throws Exception {
stepper.step();
}
}
// !!!! Testing only below !!!!
// My test counter.
static int n = 0;
// Test/demo.
public static void main(String[] args) throws InterruptedException {
try {
// Simple stepper that just increments n.
Stepper s = () -> {
n += 1;
Thread.sleep(1);
};
PauseableThread pt = PauseableThread.make(s);
// Start it up.
pt.start();
Thread.sleep(1000);
pt.pause();
System.out.println("Paused: " + n);
Thread.sleep(1000);
System.out.println("Resuminng: " + n);
pt.resume();
Thread.sleep(1000);
pt.cancel();
pt.await();
System.out.println("Finished: " + n);
// Start again to test agressive cancelling.
n = 0;
pt = PauseableThread.make(s);
// Start it up.
pt.start();
Thread.sleep(1000);
pt.pause();
System.out.println("Paused: " + n);
Thread.sleep(1000);
System.out.println("Resuminng: " + n);
pt.resume();
Thread.sleep(1000);
// Cancel aggressively.
pt.cancel(true);
pt.await();
System.out.println("Finished: " + n);
System.out.println("thrown: " + pt.getThrown());
} catch (InterruptedException e) {
}
}
}
您遇到的主要问题是只有一个线程可以持有锁。这意味着当您的 notify() 线程持有锁时,没有其他线程可以 运行ing 在持有该锁的代码块中。
将同步块移动到 if (isPaused)
块内,以便另一个线程可以 运行 在两者之间。
您遇到的另一个问题是您的 isPaused
布尔值不是可变的,因此它可以内联,即它可能永远不会停止。它不应该是静态的,即使你知道你永远不会拥有其中一个以上,对实例变量使用静态字段是不好的做法。
暂停时,您应该让 CPU 进入睡眠状态。
每当你 notify() 或 notifyAll() 这应该导致状态改变你的 wait() 应该总是检查那个状态改变。
我有 2 个线程要与 wait() 和 notify() 同步。但是,当我通知等待的线程永远不会恢复时。这是我的代码片段。 在 Lib60870 中,我启动了两个线程,线程 HandShake 与 SerialReader 同步。
public Lib60870(){ //Here I start threads
try {
myConnection=new Connection(LOCALHOST,port);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
mySerialReader.start();
myHandshake.start();}
}
Class SerialReader
public class SerialReader extends Thread {
private static boolean isPaused=true;
@Override
public void run() {
synchronized(this){
if(Lib60870.myConnection!=null){
while(true){
if(!isPaused){
byte inByte=Lib60870.myConnection.getByte();
if(inByte==0x68){
...
}
notify();
}
else if(inByte==0x10){
...
}
notify();
}
}
}
}
}
}
public void setPause(boolean pause){
isPaused=pause;
}
Class握手
public class HandShake extends Thread {
public void run() {
synchronized(Lib60870.mySerialReader){
Lib60870.mySerialReader.setPause(false);
...
try {
Lib60870.mySerialReader.wait();
} catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
Lib60870.mySerialReader.setPause(true);
...
Lib60870.mySerialReader.setPause(false);
try {
Lib60870.mySerialReader.wait();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}
提前致谢
你的方法有很多问题:
- 扩展
Thread
被认为是不好的做法。 - 将
wait
/notify
用于java.util.concurrent
中可以完成的事情不是一个好主意。 - 陷入一个紧密旋转的循环并不是暂停。
这是一个可暂停的线程 class。给自己写一个 Stepper
对象,并使用其中一个对象在不暂停的情况下连续执行 step()
方法。使用其 pause
/resume
方法干净地暂停它。
/**
* PauseableThread is a Thread with pause/resume and cancel methods.
*
* The meat of the process must implement `step`.
*
* You can either extend this and implement `step` or use the factory.
*
* I cannot extend Thread because my resume will clash.
*
*/
public abstract class PauseableThread implements Runnable {
// The lock.
private final ReadWriteLock pause = new ReentrantReadWriteLock();
private final Lock readLock = pause.readLock();
private final Lock writeLock = pause.writeLock();
// Flag to cancel the whole process.
private volatile boolean cancelled = false;
// The exception that cause it to finish.
private Exception thrown = null;
// The thread that is me.
private Thread me = null;
@Override
// The core run mechanism.
public void run() {
// Track my current thread.
me = Thread.currentThread();
try {
while (!finished()) {
// Block here if we're paused.
blockIfPaused();
// Don't do any more work if we've been asked to stop.
if (!finished()) {
// Do my work.
step();
}
}
} catch (Exception ex) {
// Just fall out when exception is thrown.
thrown = ex;
}
}
// Have we finished yet?
private boolean finished() {
return cancelled || me.isInterrupted();
}
// Block if pause has been called without a matching resume.
private void blockIfPaused() throws InterruptedException {
try {
// Grab a write lock. Will block if a read lock has been taken - i.e. we've been paused.
writeLock.lockInterruptibly();
} finally {
// Release the lock immediately to avoid blocking when pause is called.
writeLock.unlock();
}
}
// Pause the work. NB: MUST be balanced by a resume.
public void pause() {
// We can wait for a lock here.
readLock.lock();
}
// Resume the work. NB: MUST be balanced by a pause.
public void resume() {
// Release the lock.
readLock.unlock();
}
// Stop.
public void cancel() {
// Stop everything.
cancelled = true;
}
// Stop immediately (if param is true).
public void cancel(boolean interrupt) {
if (interrupt) {
// Interrupt me.
me.interrupt();
} else {
// Or cancel me.
cancel();
}
}
// Wait for completion.
public void await() throws InterruptedException {
// Wait 'till we've finished. NB: Will wait forever if you haven't instigated a cancel of some kind.
while (me.isAlive()) {
Thread.sleep(0);
}
}
// Start - like a thread.
public void start() {
// Wrap me in a thread and fire the sucker up!
new Thread(this).start();
}
// Get the exception that was thrown to stop the thread or null if the thread was cancelled.
public Exception getThrown() {
return thrown;
}
// Expose my Thread.
public Thread getThread() {
return me;
}
// Create this method to do stuff.
// Calls to this method will stop when pause is called.
// Any thrown exception stops the whole process.
public abstract void step() throws Exception;
// Factory to wrap a Stepper in a PauseableThread
public static PauseableThread make(Stepper stepper) {
StepperThread pauseableStepper = new StepperThread(stepper);
// That's the thread they can pause/resume.
return pauseableStepper;
}
// One of these must be used.
public interface Stepper {
// A Stepper has a step method.
// Any exception thrown causes the enclosing thread to stop.
public void step() throws Exception;
}
// Holder for a Stepper.
private static class StepperThread extends PauseableThread {
// The actual stepper I am proxying.
private final Stepper stepper;
StepperThread(Stepper stepper) {
this.stepper = stepper;
}
@Override
public void step() throws Exception {
stepper.step();
}
}
// !!!! Testing only below !!!!
// My test counter.
static int n = 0;
// Test/demo.
public static void main(String[] args) throws InterruptedException {
try {
// Simple stepper that just increments n.
Stepper s = () -> {
n += 1;
Thread.sleep(1);
};
PauseableThread pt = PauseableThread.make(s);
// Start it up.
pt.start();
Thread.sleep(1000);
pt.pause();
System.out.println("Paused: " + n);
Thread.sleep(1000);
System.out.println("Resuminng: " + n);
pt.resume();
Thread.sleep(1000);
pt.cancel();
pt.await();
System.out.println("Finished: " + n);
// Start again to test agressive cancelling.
n = 0;
pt = PauseableThread.make(s);
// Start it up.
pt.start();
Thread.sleep(1000);
pt.pause();
System.out.println("Paused: " + n);
Thread.sleep(1000);
System.out.println("Resuminng: " + n);
pt.resume();
Thread.sleep(1000);
// Cancel aggressively.
pt.cancel(true);
pt.await();
System.out.println("Finished: " + n);
System.out.println("thrown: " + pt.getThrown());
} catch (InterruptedException e) {
}
}
}
您遇到的主要问题是只有一个线程可以持有锁。这意味着当您的 notify() 线程持有锁时,没有其他线程可以 运行ing 在持有该锁的代码块中。
将同步块移动到 if (isPaused)
块内,以便另一个线程可以 运行 在两者之间。
您遇到的另一个问题是您的 isPaused
布尔值不是可变的,因此它可以内联,即它可能永远不会停止。它不应该是静态的,即使你知道你永远不会拥有其中一个以上,对实例变量使用静态字段是不好的做法。
暂停时,您应该让 CPU 进入睡眠状态。
每当你 notify() 或 notifyAll() 这应该导致状态改变你的 wait() 应该总是检查那个状态改变。