基于Hadoop的日志收集框架---Chukwa的源码分析(收集器)

谁践踏了优雅 2022-03-20 08:54 182阅读 0赞

1.接口、实现类简介  
    org.apache.hadoop.chukwa.datacollection.collector.CollectorStub  
    收集器服务类，使用jetty实现了一个webserver以处理连接器提交的数据块  
  
    org.apache.hadoop.chukwa.datacollection.collector.servlet.ServletCollector  
    收集器的servlet，用于处理http请求  
  
    org.apache.hadoop.chukwa.datacollection.writer.ChukwaWriter  
    chukwa写入接口  
  
    org.apache.hadoop.chukwa.datacollection.writer.PipelineStageWriter  
    chukwa写入接口的实现类，做为一级管道使用，只实现了init(Configuration)方法，add(List<Chunk>)的实现放在了二级管道中  
  
    org.apache.hadoop.chukwa.datacollection.writer.SeqFileWriter  
    chukwa写入接口的实现类，继承自PipelineStageWriter，实现了add(List<Chunk>)方法  
  
2.启动、处理流程  
  
(1)收集服务的启动  
    org.apache.hadoop.chukwa.datacollection.collector.CollectorStub

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  // 线程数   
2.  static int THREADS = 120;   
3.    
4.  // 使用jetty提供http服务   
5.  public static Server jettyServer = null;   
6.    
7.  /\*\*  
8.   \* 1. 创建守护进程"Collector",将进程号写入相应的pid文件，以便于运行stop命令时可根据此pid文件杀死进程 
9.   \* 2. 校验启动参数，可使用portno、writer、servlet来指定端口、写入数据的接口实现类、处理请求的servlet实现类 
10.  \* 3. 设置jetty服务的connector、处理相应请求的servlet,比如   
11.  \*    /\*        org.apache.hadoop.chukwa.datacollection.collector.servlet.ServletCollector 
12.  \*    /acks     org.apache.hadoop.chukwa.datacollection.collector.servlet.CommitCheckServlet 
13.  \*    /logs     org.apache.hadoop.chukwa.datacollection.collector.servlet.LogDisplayServlet 
14.  \* 4. 启动http服务  
15.  \*/  
16. public static void main(String\[\] args)

// 线程数
    static int THREADS = 120;
    
    // 使用jetty提供http服务
    public static Server jettyServer = null;
    
    /**
     * 1. 创建守护进程"Collector",将进程号写入相应的pid文件，以便于运行stop命令时可根据此pid文件杀死进程
     * 2. 校验启动参数，可使用portno、writer、servlet来指定端口、写入数据的接口实现类、处理请求的servlet实现类
     * 3. 设置jetty服务的connector、处理相应请求的servlet,比如
     * 	  /* 		org.apache.hadoop.chukwa.datacollection.collector.servlet.ServletCollector
     *    /acks		org.apache.hadoop.chukwa.datacollection.collector.servlet.CommitCheckServlet
     *    /logs		org.apache.hadoop.chukwa.datacollection.collector.servlet.LogDisplayServlet
     * 4. 启动http服务
     */
    public static void main(String[] args)

(2)处理http请求  
    org.apache.hadoop.chukwa.datacollection.collector.servlet.ServletCollector

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  // 访问路径   
2.  public static final String PATH = "chukwa";    
3.    
4.  // 写数据的接口   
5.  ChukwaWriter writer   
6.    
7.  /\*\*  
8.   \* servlet初始化，实例化writer，首先使用配置项中"chukwaCollector.writerClass"的设置构造writer，如果失败则使用SeqFileWriter 
9.   \*/  
10. public void init(ServletConfig servletConf)   
11.   
12. /\*\*  
13.  \* 对于get请求以html形式输出jetty服务的信息  
14.  \*/  
15. protected void doGet(HttpServletRequest req, HttpServletResponse resp)   
16.   
17. /\*\*  
18.  \* 对于post请求的处理放在了accept()方法中  
19.  \*/  
20. protected void doPost(HttpServletRequest req, HttpServletResponse resp)   
21.   
22. /\*\*  
23.  \* 1. 使用ServletDiagnostics封装当前请求中的数据信息  
24.  \* 2. 从req中获取输入流，依次读取数据块数量(numEvents)、数据块(ChunkImpl) 
25.  \*/  
26. protected void accept(HttpServletRequest req, HttpServletResponse resp)

