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

merging revisions 817222:817842 

git-svn-id: https://svn.apache.org/repos/asf/activemq/branches/activemq-5.3@818540 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Bosanac Dejan 2009-09-24 16:32:07 +00:00
parent 14b983cab1
commit 1c121bcded
3 changed files with 43 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java
View File

@ -51,6 +51,7 @@ public abstract class BaseDestination implements Destination {
protected SystemUsage systemUsage; protected SystemUsage systemUsage;
protected MemoryUsage memoryUsage; protected MemoryUsage memoryUsage;
private boolean producerFlowControl = true; private boolean producerFlowControl = true;
protected boolean warnOnProducerFlowControl = true;
private int maxProducersToAudit = 1024; private int maxProducersToAudit = 1024;
private int maxAuditDepth = 2048; private int maxAuditDepth = 2048;
private boolean enableAudit = true; private boolean enableAudit = true;

26
activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
View File

@ -413,8 +413,15 @@ public class Queue extends BaseDestination implements Task, UsageListener {
isFull(context, memoryUsage); isFull(context, memoryUsage);
fastProducer(context, producerInfo); fastProducer(context, producerInfo);
if (isProducerFlowControl() && context.isProducerFlowControl()) { if (isProducerFlowControl() && context.isProducerFlowControl()) {
if(warnOnProducerFlowControl) {
warnOnProducerFlowControl = false;
LOG.info("Usage Manager memory limit reached on " +getActiveMQDestination().getQualifiedName() + ". Producers will be throttled to the rate at which messages are removed from this destination to prevent flooding it." +
" See http://activemq.apache.org/producer-flow-control.html for more info");
}
if (systemUsage.isSendFailIfNoSpace()) { if (systemUsage.isSendFailIfNoSpace()) {
throw new javax.jms.ResourceAllocationException("SystemUsage memory limit reached"); throw new javax.jms.ResourceAllocationException("Usage Manager memory limit reached. Stopping producer (" + message.getProducerId() + ") to prevent flooding " +getActiveMQDestination().getQualifiedName() + "." +
" See http://activemq.apache.org/producer-flow-control.html for more info");
} }
// We can avoid blocking due to low usage if the producer is sending // We can avoid blocking due to low usage if the producer is sending
@ -498,8 +505,13 @@ public class Queue extends BaseDestination implements Task, UsageListener {
final ConnectionContext context = producerExchange.getConnectionContext(); final ConnectionContext context = producerExchange.getConnectionContext();
synchronized (sendLock) { synchronized (sendLock) {
if (store != null && message.isPersistent()) { if (store != null && message.isPersistent()) {
if (systemUsage.isSendFailIfNoSpace() && systemUsage.getStoreUsage().isFull()) { if (systemUsage.getStoreUsage().isFull()) {
throw new javax.jms.ResourceAllocationException("Usage Manager Store is Full"); final String logMessage = "Usage Manager Store is Full. Stopping producer (" + message.getProducerId() + ") to prevent flooding " + getActiveMQDestination().getQualifiedName() + "." +
" See http://activemq.apache.org/producer-flow-control.html for more info";
LOG.info(logMessage);
if (systemUsage.isSendFailIfNoSpace()) {
throw new javax.jms.ResourceAllocationException(logMessage);
}
} }
while (!systemUsage.getStoreUsage().waitForSpace(1000)) { while (!systemUsage.getStoreUsage().waitForSpace(1000)) {
if (context.getStopping().get()) { if (context.getStopping().get()) {
@ -1279,6 +1291,14 @@ public class Queue extends BaseDestination implements Task, UsageListener {
final void sendMessage(final ConnectionContext context, Message msg) throws Exception { final void sendMessage(final ConnectionContext context, Message msg) throws Exception {
if (!msg.isPersistent() && messages.getSystemUsage() != null) { if (!msg.isPersistent() && messages.getSystemUsage() != null) {
if (systemUsage.getTempUsage().isFull()) {
final String logMessage = "Usage Manager Temp Store is Full. Stopping producer (" + msg.getProducerId() + ") to prevent flooding " + getActiveMQDestination().getQualifiedName() + "." +
" See http://activemq.apache.org/producer-flow-control.html for more info";
LOG.info(logMessage);
if (systemUsage.isSendFailIfNoSpace()) {
throw new javax.jms.ResourceAllocationException(logMessage);
}
}
messages.getSystemUsage().getTempUsage().waitForSpace(); messages.getSystemUsage().getTempUsage().waitForSpace();
} }
synchronized(messages) { synchronized(messages) {

24
activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
View File

@ -280,11 +280,20 @@ public class Topic extends BaseDestination implements Task{
if(memoryUsage.isFull()) { if(memoryUsage.isFull()) {
isFull(context, memoryUsage); isFull(context, memoryUsage);
fastProducer(context, producerInfo); fastProducer(context, producerInfo);
if (isProducerFlowControl() && context.isProducerFlowControl()) { if (isProducerFlowControl() && context.isProducerFlowControl()) {
if (systemUsage.isSendFailIfNoSpace()) {
throw new javax.jms.ResourceAllocationException("Usage Manager memory limit reached"); if(warnOnProducerFlowControl) {
warnOnProducerFlowControl = false;
LOG.info("Usage Manager memory limit reached for " +getActiveMQDestination().getQualifiedName() + ". Producers will be throttled to the rate at which messages are removed from this destination to prevent flooding it." +
" See http://activemq.apache.org/producer-flow-control.html for more info");
} }
if (systemUsage.isSendFailIfNoSpace()) {
throw new javax.jms.ResourceAllocationException("Usage Manager memory limit reached. Stopping producer (" + message.getProducerId() + ") to prevent flooding " +getActiveMQDestination().getQualifiedName() + "." +
" See http://activemq.apache.org/producer-flow-control.html for more info");
}
// We can avoid blocking due to low usage if the producer is sending // We can avoid blocking due to low usage if the producer is sending
// a sync message or // a sync message or
// if it is using a producer window // if it is using a producer window
@ -390,8 +399,13 @@ public class Topic extends BaseDestination implements Task{
if (topicStore != null && message.isPersistent() if (topicStore != null && message.isPersistent()
&& !canOptimizeOutPersistence()) { && !canOptimizeOutPersistence()) {
if (systemUsage.isSendFailIfNoSpace() && systemUsage.getStoreUsage().isFull()) { if (systemUsage.getStoreUsage().isFull()) {
throw new javax.jms.ResourceAllocationException("Usage Manager Store is Full"); final String logMessage = "Usage Manager Store is Full. Stopping producer (" + message.getProducerId() + ") to prevent flooding " + getActiveMQDestination().getQualifiedName() + "." +
" See http://activemq.apache.org/producer-flow-control.html for more info";
LOG.info(logMessage);
if (systemUsage.isSendFailIfNoSpace()) {
throw new javax.jms.ResourceAllocationException(logMessage);
}
} }
while (!systemUsage.getStoreUsage().waitForSpace(1000)) { while (!systemUsage.getStoreUsage().waitForSpace(1000)) {
if (context.getStopping().get()) { if (context.getStopping().get()) {