Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

https://issues.apache.org/jira/browse/AMQ-3272 - prevent RejectedExecutionException in kahadb 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1090186 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Bosanac Dejan 2011-04-08 10:22:40 +00:00
parent 4ab47eb9aa
commit 15de238fd6
1 changed files with 28 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
View File

@ -28,14 +28,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.activemq.broker.ConnectionContext;
@ -182,7 +175,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter {
this.globalTopicSemaphore = new Semaphore(getMaxAsyncJobs());
this.asyncQueueJobQueue = new LinkedBlockingQueue<Runnable>(getMaxAsyncJobs());
this.asyncTopicJobQueue = new LinkedBlockingQueue<Runnable>(getMaxAsyncJobs());
this.queueExecutor = new ThreadPoolExecutor(1, asyncExecutorMaxThreads, 0L, TimeUnit.MILLISECONDS,
this.queueExecutor = new StoreTaskExecutor(1, asyncExecutorMaxThreads, 0L, TimeUnit.MILLISECONDS,
asyncQueueJobQueue, new ThreadFactory() {
public Thread newThread(Runnable runnable) {
Thread thread = new Thread(runnable, "ConcurrentQueueStoreAndDispatch");
@ -190,7 +183,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter {
return thread;
}
});
this.topicExecutor = new ThreadPoolExecutor(1, asyncExecutorMaxThreads, 0L, TimeUnit.MILLISECONDS,
this.topicExecutor = new StoreTaskExecutor(1, asyncExecutorMaxThreads, 0L, TimeUnit.MILLISECONDS,
asyncTopicJobQueue, new ThreadFactory() {
public Thread newThread(Runnable runnable) {
Thread thread = new Thread(runnable, "ConcurrentTopicStoreAndDispatch");
@ -1040,8 +1033,12 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter {
}
}
interface StoreTask {
public interface StoreTask {
public boolean cancel();
public void aquireLocks();
public void releaseLocks();
}
class StoreQueueTask implements Runnable, StoreTask {
@ -1071,7 +1068,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter {
return false;
}
void aquireLocks() {
public void aquireLocks() {
if (this.locked.compareAndSet(false, true)) {
try {
globalQueueSemaphore.acquire();
@ -1084,7 +1081,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter {
}
void releaseLocks() {
public void releaseLocks() {
if (this.locked.compareAndSet(true, false)) {
store.releaseLocalAsyncLock();
globalQueueSemaphore.release();
@ -1106,8 +1103,6 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter {
}
} catch (Exception e) {
this.future.setException(e);
} finally {
releaseLocks();
}
}
@ -1145,7 +1140,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter {
}
@Override
void aquireLocks() {
public void aquireLocks() {
if (this.locked.compareAndSet(false, true)) {
try {
globalTopicSemaphore.acquire();
@ -1159,7 +1154,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter {
}
@Override
void releaseLocks() {
public void releaseLocks() {
if (this.locked.compareAndSet(true, false)) {
message.decrementReferenceCount();
store.releaseLocalAsyncLock();
@ -1202,9 +1197,23 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter {
}
} catch (Exception e) {
this.future.setException(e);
} finally {
releaseLocks();
}
}
}
public class StoreTaskExecutor extends ThreadPoolExecutor {
public StoreTaskExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit timeUnit, BlockingQueue<Runnable> queue, ThreadFactory threadFactory) {
super(corePoolSize, maximumPoolSize, keepAliveTime, timeUnit, queue, threadFactory);
}
protected void afterExecute(Runnable runnable, Throwable throwable) {
super.afterExecute(runnable, throwable);
if (runnable instanceof StoreTask) {
((StoreTask)runnable).releaseLocks();
}
}
}
}