You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by js...@apache.org on 2006/03/03 14:56:30 UTC
svn commit: r382816 - in
/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker:
jmx/ region/ region/policy/
Author: jstrachan
Date: Fri Mar 3 05:56:29 2006
New Revision: 382816
URL: http://svn.apache.org/viewcvs?rev=382816&view=rev
Log:
added better MBeans for topic subscriptions so that they can see the discarded and matched statistics. Also added a configurable MessageEvictionStrategy
Added:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionView.java (with props)
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionViewMBean.java (with props)
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/MessageEvictionStrategy.java (with props)
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/OldestMessageEvictionStrategy.java (with props)
Modified:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java?rev=382816&r1=382815&r2=382816&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java Fri Mar 3 05:56:29 2006
@@ -43,6 +43,7 @@
import org.apache.activemq.broker.region.RegionBroker;
import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.broker.region.Topic;
+import org.apache.activemq.broker.region.TopicSubscription;
import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQMessage;
@@ -159,7 +160,12 @@
if(sub.getConsumerInfo().isDurable()){
view=new DurableSubscriptionView(this,context.getClientId(),sub);
}else{
- view=new SubscriptionView(context.getClientId(),sub);
+ if (sub instanceof TopicSubscription) {
+ view = new TopicSubscriptionView(context.getClientId(),(TopicSubscription) sub);
+ }
+ else {
+ view=new SubscriptionView(context.getClientId(),sub);
+ }
}
subscriptionMap.put(sub,objectName);
registerSubscription(objectName,sub.getConsumerInfo(),key,view);
Added: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionView.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionView.java?rev=382816&view=auto
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionView.java (added)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionView.java Fri Mar 3 05:56:29 2006
@@ -0,0 +1,52 @@
+/**
+ *
+ * Copyright 2005-2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.broker.jmx;
+
+import org.apache.activemq.broker.region.TopicSubscription;
+
+/**
+ *
+ * @version $Revision$
+ */
+public class TopicSubscriptionView extends SubscriptionView implements TopicSubscriptionViewMBean {
+
+ public TopicSubscriptionView(String clientId, TopicSubscription subs) {
+ super(clientId, subs);
+ }
+
+ protected TopicSubscription getTopicSubscription() {
+ return (TopicSubscription) subscription;
+ }
+
+ /**
+ * @return the number of messages discarded due to being a slow consumer
+ */
+ public int getDiscarded() {
+ TopicSubscription topicSubscription = getTopicSubscription();
+ return topicSubscription != null ? topicSubscription.discarded() : 0;
+ }
+
+ /**
+ * @return the number of matched messages (messages targeted for the
+ * subscription but not yet able to be dispatched due to the
+ * prefetch buffer being full).
+ */
+ public int getMatched() {
+ TopicSubscription topicSubscription = getTopicSubscription();
+ return topicSubscription != null ? topicSubscription.matched() : 0;
+ }
+}
Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionView.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionView.java
------------------------------------------------------------------------------
svn:keywords = Date Author Id Revision HeadURL
Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionView.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionViewMBean.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionViewMBean.java?rev=382816&view=auto
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionViewMBean.java (added)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionViewMBean.java Fri Mar 3 05:56:29 2006
@@ -0,0 +1,36 @@
+/**
+ *
+ * Copyright 2005-2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.broker.jmx;
+
+/**
+ *
+ * @version $Revision$
+ */
+public interface TopicSubscriptionViewMBean extends SubscriptionViewMBean {
+
+ /**
+ * @return the number of messages discarded due to being a slow consumer
+ */
+ public int getDiscarded();
+
+ /**
+ * @return the number of matched messages (messages targeted for the subscription but not
+ * yet able to be dispatched due to the prefetch buffer being full).
+ */
+ public int getMatched();
+
+}
Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionViewMBean.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionViewMBean.java
------------------------------------------------------------------------------
svn:keywords = Date Author Id Revision HeadURL
Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionViewMBean.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java?rev=382816&r1=382815&r2=382816&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java Fri Mar 3 05:56:29 2006
@@ -20,6 +20,8 @@
import javax.jms.JMSException;
import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.ConnectionContext;
+import org.apache.activemq.broker.region.policy.MessageEvictionStrategy;
+import org.apache.activemq.broker.region.policy.OldestMessageEvictionStrategy;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ConsumerInfo;
@@ -42,6 +44,8 @@
protected AtomicInteger dispatched=new AtomicInteger();
protected AtomicInteger delivered=new AtomicInteger();
private int maximumPendingMessages=-1;
+ private MessageEvictionStrategy messageEvictionStrategy = new OldestMessageEvictionStrategy();
+ private int discarded = 0;
private final Object matchedListMutex=new Object();
public TopicSubscription(Broker broker,ConnectionContext context,ConsumerInfo info,UsageManager usageManager)
@@ -64,8 +68,9 @@
if(maximumPendingMessages>0){
// lets discard old messages as we are a slow consumer
while(!matched.isEmpty()&&matched.size()>maximumPendingMessages){
- MessageReference oldMessage=(MessageReference) matched.removeFirst();
+ MessageReference oldMessage=messageEvictionStrategy.evictMessage(matched);
oldMessage.decrementReferenceCount();
+ discarded++;
if (log.isDebugEnabled()){
log.debug("Discarding message " + oldMessage);
}
@@ -122,7 +127,7 @@
}
public int pending(){
- return matched.size()-dispatched.get();
+ return matched()-dispatched();
}
public int dispatched(){
@@ -138,6 +143,26 @@
}
/**
+ * @return the number of messages discarded due to being a slow consumer
+ */
+ public int discarded() {
+ synchronized(matchedListMutex) {
+ return discarded;
+ }
+ }
+
+ /**
+ * @return the number of matched messages (messages targeted for the subscription but not
+ * yet able to be dispatched due to the prefetch buffer being full).
+ */
+ public int matched() {
+ synchronized(matchedListMutex) {
+ return matched.size();
+ }
+ }
+
+
+ /**
* Sets the maximum number of pending messages that can be matched against this consumer before old messages are
* discarded.
*/
@@ -145,6 +170,22 @@
this.maximumPendingMessages=maximumPendingMessages;
}
+ public MessageEvictionStrategy getMessageEvictionStrategy() {
+ return messageEvictionStrategy;
+ }
+
+ /**
+ * Sets the eviction strategy used to decide which message to evict when the slow consumer
+ * needs to discard messages
+ */
+ public void setMessageEvictionStrategy(MessageEvictionStrategy messageEvictionStrategy) {
+ this.messageEvictionStrategy = messageEvictionStrategy;
+ }
+
+
+ // Implementation methods
+ // -------------------------------------------------------------------------
+
private boolean isFull(){
return dispatched.get()-delivered.get()>=info.getPrefetchSize();
}
@@ -182,6 +223,6 @@
public String toString(){
return "TopicSubscription:"+" consumer="+info.getConsumerId()+", destinations="+destinations.size()
- +", dispatched="+dispatched+", delivered="+this.delivered+", matched="+this.matched.size();
+ +", dispatched="+dispatched()+", delivered="+delivered()+", matched="+matched()+", discarded="+discarded();
}
}
Added: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/MessageEvictionStrategy.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/MessageEvictionStrategy.java?rev=382816&view=auto
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/MessageEvictionStrategy.java (added)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/MessageEvictionStrategy.java Fri Mar 3 05:56:29 2006
@@ -0,0 +1,40 @@
+/**
+ *
+ * Copyright 2005-2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.broker.region.policy;
+
+import org.apache.activemq.broker.region.MessageReference;
+
+import java.io.IOException;
+import java.util.LinkedList;
+
+/**
+ * A strategy for evicting messages from slow consumers.
+ *
+ * @version $Revision$
+ */
+public interface MessageEvictionStrategy {
+
+ /**
+ * Find the message reference in the given list with oldest messages at the front and newer messages at the end
+ *
+ * @return the message that has been evicted.
+ * @throws IOException if an exception occurs such as reading a message content (but should not ever happen
+ * as usually all the messages will be in RAM when this method is called).
+ */
+ MessageReference evictMessage(LinkedList messages) throws IOException;
+
+}
Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/MessageEvictionStrategy.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/MessageEvictionStrategy.java
------------------------------------------------------------------------------
svn:keywords = Date Author Id Revision HeadURL
Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/MessageEvictionStrategy.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/OldestMessageEvictionStrategy.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/OldestMessageEvictionStrategy.java?rev=382816&view=auto
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/OldestMessageEvictionStrategy.java (added)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/OldestMessageEvictionStrategy.java Fri Mar 3 05:56:29 2006
@@ -0,0 +1,36 @@
+/**
+ *
+ * Copyright 2005-2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.broker.region.policy;
+
+import org.apache.activemq.broker.region.MessageReference;
+
+import java.util.LinkedList;
+
+/**
+ * An eviction strategy which evicts the oldest message first (which is the
+ * default).
+ *
+ * @org.apache.xbean.XBean
+ *
+ * @version $Revision$
+ */
+public class OldestMessageEvictionStrategy implements MessageEvictionStrategy {
+
+ public MessageReference evictMessage(LinkedList messages) {
+ return (MessageReference) messages.removeFirst();
+ }
+}
Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/OldestMessageEvictionStrategy.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/OldestMessageEvictionStrategy.java
------------------------------------------------------------------------------
svn:keywords = Date Author Id Revision HeadURL
Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/OldestMessageEvictionStrategy.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java?rev=382816&r1=382815&r2=382816&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java Fri Mar 3 05:56:29 2006
@@ -41,6 +41,7 @@
private DeadLetterStrategy deadLetterStrategy;
private int messageGroupHashBucketCount = 1024;
private PendingMessageLimitStrategy pendingMessageLimitStrategy;
+ private MessageEvictionStrategy messageEvictionStrategy;
public void configure(Queue queue) {
if (dispatchPolicy != null) {
@@ -81,6 +82,9 @@
subscription.setMaximumPendingMessages(value);
}
}
+ if (messageEvictionStrategy != null) {
+ subscription.setMessageEvictionStrategy(messageEvictionStrategy);
+ }
}
// Properties
@@ -156,4 +160,17 @@
public void setPendingMessageLimitStrategy(PendingMessageLimitStrategy pendingMessageLimitStrategy) {
this.pendingMessageLimitStrategy = pendingMessageLimitStrategy;
}
+
+ public MessageEvictionStrategy getMessageEvictionStrategy() {
+ return messageEvictionStrategy;
+ }
+
+ /**
+ * Sets the eviction strategy used to decide which message to evict when the
+ * slow consumer needs to discard messages
+ */
+ public void setMessageEvictionStrategy(MessageEvictionStrategy messageEvictionStrategy) {
+ this.messageEvictionStrategy = messageEvictionStrategy;
+ }
+
}