// 访问路径
    public static final String PATH = "chukwa"; 
    
    // 写数据的接口
    ChukwaWriter writer
    
    /**
     * servlet初始化，实例化writer，首先使用配置项中"chukwaCollector.writerClass"的设置构造writer，如果失败则使用SeqFileWriter
     */
    public void init(ServletConfig servletConf)
    
    /**
     * 对于get请求以html形式输出jetty服务的信息
     */
    protected void doGet(HttpServletRequest req, HttpServletResponse resp)
    
    /**
     * 对于post请求的处理放在了accept()方法中
     */
    protected void doPost(HttpServletRequest req, HttpServletResponse resp)
    
    /**
     * 1. 使用ServletDiagnostics封装当前请求中的数据信息
     * 2. 从req中获取输入流，依次读取数据块数量(numEvents)、数据块(ChunkImpl)
     */
    protected void accept(HttpServletRequest req, HttpServletResponse resp)

(3)写入数据信息  
    org.apache.hadoop.chukwa.datacollection.writer.ChukwaWriter  
    writer接口

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  /\*\*  
2.   \* 初始化Writer  
3.   \*/  
4.  public void init(Configuration c)   
5.    
6.  /\*\*  
7.   \* 将数据列表写入  
8.   \*/  
9.  public CommitStatus add(List<Chunk> chunks)

/**
     * 初始化Writer
     */
    public void init(Configuration c)
    
    /**
     * 将数据列表写入
     */
    public CommitStatus add(List<Chunk> chunks)

org.apache.hadoop.chukwa.datacollection.writer.PipelineStageWriter  
    writer接口的实现类

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  // 将写入数据的实现有此对象完成   
2.  ChukwaWriter writer;   
3.    
4.  /\*\*  
5.   \* 1. 读取配置"chukwaCollector.pipeline",对数据块做依次处理   
6.   \* 2. 将第一个管道赋给writer  
7.   \* 3. 遍历其它管道，如果是PipelineableWriter的子类则将其设置为下一步要处理数据块所使用的管道 
8.   \* 4. 将最后一个管道设置为当前管道下一步要处理时所使用的管道  
9.   \*    即此write并未做数据块的处理，只是使用类似unix管道的方式，在"chukwaCollector.pipeline"配置项中 
10.  \*    配置数据块处理的先后顺序    
11.  \*/  
12. public void init(Configuration conf)

// 将写入数据的实现有此对象完成
    ChukwaWriter writer;
    
    /**
     * 1. 读取配置"chukwaCollector.pipeline",对数据块做依次处理
     * 2. 将第一个管道赋给writer
     * 3. 遍历其它管道，如果是PipelineableWriter的子类则将其设置为下一步要处理数据块所使用的管道
     * 4. 将最后一个管道设置为当前管道下一步要处理时所使用的管道
     *	  即此write并未做数据块的处理，只是使用类似unix管道的方式，在"chukwaCollector.pipeline"配置项中
     *    配置数据块处理的先后顺序	
     */
    public void init(Configuration conf)

org.apache.hadoop.chukwa.datacollection.writer.PipelineableWriter  
    管道处理的writer实现类，实现了ChukwaWriter接口，但做为抽象类，只定义了管道处理的实现方式

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  // 下一步处理时所使用的writer   
2.  ChukwaWriter next;   
3.    
4.  /\*\*  
5.   \* 使用此方法设置writer  
6.   \*/  
7.  public void setNextStage(ChukwaWriter next) \{   
8.      this.next = next;   
9.  \}

