001/** 002 * Licensed to the Apache Software Foundation (ASF) under one or more 003 * contributor license agreements. See the NOTICE file distributed with 004 * this work for additional information regarding copyright ownership. 005 * The ASF licenses this file to You under the Apache License, Version 2.0 006 * (the "License"); you may not use this file except in compliance with 007 * the License. You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017package org.apache.activemq.broker.region.policy; 018 019import java.util.*; 020import java.util.Map.Entry; 021import java.util.concurrent.ConcurrentHashMap; 022 023import org.apache.activemq.broker.Broker; 024import org.apache.activemq.broker.ConnectionContext; 025import org.apache.activemq.broker.region.AbstractSubscription; 026import org.apache.activemq.broker.region.Destination; 027import org.apache.activemq.broker.region.Subscription; 028import org.slf4j.Logger; 029import org.slf4j.LoggerFactory; 030 031/** 032 * Abort slow consumers when they reach the configured threshold of slowness, 033 * 034 * default is that a consumer that has not Ack'd a message for 30 seconds is slow. 035 * 036 * @org.apache.xbean.XBean 037 */ 038public class AbortSlowAckConsumerStrategy extends AbortSlowConsumerStrategy { 039 040 private static final Logger LOG = LoggerFactory.getLogger(AbortSlowAckConsumerStrategy.class); 041 042 private final Map<String, Destination> destinations = new ConcurrentHashMap<String, Destination>(); 043 private long maxTimeSinceLastAck = 30*1000; 044 private boolean ignoreIdleConsumers = true; 045 046 public AbortSlowAckConsumerStrategy() { 047 this.name = "AbortSlowAckConsumerStrategy@" + hashCode(); 048 } 049 050 @Override 051 public void setBrokerService(Broker broker) { 052 super.setBrokerService(broker); 053 054 // Task starts right away since we may not receive any slow consumer events. 055 if (taskStarted.compareAndSet(false, true)) { 056 scheduler.executePeriodically(this, getCheckPeriod()); 057 } 058 } 059 060 @Override 061 public void slowConsumer(ConnectionContext context, Subscription subs) { 062 // Ignore these events, we just look at time since last Ack. 063 } 064 065 @Override 066 public void run() { 067 068 if (maxTimeSinceLastAck < 0) { 069 // nothing to do 070 LOG.info("no limit set, slowConsumer strategy has nothing to do"); 071 return; 072 } 073 074 075 List<Subscription> subscribersDestroyed = new LinkedList<Subscription>(); 076 // check for removed consumers also 077 for (Map.Entry<Subscription, SlowConsumerEntry> entry : slowConsumers.entrySet()) { 078 if (getMaxSlowDuration() > 0) { 079 // For subscriptions that are already slow we mark them again and check below if 080 // they've exceeded their configured lifetime. 081 entry.getValue().mark(); 082 } 083 if (!entry.getKey().isSlowConsumer()) { 084 subscribersDestroyed.add(entry.getKey()); 085 } 086 } 087 088 for (Subscription subscription: subscribersDestroyed) { 089 slowConsumers.remove(subscription); 090 } 091 092 List<Destination> disposed = new ArrayList<Destination>(); 093 094 for (Destination destination : destinations.values()) { 095 if (destination.isDisposed()) { 096 disposed.add(destination); 097 continue; 098 } 099 100 // Not explicitly documented but this returns a stable copy. 101 List<Subscription> subscribers = destination.getConsumers(); 102 103 updateSlowConsumersList(subscribers); 104 } 105 106 // Clean up an disposed destinations to save space. 107 for (Destination destination : disposed) { 108 destinations.remove(destination.getName()); 109 } 110 111 abortAllQualifiedSlowConsumers(); 112 } 113 114 private void updateSlowConsumersList(List<Subscription> subscribers) { 115 for (Subscription subscriber : subscribers) { 116 if (isIgnoreNetworkSubscriptions() && subscriber.getConsumerInfo().isNetworkSubscription()) { 117 if (slowConsumers.remove(subscriber) != null) { 118 LOG.info("network sub: {} is no longer slow", subscriber.getConsumerInfo().getConsumerId()); 119 } 120 continue; 121 } 122 123 if (isIgnoreIdleConsumers() && subscriber.getDispatchedQueueSize() == 0) { 124 // Not considered Idle so ensure its cleared from the list 125 if (slowConsumers.remove(subscriber) != null) { 126 LOG.info("idle sub: {} is no longer slow", subscriber.getConsumerInfo().getConsumerId()); 127 } 128 continue; 129 } 130 131 long lastAckTime = subscriber.getTimeOfLastMessageAck(); 132 long timeDelta = System.currentTimeMillis() - lastAckTime; 133 134 if (timeDelta > maxTimeSinceLastAck) { 135 if (!slowConsumers.containsKey(subscriber)) { 136 LOG.debug("sub: {} is now slow", subscriber.getConsumerInfo().getConsumerId()); 137 SlowConsumerEntry entry = new SlowConsumerEntry(subscriber.getContext()); 138 entry.mark(); // mark consumer on first run 139 if (subscriber instanceof AbstractSubscription) { 140 AbstractSubscription abstractSubscription = (AbstractSubscription) subscriber; 141 if (!abstractSubscription.isSlowConsumer()) { 142 abstractSubscription.setSlowConsumer(true); 143 for (Destination destination: abstractSubscription.getDestinations()) { 144 destination.slowConsumer(broker.getAdminConnectionContext(), abstractSubscription); 145 } 146 } 147 } 148 slowConsumers.put(subscriber, entry); 149 } else if (getMaxSlowCount() > 0) { 150 slowConsumers.get(subscriber).slow(); 151 } 152 } else { 153 if (slowConsumers.remove(subscriber) != null) { 154 LOG.info("sub: {} is no longer slow", subscriber.getConsumerInfo().getConsumerId()); 155 } 156 } 157 } 158 } 159 160 private void abortAllQualifiedSlowConsumers() { 161 HashMap<Subscription, SlowConsumerEntry> toAbort = new HashMap<Subscription, SlowConsumerEntry>(); 162 for (Entry<Subscription, SlowConsumerEntry> entry : slowConsumers.entrySet()) { 163 if (getMaxSlowDuration() > 0 && (entry.getValue().markCount * getCheckPeriod() >= getMaxSlowDuration()) || 164 getMaxSlowCount() > 0 && entry.getValue().slowCount >= getMaxSlowCount()) { 165 166 LOG.trace("Transferring consumer{} to the abort list: {} slow duration = {}, slow count = {}", 167 new Object[]{ entry.getKey().getConsumerInfo().getConsumerId(), 168 entry.getValue().markCount * getCheckPeriod(), 169 entry.getValue().getSlowCount() }); 170 171 toAbort.put(entry.getKey(), entry.getValue()); 172 slowConsumers.remove(entry.getKey()); 173 } else { 174 175 LOG.trace("Not yet time to abort consumer {}: slow duration = {}, slow count = {}", new Object[]{ entry.getKey().getConsumerInfo().getConsumerId(), entry.getValue().markCount * getCheckPeriod(), entry.getValue().slowCount }); 176 177 } 178 } 179 180 // Now if any subscriptions made it into the aborts list we can kick them. 181 abortSubscription(toAbort, isAbortConnection()); 182 } 183 184 @Override 185 public void addDestination(Destination destination) { 186 this.destinations.put(destination.getName(), destination); 187 } 188 189 /** 190 * Gets the maximum time since last Ack before a subscription is considered to be slow. 191 * 192 * @return the maximum time since last Ack before the consumer is considered to be slow. 193 */ 194 public long getMaxTimeSinceLastAck() { 195 return maxTimeSinceLastAck; 196 } 197 198 /** 199 * Sets the maximum time since last Ack before a subscription is considered to be slow. 200 * 201 * @param maxTimeSinceLastAck 202 * the maximum time since last Ack (mills) before the consumer is considered to be slow. 203 */ 204 public void setMaxTimeSinceLastAck(long maxTimeSinceLastAck) { 205 this.maxTimeSinceLastAck = maxTimeSinceLastAck; 206 } 207 208 /** 209 * Returns whether the strategy is configured to ignore consumers that are simply idle, i.e 210 * consumers that have no pending acks (dispatch queue is empty). 211 * 212 * @return true if the strategy will ignore idle consumer when looking for slow consumers. 213 */ 214 public boolean isIgnoreIdleConsumers() { 215 return ignoreIdleConsumers; 216 } 217 218 /** 219 * Sets whether the strategy is configured to ignore consumers that are simply idle, i.e 220 * consumers that have no pending acks (dispatch queue is empty). 221 * 222 * When configured to not ignore idle consumers this strategy acks not only on consumers 223 * that are actually slow but also on any consumer that has not received any messages for 224 * the maxTimeSinceLastAck. This allows for a way to evict idle consumers while also 225 * aborting slow consumers. 226 * 227 * @param ignoreIdleConsumers 228 * Should this strategy ignore idle consumers or consider all consumers when checking 229 * the last ack time verses the maxTimeSinceLastAck value. 230 */ 231 public void setIgnoreIdleConsumers(boolean ignoreIdleConsumers) { 232 this.ignoreIdleConsumers = ignoreIdleConsumers; 233 } 234}