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