activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jstrac...@apache.org
Subject svn commit: r382816 - in /incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker: jmx/ region/ region/policy/
Date Fri, 03 Mar 2006 13:56:30 GMT
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;
+    }
+
 }



Mime
View raw message