// 下一步处理时所使用的writer
    ChukwaWriter next;
    
    /**
     * 使用此方法设置writer
     */
    public void setNextStage(ChukwaWriter next) {
    	this.next = next;
    }

org.apache.hadoop.chukwa.datacollection.writer.SocketTeeWriter  
    继承自PipelineableWriter，使用实现了管道处理数据块

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  /\*\*  
2.   \* 数据格式化的三种类型:原始方式、写入方式、带头方式  
3.   \*/  
4.  static enum DataFormat \{   
5.      Raw, Writable, Header   
6.  \};   
7.    
8.  // 监听数据传入请求   
9.  SocketListenThread listenThread;   
10.   
11. // 把使用管道处理数据的过程放在Tee线程中   
12. List<Tee> tees;   
13.   
14. // 下一步处理时所使用的writer   
15. ChukwaWriter next;   
16.   
17. /\*\*  
18.  \* 启动监听线程，初始化Tee线程列表  
19.  \*/  
20. public void init(Configuration c)   
21.   
22. /\*\*  
23.  \* 1. 将数据块传递给下一个管道处理  
24.  \* 2. 遍历Tee线程列表，循环处理数据块  
25.  \*/  
26. public CommitStatus add(List<Chunk> chunks)

/**
     * 数据格式化的三种类型:原始方式、写入方式、带头方式
     */
    static enum DataFormat {
    	Raw, Writable, Header
    };
    
    // 监听数据传入请求
    SocketListenThread listenThread;
    
    // 把使用管道处理数据的过程放在Tee线程中
    List<Tee> tees;
    
    // 下一步处理时所使用的writer
    ChukwaWriter next;
    
    /**
     * 启动监听线程，初始化Tee线程列表
     */
    public void init(Configuration c)
    
    /**
     * 1. 将数据块传递给下一个管道处理
     * 2. 遍历Tee线程列表，循环处理数据块
     */
    public CommitStatus add(List<Chunk> chunks)

org.apache.hadoop.chukwa.datacollection.writer.SocketTeeWriter$SocketListenThread  
    监听socket连接线程

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  /\*\*  
2.   \* 以9094端口启动socketserver，端口可通过"chukwaCollector.tee.port"配置 
3.   \*/  
4.  public SocketListenThread(Configuration conf)   
5.    
6.  /\*\*  
7.   \* 接收到请求后交给Tee线程处理  
8.   \*/\*   
9.  public void run()

/**
     * 以9094端口启动socketserver，端口可通过"chukwaCollector.tee.port"配置
     */
    public SocketListenThread(Configuration conf)
    
    /**
     * 接收到请求后交给Tee线程处理
     */*
    public void run()

org.apache.hadoop.chukwa.datacollection.writer.SocketTeeWriter$Tee  
    处理一个socket连接的线程

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  // 使用阻塞队列存储数据块   
2.  final BlockingQueue<Chunk> sendQ;   
3.    
4.  /\*\*  
5.   \* 1. 初始化, 从socket连接中获取输入输出  
6.   \* 2. 从输入中以readline方式读取命令内容  
7.   \* 3. 从命令中获取数据格式化类型，过滤类型，将其添加到SocketTeeWriter的tees中 
8.   \*/  
9.  public void setup()   
10.   
11. /\*\*  
12.  \* 使用过滤器校验数据块，通过校验则添加到阻塞队列  
13.  \*/  
14. public void handle(Chunk c)   
15.   
16. /\*\*  
17.  \* 1. 调用setup方法完成初始化  
18.  \* 2. 根据数据格式化类型进行数据处理  
19.  \*    RAW       以原始方式写入数据，即先写数据长度再写入数据内容  
20.  \*    Writable  以写入方式写入数据，即直接写入数据内容  
21.  \*    Header    以带头方式写入数据，即先构造由数据块属性(来源、类型、流名称、偏移大小)组成的头信息，然后写入长度，再写入数据 
22.  \*/  
23. public void run()

