Fix ConsumeDriver running status (#748)
This commit is contained in:
parent
35495375d0
commit
affbaa8bfd
|
|
@ -28,6 +28,7 @@ Release Notes.
|
||||||
* Add Caffeine plugin as optional.
|
* Add Caffeine plugin as optional.
|
||||||
* Add Undertow 2.1.7.final+ worker thread pool metrics.
|
* Add Undertow 2.1.7.final+ worker thread pool metrics.
|
||||||
* Support for tracking in spring gateway versions 4.1.2 and above.
|
* Support for tracking in spring gateway versions 4.1.2 and above.
|
||||||
|
* Fix `ConsumeDriver` running status concurrency issues.
|
||||||
|
|
||||||
All issues and pull requests are [here](https://github.com/apache/skywalking/milestone/222?closed=1)
|
All issues and pull requests are [here](https://github.com/apache/skywalking/milestone/222?closed=1)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -27,7 +27,7 @@ import org.apache.skywalking.apm.commons.datacarrier.buffer.Channels;
|
||||||
* Pool of consumers <p> Created by wusheng on 2016/10/25.
|
* Pool of consumers <p> Created by wusheng on 2016/10/25.
|
||||||
*/
|
*/
|
||||||
public class ConsumeDriver<T> implements IDriver {
|
public class ConsumeDriver<T> implements IDriver {
|
||||||
private boolean running;
|
private volatile boolean running;
|
||||||
private ConsumerThread[] consumerThreads;
|
private ConsumerThread[] consumerThreads;
|
||||||
private Channels<T> channels;
|
private Channels<T> channels;
|
||||||
private ReentrantLock lock;
|
private ReentrantLock lock;
|
||||||
|
|
@ -88,6 +88,9 @@ public class ConsumeDriver<T> implements IDriver {
|
||||||
}
|
}
|
||||||
lock.lock();
|
lock.lock();
|
||||||
try {
|
try {
|
||||||
|
if (running) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
this.allocateBuffer2Thread();
|
this.allocateBuffer2Thread();
|
||||||
for (ConsumerThread consumerThread : consumerThreads) {
|
for (ConsumerThread consumerThread : consumerThreads) {
|
||||||
consumerThread.start();
|
consumerThread.start();
|
||||||
|
|
@ -124,8 +127,14 @@ public class ConsumeDriver<T> implements IDriver {
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void close(Channels channels) {
|
public void close(Channels channels) {
|
||||||
|
if (!running) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
lock.lock();
|
lock.lock();
|
||||||
try {
|
try {
|
||||||
|
if (!running) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
this.running = false;
|
this.running = false;
|
||||||
for (ConsumerThread consumerThread : consumerThreads) {
|
for (ConsumerThread consumerThread : consumerThreads) {
|
||||||
consumerThread.shutdown();
|
consumerThread.shutdown();
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue