Flume源码分析–flume-ng解析
以此篇作为研究、学习源码的起步。
Flume简介
Flume 是一个高可用的、高可靠的分布式系统，可以从不同的数据源进行日志的采集、聚合、传输，可以将采集到的日志传输到HDFS、HBase、MySQL等。数据源和接收数据的端都是可定制的。
Flume本地化调试
1、安装java JDK。
2、从FLume官网下载编译好的flume包和flume源码包，并进行解压（我这里是flume1.8.0）。
3、编写一份简单的flume配置文件（我这里是flume.conf）：
a1.sources = r1
a1.channels = c1
a1.sinks = s1
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444
a1.sinks.s1.type = logger
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.sources.r1.channels = c1
a1.sinks.s1.channel = c1
4、启动flume，验证flume是否搭建成功：
bin/flume-ng agent --conf ./conf/ -f conf/flume.conf -Dflume.root.logger=DEBUG,console -n a1
5、将源码包进行解压，导入IDEA中，加载相关的maven依赖。
6、flume agent 启动是用的flume-ng-nodezhogn de org.apache.flume.node.Application.java, 配置 Application.java 的运行参数：
7、编译源码：mvn clean package -DskipTests
8、将编译后的target中的avro文件拷贝到 flume-ng-sdk中的 org.apache.flume.source.avro目录下。
9、将之前编写好的flume.conf文件复制到flume-ng-node的根目录下，复制log4j.properties复制到flume-ng-node中的src.main.java目录下。
10、debug Application.java
flume-ng脚本
之前启动flume的命令是：
bin/flume-ng agent --conf ./conf/ -f conf/flume.conf -Dflume.root.logger=DEBUG,console -n a1
flume-ng 是启动flume的入口，直接从flume-ng的入口开始解析(截取部分脚本分析)
#!/bin/bash
mode=$1 # 获取flume的启动模式
shift
case "$mode" in
help)
display_help # 调用display_play方法
exit 0
;;
agent)
opt_agent=1
;;
node)
opt_agent=1
warn "The \"node\" command is deprecated. Please use \"agent\" instead."
;;
avro-client)
opt_avro_client=1
;;
tool)
opt_tool=1
;;
version)
opt_version=1
CLEAN_FLAG=0
;;
*)
error "Unknown or unspecified command '$mode'"
echo
display_help
exit 1
;;
esac
根据脚本flume匹配的有5中模式：
1.agent | node ; 启动类：FLUME_AGENT_CLASS=“org.apache.flume.node.Application”。（官网推荐使用agent，node的效果和agent一样）
2. avro-client ; 启动类：FLUME_AVRO_CLIENT_CLASS=“org.apache.flume.client.avro.AvroCLIClient”
3. help ; 调用 display_help 方法，打印flume help 信息 （-h 可以跳过help，但在主类中也是打印help信息），然后退出
4. tool ; 启动类：FLUME_TOOLS_CLASS=“org.apache.flume.tools.FlumeToolsMain”
5. version ; 启动类：FLUME_VERSION_CLASS=“org.apache.flume.tools.VersionInfo”
#!/bin/bash
args=""
while [ -n "$*" ] ; do
arg=$1
shift
case "$arg" in
--conf|-c)
[ -n "$1" ] || error "Option --conf requires an argument" 1
opt_conf=$1 # opt_conf=./conf
shift
;;
--classpath|-C)
[ -n "$1" ] || error "Option --classpath requires an argument" 1
opt_classpath=$1
shift
;;
--dryrun|-d)
opt_dryrun="1"
;;
--plugins-path)
opt_plugins_dirs=$1
shift
;;
-agentlib*)
arr_java_props[arr_java_props_ct]=$arg
((++arr_java_props_ct))
;;
-agentpath*)
arr_java_props[arr_java_props_ct]=$arg
((++arr_java_props_ct))
;;
-javaagent*)
arr_java_props[arr_java_props_ct]=$arg
((++arr_java_props_ct))
;;
-D*)
arr_java_props[arr_java_props_ct]=$arg # arr_java_props[0] = -Dflume.root.logger=DEBUG,console
((++arr_java_props_ct))
;;
-X*)
arr_java_props[arr_java_props_ct]=$arg
((++arr_java_props_ct))
;;
*)
args="$args $arg" # args: "-f conf/flume.conf -n agent1"
;;
esac
done
接着进行配置初始化
1.初始化 flume启动时需要的运行信息：（1.配置文件所在的目录 2. flume运行的环境变量 3. 是否启动flume(dryrun时不会启动flume，只打印一些启动信息) … ）
2.获取flume启动类对应的参数(agent对应的配置文件|avro的端口信息等)
#!/bin/bash
# 初始化opt_conf的值
if [[ -n "$opt_conf" && -d "$opt_conf" ]]; then
opt_conf=$(cd $opt_conf; pwd)
fi
# 传入的参数 会覆盖 flume-env.sh的内容
if [ -z "$opt_conf" ]; then
warn "No configuration directory set! Use --conf <dir> to override."
elif [ -f "$opt_conf/flume-env.sh" ]; then
info "Sourcing environment configuration script $opt_conf/flume-env.sh"
# 刷新配置文件
source "$opt_conf/flume-env.sh"
fi
根据脚本可以发现：通过启动flume命令传入的参数可以覆盖flume-env.sh中的配置项
#!/bin/bash
if [ -n "${opt_classpath}" ]; then
if [ -n "${FLUME_CLASSPATH}" ]; then
FLUME_CLASSPATH="${opt_classpath}:${FLUME_CLASSPATH}"
else
FLUME_CLASSPATH="${opt_classpath}"
fi
fi
# 判断 配置的环境变量中的 flume的home路径
if [ -z "${FLUME_HOME}" ]; then
FLUME_HOME=$(cd $(dirname $0)/..; pwd)
fi
# prepend $FLUME_HOME/lib jars to the specified classpath (if any)
# 将flume_home/lib/* 加入到环境变量的前边
if [ -n "${FLUME_CLASSPATH}" ] ; then
FLUME_CLASSPATH="${FLUME_HOME}/lib/*:$FLUME_CLASSPATH"
else
FLUME_CLASSPATH="${FLUME_HOME}/lib/*"
fi
# 加载 plugins.d directories
PLUGINS_DIRS=""
if [ -n "${opt_plugins_dirs}" ]; then
PLUGINS_DIRS=$(sed -e 's/:/ /g' <<<${opt_plugins_dirs})
else
PLUGINS_DIRS="${FLUME_HOME}/plugins.d"
fi
unset plugin_lib plugin_libext plugin_native
for PLUGINS_DIR in $PLUGINS_DIRS; do
if [[ -d ${PLUGINS_DIR} ]]; then
for plugin in ${PLUGINS_DIR}/*; do
if [[ -d "$plugin/lib" ]]; then
plugin_lib="${plugin_lib}${plugin_lib+:}${plugin}/lib/*"
fi
if [[ -d "$plugin/libext" ]]; then
plugin_libext="${plugin_libext}${plugin_libext+:}${plugin}/libext/*"
fi
if [[ -d "$plugin/native" ]]; then
plugin_native="${plugin_native}${plugin_native+:}${plugin}/native"
fi
done
fi
done
if [[ -n "${plugin_lib}" ]]
then
FLUME_CLASSPATH="${FLUME_CLASSPATH}:${plugin_lib}"
fi
if [[ -n "${plugin_libext}" ]]
then
FLUME_CLASSPATH="${FLUME_CLASSPATH}:${plugin_libext}"
fi
if [[ -n "${plugin_native}" ]]
then
if [[ -n "${FLUME_JAVA_LIBRARY_PATH}" ]]
then
FLUME_JAVA_LIBRARY_PATH="${FLUME_JAVA_LIBRARY_PATH}:${plugin_native}"
else
FLUME_JAVA_LIBRARY_PATH="${plugin_native}"
fi
fi
加载flume启动时需要的环境变量，同时还加载JAVA、hadoop、HBASE、hive的环境信息。
# allow dryrun
EXEC="exec"
if [ -n "${opt_dryrun}" ]; then
warn "Dryrun mode enabled (will not actually initiate startup)"
EXEC="echo"
fi
在flume启动的时候，如果命令中包含--dryrun,脚本在执行的时候则打印初始化好的变量、环境信息，不会启动flume，如果没有指定该命令，则启动flume
# 根据不同的模式进行反射调用不同的类
if [ -n "$opt_agent" ] ; then
run_flume $FLUME_AGENT_CLASS $args # args: "-f conf/flume.conf -n agent1"
elif [ -n "$opt_avro_client" ] ; then
run_flume $FLUME_AVRO_CLIENT_CLASS $args
elif [ -n "${opt_version}" ] ; then
run_flume $FLUME_VERSION_CLASS $args
elif [ -n "${opt_tool}" ] ; then
run_flume $FLUME_TOOLS_CLASS $args
else
error "This message should never appear" 1
fi
# 启动flume的函数
run_flume() {
local FLUME_APPLICATION_CLASS
if [ "$#" -gt 0 ]; then
FLUME_APPLICATION_CLASS=$1
shift
else
error "Must specify flume application class" 1
fi
if [ ${CLEAN_FLAG} -ne 0 ]; then
set -x
fi
# 命令行中指定dryrun时，$EXEC为echo，打印信息，否则为exec，启动flume
$EXEC $JAVA_HOME/bin/java $JAVA_OPTS $FLUME_JAVA_OPTS "${arr_java_props[@]}" -cp "$FLUME_CLASSPATH" \
-Djava.library.path=$FLUME_JAVA_LIBRARY_PATH "$FLUME_APPLICATION_CLASS" $*
}
以上截取了flume-ng中的部分脚本进行启动flume流程的分析。
总结
通过分析 flume-ng脚本，掌握了flume-ng的执行流程，对flume启动命令传入的参数有了清晰的认识（文中只列举了部分参数），同时对linux中的命令进行了回顾，再接再厉。