flume源码分析-SinkProcessor

xglys

　　flume包括三种sink processor，DefaultSinkProcessor，FailoverSinkProcessor，LoadBalancingSinkProcessor
　　Default sink processor that only accepts a single sink, passing on process results without any additional handling. Suitable for all sinks that aren't assigned to a group.

public class DefaultSinkProcessor implements SinkProcessor,
ConfigurableComponent {
private Sink sink;
private LifecycleState lifecycleState;
@Override
public void start() {
Preconditions.checkNotNull(sink, "DefaultSinkProcessor sink not set");
sink.start();
lifecycleState = LifecycleState.START;
}
@Override
public void stop() {
Preconditions.checkNotNull(sink, "DefaultSinkProcessor sink not set");
sink.stop();
lifecycleState = LifecycleState.STOP;
}

@Override
public Status process() throws EventDeliveryException {
return sink.process();
}
@Override
public void setSinks(List<Sink> sinks) {
Preconditions.checkNotNull(sinks);
Preconditions.checkArgument(sinks.size() == 1, "DefaultSinkPolicy can "
+ "only handle one sink, "
+ "try using a policy that supports multiple sinks");
sink = sinks.get(0);
}
@Override
public void configure(ComponentConfiguration conf) {
}
}

　　[size=1em]FailoverSinkProcessor用来处理一个sink的group组，当高优先级的sink处理失败后，FailoverSinkProcessor会选择另一个sink来处理.

/**
* FailoverSinkProcessor maintains a prioritized list of sinks,
* guarranteeing that so long as one is available events will be processed.
*
* The failover mechanism works by relegating failed sinks to a pool
* where they are assigned a cooldown period, increasing with sequential
* failures before they are retried. Once a sink succesfully sends an
* event it is restored to the live pool.
*
* FailoverSinkProcessor is in no way thread safe and expects to be run via
* SinkRunner Additionally, setSinks must be called before configure, and
* additional sinks cannot be added while running
*
* To configure, set a sink groups processor to "failover" and set priorities
* for individual sinks, all priorities must be unique. Furthermore, an
* upper limit to failover time can be set(in miliseconds) using maxpenalty
*
* Ex)
*
* host1.sinkgroups = group1
*
* host1.sinkgroups.group1.sinks = sink1 sink2
* host1.sinkgroups.group1.processor.type = failover
* host1.sinkgroups.group1.processor.priority.sink1 = 5
* host1.sinkgroups.group1.processor.priority.sink2 = 10
* host1.sinkgroups.group1.processor.maxpenalty = 10000
*
*/
public class FailoverSinkProcessor extends AbstractSinkProcessor {
private static final int FAILURE_PENALTY = 1000;
private static final int DEFAULT_MAX_PENALTY = 30000;
private class FailedSink implements Comparable<FailedSink> {
private Long refresh;
private Integer priority;
private Sink sink;
private Integer sequentialFailures;
public FailedSink(Integer priority, Sink sink, int seqFailures) {
this.sink = sink;
this.priority = priority;
this.sequentialFailures = seqFailures;
adjustRefresh();
}
@Override
public int compareTo(FailedSink arg0) {
return refresh.compareTo(arg0.refresh);
}
public Long getRefresh() {
return refresh;
}
public Sink getSink() {
return sink;
}
public Integer getPriority() {
return priority;
}
public void incFails() {
sequentialFailures++;
adjustRefresh();
logger.debug("Sink {} failed again, new refresh is at {}, " +
"current time {}", new Object[] {
sink.getName(), refresh, System.currentTimeMillis()});
}
private void adjustRefresh() {
refresh = System.currentTimeMillis()
+ Math.min(maxPenalty, (1 << sequentialFailures) * FAILURE_PENALTY);
}
}
private static final Logger logger = LoggerFactory
.getLogger(FailoverSinkProcessor.class);
private static final String PRIORITY_PREFIX = "priority.";
private static final String MAX_PENALTY_PREFIX = "maxpenalty";
private Map<String, Sink> sinks;
private Sink activeSink;
private SortedMap<Integer, Sink> liveSinks;
private Queue<FailedSink> failedSinks;
private int maxPenalty;
@Override
public void configure(Context context) {
liveSinks = new TreeMap<Integer, Sink>();
failedSinks = new PriorityQueue<FailedSink>();
Integer nextPrio = 0;
String maxPenaltyStr = context.getString(MAX_PENALTY_PREFIX);
if(maxPenaltyStr == null) {
maxPenalty = DEFAULT_MAX_PENALTY;
} else {
try {
maxPenalty = Integer.parseInt(maxPenaltyStr);
} catch (NumberFormatException e) {
logger.warn("{} is not a valid value for {}",
new Object[] { maxPenaltyStr, MAX_PENALTY_PREFIX });
maxPenalty  = DEFAULT_MAX_PENALTY;
}
}
for (Entry<String, Sink> entry : sinks.entrySet()) {
String priStr = PRIORITY_PREFIX + entry.getKey();
Integer priority;
try {
priority =  Integer.parseInt(context.getString(priStr));
} catch (Exception e) {
priority = --nextPrio;
}
if(!liveSinks.containsKey(priority)) {
liveSinks.put(priority, sinks.get(entry.getKey()));
} else {
logger.warn("Sink {} not added to FailverSinkProcessor as priority" +
"duplicates that of sink {}", entry.getKey(),
liveSinks.get(priority));
}
}
activeSink = liveSinks.get(liveSinks.lastKey());
}
@Override
public Status process() throws EventDeliveryException {
// Retry any failed sinks that have gone through their "cooldown" period
Long now = System.currentTimeMillis();
while(!failedSinks.isEmpty() && failedSinks.peek().getRefresh() < now) {
FailedSink cur = failedSinks.poll();
Status s;
try {
s = cur.getSink().process();
if (s  == Status.READY) {
liveSinks.put(cur.getPriority(), cur.getSink());
activeSink = liveSinks.get(liveSinks.lastKey());
logger.debug("Sink {} was recovered from the fail list",
cur.getSink().getName());
} else {
// if it's a backoff it needn't be penalized.
failedSinks.add(cur);
}
return s;
} catch (Exception e) {
cur.incFails();
failedSinks.add(cur);
}
}
Status ret = null;
while(activeSink != null) {
try {
ret = activeSink.process();
return ret;
} catch (Exception e) {
logger.warn("Sink {} failed and has been sent to failover list",
activeSink.getName(), e);
activeSink = moveActiveToDeadAndGetNext();
}
}
throw new EventDeliveryException("All sinks failed to process, " +
"nothing left to failover to");
}
}