// 使用阻塞队列存储数据块
    final BlockingQueue<Chunk> sendQ;
    
    /**
     * 1. 初始化, 从socket连接中获取输入输出
     * 2. 从输入中以readline方式读取命令内容
     * 3. 从命令中获取数据格式化类型，过滤类型，将其添加到SocketTeeWriter的tees中
     */
    public void setup()
    
    /**
     * 使用过滤器校验数据块，通过校验则添加到阻塞队列
     */
    public void handle(Chunk c)
    
    /**
     * 1. 调用setup方法完成初始化
     * 2. 根据数据格式化类型进行数据处理
     *	  RAW		以原始方式写入数据，即先写数据长度再写入数据内容
     * 	  Writable	以写入方式写入数据，即直接写入数据内容
     *	  Header	以带头方式写入数据，即先构造由数据块属性(来源、类型、流名称、偏移大小)组成的头信息，然后写入长度，再写入数据
     */
    public void run()

org.apache.hadoop.chukwa.datacollection.writer.SeqFileWriter  
    继承自PipelineableWriter,实现了ChukwaWriter接口，其作为SocketTeeWriter的下一个管道(或认为其为管道流的最后一个  管道)

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  /\*\*  
2.   \* 完成writer的初始化  
3.   \* 1. 使用"chukwaCollector.outputDir"设置输出目录, 默认为/chukwa, 一般设为/chukwa/logs 
4.   \* 2. 使用"chukwaCollector.rotateInterval"设置生成文件的时间间隔, 默认为5分钟 
5.   \* 3. 使用"chukwaCollector.stats.period"设置统计写入速度的时间间隔, 默认为30秒 
6.   \* 4. 使用"writer.hdfs.filesystem"设置写入数据时使用的HDFS文件系统的URL, 如果没有设置则使用"fs.default.name", 即hadoop的默认配置 
7.   \* 5. 设置此writer运行状态为true, 启动循环处理文件的定时任务, 启动循环报告统计信息的定时任务  
8.   \*/  
9.  public void init(Configuration conf)   
10.   
11. /\*\*  
12.  \* 循环处理文件的定时任务, 放在TimerTask中循环调用自身  
13.  \* 1. 获取当前时间, 转换为"yyyyddHHmmssSSS"格式, 并于本机名称以及使用java.rmi.server.UID()构造的UID拼接在一起, 构成如下形式的文件 
14.  \*    /chukwa/logs/yyyyddHHmmssSSS\_localHostAddr\_UID.chukwa 
15.  \* 2. 通过信号量控制访问的进程数, 这里的Semaphore lock被设置为只能有一个进程访问 
16.  \*    获取许可    
17.  \*    如果存在长度大于0的文件, 则将其改名为.done文件, 否则将当前文件删除   
18.  \*    使用新生成的文件名创建.chukwa文件, 并更新相应变量的值  
19.  \*    构造序列化写入数据的seqFileWriter, 使用/chukwa/logs做为输出目录, 以ChukwaArchiveKey为Key, 以ChunkImpl为value 
20.  \*    释放许可        
21.  \*/  
22. void rotate()   
23.   
24. /\*\*  
25.  \* 写入数据块内容  
26.  \* 1. 获得许可  
27.  \* 2. 遍历数据块列表  
28.  \*    根据数据块信息设置ChukwaArchiveKey的各个属性值:时间、数据类型、流名称(cluster/source/streamName)、seqID 
29.  \*    使用seqFileWriter写入数据  
30.  \*    将文件名由.chukwa修改为.done  
31.  \* 3. 释放许可  
32.  \*/  
33. public CommitStatus add(List<Chunk> chunks)

[icon_copy.gif]: http://savagegarden.iteye.com/images/icon_copy.gif
[_icon_copy.gif]: http://savagegarden.iteye.com/blog/1441823#
[icon_star.png]: http://savagegarden.iteye.com/images/icon_star.png
[spinner.gif]: /images/20220320/e3db88ba161c4ac9b6f4110ff0900a21.png