使用 AtomicBoolean 创建 Happens Before 关系
Creating a Happens Before Relationship with AtomicBoolean
阅读此代码 AsyncSubscriber.java:
编码器使用 AtomicBoolean 创建一个 Happens Before 关系,我想知道:
1_是否等同于使用同步块?
看起来这些线
if (on.get())
不确保块
try {
final Signal s = inboundSignals.poll(); // We take a signal off the queue
if (!done) { // If we're done, we shouldn't process any more signals, obeying rule 2.8
// Below we simply unpack the `Signal`s and invoke the corresponding methods
if (s instanceof OnNext<?>)
handleOnNext(((OnNext<T>)s).next);
else if (s instanceof OnSubscribe)
handleOnSubscribe(((OnSubscribe)s).subscription);
else if (s instanceof OnError) // We are always able to handle OnError, obeying rule 2.10
handleOnError(((OnError)s).error);
else if (s == OnComplete.Instance) // We are always able to handle OnComplete, obeying rule 2.9
handleOnComplete();
}
}
一次会被1个线程执行。
的确,当 on.get()
return 为真时,是什么阻止了另一个线程进入临界区?!
2_ 它比同步块更有效吗? (假设 AtomicBoolean 使用 Volatile
变量)
这里是部分代码:
// We are using this `AtomicBoolean` to make sure that this `Subscriber` doesn't run concurrently with itself,
// obeying rule 2.7 and 2.11
private final AtomicBoolean on = new AtomicBoolean(false);
@SuppressWarnings("unchecked")
@Override public final void run() {
if(on.get()) { // establishes a happens-before relationship with the end of the previous run
try {
final Signal s = inboundSignals.poll(); // We take a signal off the queue
if (!done) { // If we're done, we shouldn't process any more signals, obeying rule 2.8
// Below we simply unpack the `Signal`s and invoke the corresponding methods
if (s instanceof OnNext<?>)
handleOnNext(((OnNext<T>)s).next);
else if (s instanceof OnSubscribe)
handleOnSubscribe(((OnSubscribe)s).subscription);
else if (s instanceof OnError) // We are always able to handle OnError, obeying rule 2.10
handleOnError(((OnError)s).error);
else if (s == OnComplete.Instance) // We are always able to handle OnComplete, obeying rule 2.9
handleOnComplete();
}
} finally {
on.set(false); // establishes a happens-before relationship with the beginning of the next run
if(!inboundSignals.isEmpty()) // If we still have signals to process
tryScheduleToExecute(); // Then we try to schedule ourselves to execute again
}
}
}
// What `signal` does is that it sends signals to the `Subscription` asynchronously
private void signal(final Signal signal) {
if (inboundSignals.offer(signal)) // No need to null-check here as ConcurrentLinkedQueue does this for us
tryScheduleToExecute(); // Then we try to schedule it for execution, if it isn't already
}
// This method makes sure that this `Subscriber` is only executing on one Thread at a time
private final void tryScheduleToExecute() {
if(on.compareAndSet(false, true)) {
try {
executor.execute(this);
} catch(Throwable t) { // If we can't run on the `Executor`, we need to fail gracefully and not violate rule 2.13
if (!done) {
try {
done(); // First of all, this failure is not recoverable, so we need to cancel our subscription
} finally {
inboundSignals.clear(); // We're not going to need these anymore
// This subscription is cancelled by now, but letting the Subscriber become schedulable again means
// that we can drain the inboundSignals queue if anything arrives after clearing
on.set(false);
}
}
}
}
3_安全吗?
4_是否常用于此目的(创建关系之前发生的事情)?
是的,write/read 到 AtomicBolean 建立了一个 happens before 关系:
compareAndSet and all other read-and-update operations such as
getAndIncrement have the memory effects of both reading and writing
volatile variables.
由于您没有 post 整个代码,而且我们不知道它究竟是如何使用的,所以很难说它是否线程安全,但是:
ad 1. 它不等同于同步块 - 线程不等待
广告 2。是的,它可能更有效,但 compareAndSwap 没有义务由 volatile
变量支持 - 这是实施的数据。
ad 3. 很难说,但是 run
是一个 public 方法这一事实暴露了一些错误的可能性,例如,如果两个线程将直接调用 run
时 go
将具有 true
的值。从我的角度来看,直接在 run
方法中进行 compareAndSwap 会更好,但我不知道所有的要求,所以这只是一个建议。
ad 4. 是的,常用的是 AtomicBoolean。
阅读此代码 AsyncSubscriber.java: 编码器使用 AtomicBoolean 创建一个 Happens Before 关系,我想知道:
1_是否等同于使用同步块?
看起来这些线
if (on.get())
不确保块
try {
final Signal s = inboundSignals.poll(); // We take a signal off the queue
if (!done) { // If we're done, we shouldn't process any more signals, obeying rule 2.8
// Below we simply unpack the `Signal`s and invoke the corresponding methods
if (s instanceof OnNext<?>)
handleOnNext(((OnNext<T>)s).next);
else if (s instanceof OnSubscribe)
handleOnSubscribe(((OnSubscribe)s).subscription);
else if (s instanceof OnError) // We are always able to handle OnError, obeying rule 2.10
handleOnError(((OnError)s).error);
else if (s == OnComplete.Instance) // We are always able to handle OnComplete, obeying rule 2.9
handleOnComplete();
}
}
一次会被1个线程执行。
的确,当 on.get()
return 为真时,是什么阻止了另一个线程进入临界区?!
2_ 它比同步块更有效吗? (假设 AtomicBoolean 使用 Volatile
变量)
这里是部分代码:
// We are using this `AtomicBoolean` to make sure that this `Subscriber` doesn't run concurrently with itself,
// obeying rule 2.7 and 2.11
private final AtomicBoolean on = new AtomicBoolean(false);
@SuppressWarnings("unchecked")
@Override public final void run() {
if(on.get()) { // establishes a happens-before relationship with the end of the previous run
try {
final Signal s = inboundSignals.poll(); // We take a signal off the queue
if (!done) { // If we're done, we shouldn't process any more signals, obeying rule 2.8
// Below we simply unpack the `Signal`s and invoke the corresponding methods
if (s instanceof OnNext<?>)
handleOnNext(((OnNext<T>)s).next);
else if (s instanceof OnSubscribe)
handleOnSubscribe(((OnSubscribe)s).subscription);
else if (s instanceof OnError) // We are always able to handle OnError, obeying rule 2.10
handleOnError(((OnError)s).error);
else if (s == OnComplete.Instance) // We are always able to handle OnComplete, obeying rule 2.9
handleOnComplete();
}
} finally {
on.set(false); // establishes a happens-before relationship with the beginning of the next run
if(!inboundSignals.isEmpty()) // If we still have signals to process
tryScheduleToExecute(); // Then we try to schedule ourselves to execute again
}
}
}
// What `signal` does is that it sends signals to the `Subscription` asynchronously
private void signal(final Signal signal) {
if (inboundSignals.offer(signal)) // No need to null-check here as ConcurrentLinkedQueue does this for us
tryScheduleToExecute(); // Then we try to schedule it for execution, if it isn't already
}
// This method makes sure that this `Subscriber` is only executing on one Thread at a time
private final void tryScheduleToExecute() {
if(on.compareAndSet(false, true)) {
try {
executor.execute(this);
} catch(Throwable t) { // If we can't run on the `Executor`, we need to fail gracefully and not violate rule 2.13
if (!done) {
try {
done(); // First of all, this failure is not recoverable, so we need to cancel our subscription
} finally {
inboundSignals.clear(); // We're not going to need these anymore
// This subscription is cancelled by now, but letting the Subscriber become schedulable again means
// that we can drain the inboundSignals queue if anything arrives after clearing
on.set(false);
}
}
}
}
3_安全吗?
4_是否常用于此目的(创建关系之前发生的事情)?
是的,write/read 到 AtomicBolean 建立了一个 happens before 关系:
compareAndSet and all other read-and-update operations such as getAndIncrement have the memory effects of both reading and writing volatile variables.
由于您没有 post 整个代码,而且我们不知道它究竟是如何使用的,所以很难说它是否线程安全,但是:
ad 1. 它不等同于同步块 - 线程不等待
广告 2。是的,它可能更有效,但 compareAndSwap 没有义务由 volatile
变量支持 - 这是实施的数据。
ad 3. 很难说,但是 run
是一个 public 方法这一事实暴露了一些错误的可能性,例如,如果两个线程将直接调用 run
时 go
将具有 true
的值。从我的角度来看,直接在 run
方法中进行 compareAndSwap 会更好,但我不知道所有的要求,所以这只是一个建议。
ad 4. 是的,常用的是 AtomicBoolean。