public class LoadBalancingSinkProcessor extends AbstractSinkProcessor {
public static final String CONFIG_SELECTOR = "selector";
public static final String CONFIG_SELECTOR_PREFIX = CONFIG_SELECTOR + ".";
public static final String SELECTOR_NAME_ROUND_ROBIN = "ROUND_ROBIN";
public static final String SELECTOR_NAME_RANDOM = "RANDOM";

private static final Logger LOGGER = LoggerFactory
.getLogger(LoadBalancingSinkProcessor.class);
private SinkSelector selector;
@Override
public void configure(Context context) {
Preconditions.checkState(getSinks().size() > 1,
"The LoadBalancingSinkProcessor cannot be used for a single sink. "
+ "Please configure more than one sinks and try again.");
String selectorTypeName = context.getString(CONFIG_SELECTOR,
SELECTOR_NAME_ROUND_ROBIN);
selector = null;
if (selectorTypeName.equalsIgnoreCase(SELECTOR_NAME_ROUND_ROBIN)) {
selector = new RoundRobinSinkSelector();
} else if (selectorTypeName.equalsIgnoreCase(SELECTOR_NAME_RANDOM)) {
selector = new RandomOrderSinkSelector();
} else {
try {
@SuppressWarnings("unchecked")
Class<? extends SinkSelector> klass = (Class<? extends SinkSelector>)
Class.forName(selectorTypeName);
selector = klass.newInstance();
} catch (Exception ex) {
throw new FlumeException("Unable to instantiate sink selector: "
+ selectorTypeName, ex);
}
}
selector.setSinks(getSinks());
selector.configure(
new Context(context.getSubProperties(CONFIG_SELECTOR_PREFIX)));
LOGGER.debug("Sink selector: " + selector + " initialized");
}
@Override
public Status process() throws EventDeliveryException {
Status status = null;
Iterator<Sink> sinkIterator = selector.createSinkIterator();
while (sinkIterator.hasNext()) {
Sink sink = sinkIterator.next();
try {
status = sink.process();
break;
} catch (Exception ex) {
LOGGER.warn("Sink failed to consume event. "
+ "Attempting next sink if available.", ex);
}
}
if (status == null) {
throw new EventDeliveryException("All configured sinks have failed");
}
return status;
}

/**
* <p>
* An interface that allows the LoadBalancingSinkProcessor to use
* a load-balancing strategy such as round-robin, random distribution etc.
* Implementations of this class can be plugged into the system via
* processor configuration and are used to select a sink on every invocation.
* </p>
* <p>
* An instance of the configured sink selector is create during the processor
* configuration, its {@linkplain #setSinks(List)} method is invoked following
* which it is configured via a subcontext. Once configured, the lifecycle of
* this selector is tied to the lifecycle of the sink processor.
* </p>
* <p>
* At runtime, the processor invokes the {@link #createSinkIterator()}
* method for every <tt>process</tt> call to create an iteration order over
* the available sinks. The processor then loops through this iteration order
* until one of the sinks succeeds in processing the event. If the iterator
* is exhausted and none of the sinks succeed, the processor will raise
* an <tt>EventDeliveryException</tt>.
* </p>
*/
public interface SinkSelector extends Configurable, LifecycleAware {
void setSinks(List<Sink> sinks);
Iterator<Sink> createSinkIterator();
}
/**
* A sink selector that implements the round-robin sink selection policy.
* This implementation is not MT safe.
*/
private static class RoundRobinSinkSelector extends AbstractSinkSelector {
private int nextHead = 0;
@Override
public Iterator<Sink> createSinkIterator() {
int size = getSinks().size();
int[] indexOrder = new int[size];
int begin = nextHead++;
if (nextHead == size) {
nextHead = 0;
}
for (int i=0; i < size; i++) {
indexOrder = (begin + i)%size;
}
return new SpecificOrderIterator<Sink>(indexOrder, getSinks());
}
}
/**
* A sink selector that implements a random sink selection policy. This
* implementation is not thread safe.
*/
private static class RandomOrderSinkSelector extends AbstractSinkSelector {
private Random random = new Random(System.currentTimeMillis());
@Override
public Iterator<Sink> createSinkIterator() {
int size = getSinks().size();
int[] indexOrder = new int[size];
List<Integer> indexList = new ArrayList<Integer>();
for (int i=0; i<size; i++) {
indexList.add(i);
}
while (indexList.size() != 1) {
int pick = random.nextInt(indexList.size());
indexOrder[indexList.size() - 1] = indexList.remove(pick);
}
indexOrder[0] = indexList.get(0);
return new SpecificOrderIterator<Sink>(indexOrder, getSinks());
}
}
}

