flume channel monitor实现

   对于flume的监控，只需要监控channel的性能数据即可，source和sink的性能一部分可以从channel中表现出来。
以MemoryChannel为例，在MemoryTransaction的构造函数中会实例化一个org.apache.flume.instrumentation.ChannelCounter对象

    public MemoryTransaction( int transCapacity, ChannelCounter counter) {
      putList = new LinkedBlockingDeque<Event>(transCapacity);
      takeList = new LinkedBlockingDeque<Event>(transCapacity);
      channelCounter = counter;
    }

org.apache.flume.instrumentation.ChannelCounter 定义了几个计数器用来记录channel的性能数据

  private static final String COUNTER_CHANNEL_SIZE = "channel.current.size"; //已经使用的容量大小
  private static final String COUNTER_EVENT_PUT_ATTEMPT = "channel.event.put.attempt"; //source到channel尝试插入的数据（不管是否成功）
  private static final String COUNTER_EVENT_TAKE_ATTEMPT = "channel.event.take.attempt"; //sink从channel尝试消费的数据（不管是否成功）
  private static final String COUNTER_EVENT_PUT_SUCCESS = "channel.event.put.success"; //source到channel成功插入的数据
  private static final String COUNTER_EVENT_TAKE_SUCCESS = "channel.event.take.success"; //sink从channel成功消费的数据
  private static final String COUNTER_CHANNEL_CAPACITY = "channel.capacity";  //总容量大小

并封装了相关的方法，来操作这些性能计数器：

比如channel.event.put.attempt的由getEventPutAttemptCount和incrementEventPutAttemptCount操作：

  public long incrementEventPutAttemptCount() { //用于数量增加1
    return increment( COUNTER_EVENT_PUT_ATTEMPT);
  }
  public long getEventPutAttemptCount() { //用于获取值
    return get( COUNTER_EVENT_PUT_ATTEMPT);
  }

而在channel的相关操作中会使用到这些方法：

    protected void doPut(Event event) throws InterruptedException {
      channelCounter.incrementEventPutAttemptCount(); //比如在插入数据的操作开始时，增加channel.event.put.attempt的值
      int eventByteSize = (int)Math.ceil(estimateEventSize(event)/ byteCapacitySlotSize);
      if (! putList.offer(event)) {
        throw new ChannelException(
          "Put queue for MemoryTransaction of capacity " +
            putList.size() + " full, consider committing more frequently, " +
            "increasing capacity or increasing thread count" );
      }
      putByteCounter += eventByteSize;
    }

counter的注册使用，以memorychannel相关为例：
ChannelCounter扩展了MonitoredCounterGroup类并实现了ChannelCounterMBean接口
MonitoredCounterGroup是一个抽象类，其具体的实现类定义了具体的组件的性能计数器和对应的封装方法

ChannelCounter中包含的所有的可用的counter:

private static final String[] ATTRIBUTES = {
    COUNTER_CHANNEL_SIZE, COUNTER_EVENT_PUT_ATTEMPT,
    COUNTER_EVENT_TAKE_ATTEMPT, COUNTER_EVENT_PUT_SUCCESS,
    COUNTER_EVENT_TAKE_SUCCESS, COUNTER_CHANNEL_CAPACITY
  };

ChannelCounter的构造方法调用MonitoredCounterGroup的构造方法：

  public ChannelCounter(String name) {
    super(MonitoredCounterGroup.Type. CHANNEL, name, ATTRIBUTES ); //调用MonitoredCounterGroup构造方法
  }
MonitoredCounterGroup构造方法：
private final Map<String, AtomicLong> counterMap;
....
  protected MonitoredCounterGroup(Type type, String name, String... attrs) {
    this. type = type;
    this. name = name;
    Map<String, AtomicLong> counterInitMap = new HashMap<String, AtomicLong>(); // 声明一个初始的hashmap，用来存放counter name到value的对应关系
    // Initialize the counters
    for (String attribute : attrs) {
      counterInitMap.put(attribute, new AtomicLong(0L)); //初始value都为0
    }
    counterMap = Collections.unmodifiableMap(counterInitMap); //返回hashmap不可更改的映射视图
    startTime = new AtomicLong(0L);
    stopTime = new AtomicLong(0L);
  }

这里Type是一个enum类型，可取值：

  public static enum Type {
    SOURCE,
    CHANNEL_PROCESSOR,
    CHANNEL,
    SINK_PROCESSOR,
    SINK,
    INTERCEPTOR,
    SERIALIZER,
    OTHER
  };

在MemoryChannel中的start方法中启动：

  public synchronized void start() {
    channelCounter.start(); //调用MonitoredCounterGroup的start方法
    channelCounter.setChannelSize( queue.size());
    channelCounter.setChannelCapacity(Long. valueOf(
            queue.size() + queue.remainingCapacity()));
    super.start();
  }

MonitoredCounterGroup.start:

  public void start() {
    register(); //调用register方法注册counter，主要是调用ManagementFactory. getPlatformMBeanServer().registerMBean( this, objName);进行注册mbean操作，把ChannelCounter对象作为mbean进行注册
    stopTime.set(0L);
    for (String counter : counterMap.keySet()) {
      counterMap.get(counter).set(0L); //设置值都为0
    }
    startTime.set(System. currentTimeMillis());
    logger.info( "Component type: " + type + ", name: " + name + " started" );
  }

这样就可以通过jmx获取的监控项

本文出自 “菜光光的博客” 博客，请务必保留此出处http://caiguangguang.blog.51cto.com/1652935/1617019