　　[size=1em]LoadBalancingSinkProcessor是用来做load balance的，分为两种selector，RandomOrderSinkSelector和RoundRobinSinkSelector。RoundRobinSinkSelector循环选取一个sink作为最先处理。[size=1em]RandomOrderSinkSelector随机选取一个作为最先处理。

public class LoadBalancingSinkProcessor extends AbstractSinkProcessor {
public static final String CONFIG_SELECTOR = "selector";
public static final String CONFIG_SELECTOR_PREFIX = CONFIG_SELECTOR + ".";
public static final String SELECTOR_NAME_ROUND_ROBIN = "ROUND_ROBIN";
public static final String SELECTOR_NAME_RANDOM = "RANDOM";

private static final Logger LOGGER = LoggerFactory
.getLogger(LoadBalancingSinkProcessor.class);
private SinkSelector selector;
@Override
public void configure(Context context) {
Preconditions.checkState(getSinks().size() > 1,
"The LoadBalancingSinkProcessor cannot be used for a single sink. "
+ "Please configure more than one sinks and try again.");
String selectorTypeName = context.getString(CONFIG_SELECTOR,
SELECTOR_NAME_ROUND_ROBIN);
selector = null;
if (selectorTypeName.equalsIgnoreCase(SELECTOR_NAME_ROUND_ROBIN)) {
selector = new RoundRobinSinkSelector();
} else if (selectorTypeName.equalsIgnoreCase(SELECTOR_NAME_RANDOM)) {
selector = new RandomOrderSinkSelector();
} else {
try {
@SuppressWarnings("unchecked")
Class<? extends SinkSelector> klass = (Class<? extends SinkSelector>)
Class.forName(selectorTypeName);
selector = klass.newInstance();
} catch (Exception ex) {
throw new FlumeException("Unable to instantiate sink selector: "
+ selectorTypeName, ex);
}
}
selector.setSinks(getSinks());
selector.configure(
new Context(context.getSubProperties(CONFIG_SELECTOR_PREFIX)));
LOGGER.debug("Sink selector: " + selector + " initialized");
}
@Override
public void start() {
super.start();
selector.start();
}
@Override
public void stop() {
super.stop();
selector.stop();
}
@Override
public Status process() throws EventDeliveryException {
Status status = null;
Iterator<Sink> sinkIterator = selector.createSinkIterator();
while (sinkIterator.hasNext()) {
Sink sink = sinkIterator.next();
try {
status = sink.process();
break;
} catch (Exception ex) {
LOGGER.warn("Sink failed to consume event. "
+ "Attempting next sink if available.", ex);
}
}
if (status == null) {
throw new EventDeliveryException("All configured sinks have failed");
}
return status;
}

/**
* <p>
* An interface that allows the LoadBalancingSinkProcessor to use
* a load-balancing strategy such as round-robin, random distribution etc.
* Implementations of this class can be plugged into the system via
* processor configuration and are used to select a sink on every invocation.
* </p>
* <p>
* An instance of the configured sink selector is create during the processor
* configuration, its {@linkplain #setSinks(List)} method is invoked following
* which it is configured via a subcontext. Once configured, the lifecycle of
* this selector is tied to the lifecycle of the sink processor.
* </p>
* <p>
* At runtime, the processor invokes the {@link #createSinkIterator()}
* method for every <tt>process</tt> call to create an iteration order over
* the available sinks. The processor then loops through this iteration order
* until one of the sinks succeeds in processing the event. If the iterator
* is exhausted and none of the sinks succeed, the processor will raise
* an <tt>EventDeliveryException</tt>.
* </p>
*/
public interface SinkSelector extends Configurable, LifecycleAware {
void setSinks(List<Sink> sinks);
Iterator<Sink> createSinkIterator();
}
/**
* A sink selector that implements the round-robin sink selection policy.
* This implementation is not MT safe.
*/
private static class RoundRobinSinkSelector extends AbstractSinkSelector {
private int nextHead = 0;
@Override
public Iterator<Sink> createSinkIterator() {
int size = getSinks().size();
int[] indexOrder = new int[size];
int begin = nextHead++;
if (nextHead == size) {
nextHead = 0;
}
for (int i=0; i < size; i++) {
indexOrder = (begin + i)%size;
}
return new SpecificOrderIterator<Sink>(indexOrder, getSinks());
}
}
/**
* A sink selector that implements a random sink selection policy. This
* implementation is not thread safe.
*/
private static class RandomOrderSinkSelector extends AbstractSinkSelector {
private Random random = new Random(System.currentTimeMillis());
@Override
public Iterator<Sink> createSinkIterator() {
int size = getSinks().size();
int[] indexOrder = new int[size];
List<Integer> indexList = new ArrayList<Integer>();
for (int i=0; i<size; i++) {
indexList.add(i);
}
while (indexList.size() != 1) {
int pick = random.nextInt(indexList.size());
indexOrder[indexList.size() - 1] = indexList.remove(pick);
}
indexOrder[0] = indexList.get(0);
return new SpecificOrderIterator<Sink>(indexOrder, getSinks());
}
}
}