InfluxDB是一个由InfluxData开发的开源时序数据库，专一于海量时序数据的高性能读、写、高效存储与实时分析等，在DB-Engines Ranking时序型数据库排行榜上常年排名第一。数据库
InfluxDB能够说是当之无愧的佼佼者，但 InfluxDB CTO Paul 在 2020/12/10 号在博客中发表一篇名为：Announcing InfluxDB IOx – The Future Core of InfluxDB Built with Rust and Arrow的文章，介绍了一个新项目 InfluxDB IOx，InfluxDB 的下一代时序引擎。微信
接下来，我将连载对于InfluxDB IOx的源码解析过程，欢迎各位批评指正，联系方式见文章末尾。ide
上篇介绍到：InfluxDB-IOx的环境搭建,详情见：https://my.oschina.net/u/3374539/blog/5016798oop
本章开始，讲解启动的主流程！性能
打开src/main.rs文件能够找到下面的代码学习
fn main() -> Result<(), std::io::Error> {
// load all environment variables from .env before doing anything
load_dotenv();
let config = Config::from_args();
println!("{:?}", config);
//省略
.....
Ok(())
}
在main方法中映入眼帘的第一行就是load_dotenv()方法，而后是Config::from_args()接下来就分别跟踪这两个方法，看明白是怎么工做的。ui
加载配置文件
在README文件中，咱们能够看到这样一行：google
Should you desire specifying config via a file, you can do so using a .env formatted file in the working directory. You can use the provided example as a template if you want:url
cp docs/env.example .env
意思就是这个工程使用的配置文件，名字是.env。了解这个特殊的名字以后，咱们看代码src/main.rs:276：.net
fn load_dotenv() {
//调用dotenv方法，并对其返回值进行判断
match dotenv() {
//若是返回成功，程序什么都不作，继续执行。
Ok(_) => {}
//返回的是错误，那么判断一下是否为'未找到'错误，
//若是是未找到，那么就什么都不作（也就是有默认值填充）
Err(dotenv::Error::Io(err)) if err.kind() == std::io::ErrorKind::NotFound => {
}
//这里就是真真正正必需要处理的错误了，直接退出程序
Err(e) => {
eprintln!("FATAL Error loading config from: {}", e);
eprintln!("Aborting");
std::process::exit(1);
}
};
}
而后跟踪dotenv()方法看看如何执行(这里就进入了dotenv这个crate了)： 为了方便写，我就直接把全部调用，从上到下的顺序全都写出来了
//返回一个PathBuf的Result，以后再看这个Result
pub fn dotenv() -> Result<PathBuf> {
//new一个Finder结构并调用find方法
//?表明错误的时候直接抛出错误
let (path, iter) = Finder::new().find()?;
//返回一个自定义的Iter结构，并调用load方法
iter.load()?;
//成功返回
Ok(path)
}
//建立一个Finder结构体，filename使用`.env`填充
pub fn new() -> Self {
Finder {
filename: Path::new(".env"),
}
}
//返回一个元组，多个返回值，(路径，文件读取相关记录)
pub fn find(self) -> Result<(PathBuf, Iter<File>)> {
//使用标准库中的current_dir()方法获得当前的路径
//出错就返回Error::Io错误，正常就调用find方法
let path = find(&env::current_dir().map_err(Error::Io)?, self.filename)?;
//若是找到了.env文件就打开，打开错误就返回Error::Io错误
let file = File::open(&path).map_err(Error::Io)?;
//使用打开的文件建立一个Iter的结构
let iter = Iter::new(file);
//返回
Ok((path, iter))
}
//递归查找.env文件
pub fn find(directory: &Path, filename: &Path) -> Result<PathBuf> {
//拼装一个全路径
let candidate = directory.join(filename);
//尝试打开这个文件
match fs::metadata(&candidate) {
//成功打开了，说明找到了.env文件，就返回成功
//但我有个疑问文件内容为啥不校验一下呢？
Ok(metadata) => if metadata.is_file() {
return Ok(candidate);
},
//除了没找到文件的错误以外，其它错误都直接返回异常
Err(error) => {
if error.kind() != io::ErrorKind::NotFound {
return Err(Error::Io(error));
}
}
}
//没找到的时候，就返回到父级文件夹里，继续找，一直到根文件夹
if let Some(parent) = directory.parent() {
find(parent, filename)
} else {
//一直到根文件夹，还没找到就返回一个NotFound的IO错误，
//这个在上面的代码中提到，这个错误会被忽略
Err(Error::Io(io::Error::new(io::ErrorKind::NotFound, "path not found")))
}
}
//对应的iter.load()?;方法实现
pub fn load(self) -> Result<()> {
//可使用for是由于实现了Iterator 这个trait
for item in self {
//获取读取出来的一行一行的配置项
let (key, value) = item?;
//验证key没有什么问题，就放到env中
if env::var(&key).is_err() {
env::set_var(&key, value);
}
}
Ok(())
}
// 为了可以for循环，实现的Iterator
impl<R: Read> Iterator for Iter<R> {
type Item = Result<(String, String)>;
fn next(&mut self) -> Option<Self::Item> {
loop {
//一行一行的读取文件内容
let line = match self.lines.next() {
Some(Ok(line)) => line,
Some(Err(err)) => return Some(Err(Error::Io(err))),
None => return None,
};
//解析配置项目，这里就不在深刻跟了
match parse::parse_line(&line, &mut self.substitution_data) {
Ok(Some(result)) => return Some(Ok(result)),
Ok(None) => {}
Err(err) => return Some(Err(err)),
}
}
}
}
研究这里的时候，我发现了一个比较好玩儿的东西就是返回值的Result<PathBuf>。标准库的定义中，Result是有两个值，分别是<T,E>。
自定义的类型，节省了Error这个模板代码
pub type Result<T> = std::result::Result<T, Error>;
//Error也本身定义
pub enum Error {
LineParse(String, usize),
Io(io::Error),
EnvVar(std::env::VarError),
#[doc(hidden)]
__Nonexhaustive
}
//实现一个not_found()的方法来判断是否为not_found的一个错误类型
impl Error {
pub fn not_found(&self) -> bool {
if let Error::Io(ref io_error) = *self {
return io_error.kind() == io::ErrorKind::NotFound;
}
false
}
}
//实现标准库中的error::Error这个trait
impl error::Error for Error {
//追踪错误的上一级，应该是打印堆栈这种功能
//若是内部有错误类型Err返回：Some(e),若是没有返回：None
//关于'static这个生命周期的标注，我也不是很理解
//是指存储的错误生命周期足够长仍是什么？
fn source(&self) -> Option<&(dyn error::Error + 'static)> {
match self {
Error::Io(err) => Some(err),
Error::EnvVar(err) => Some(err),
_ => None,
}
}
}
//实现错误的打印
impl fmt::Display for Error {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
match self {
Error::Io(err) => write!(fmt, "{}", err),
Error::EnvVar(err) => write!(fmt, "{}", err),
Error::LineParse(line, error_index) => write!(fmt, "Error parsing line: '{}', error at line index: {}", line, error_index),
_ => unreachable!(),
}
}
}
更详细的rust错误处理，能够参见：https://zhuanlan.zhihu.com/p/109242831
命令行参数
在main方法中咱们能够看到第二行，
let config = Config::from_args();
这是influx使用了structopt这个crate，调用该方法后，程序会根据结构体上的#[structopt()]中的参数进行执行命令行解析。
#[derive(Debug, StructOpt)]
#[structopt(
//cargo的crate名字
name = "influxdb_iox",
//打印出来介绍
about = "InfluxDB IOx server and command line tools",
long_about = // 省略 ...
)]
struct Config {
// from_occurrences表明出现了几回，就是-vvv的时候v出现的次数
#[structopt(short, long, parse(from_occurrences))]
verbose: u64,
#[structopt(
short,
long,
global = true,
env = "IOX_ADDR",
default_value = "http://127.0.0.1:8082"
)]
host: String,
#[structopt(long)]
num_threads: Option<usize>,
//subcommand表明是一个子类型的，
//具体还有什么命令行要去子类型里继续解析，
//这个字段不展现在命令行中
#[structopt(subcommand)]
command: Command,
}
//在influx的命令行中提供了8个主要的命令，
//在上一章中使用到的run参数就是属于Run(Box<commands::run::Config>)里的调用。
//这里都是subcommand，须要继续解析，这个在之后学习每一个具体功能的时候再分析
#[derive(Debug, StructOpt)]
enum Command {
Convert { // 省略 ...},
Meta {// 省略 ...},
Database(commands::database::Config),
Run(Box<commands::run::Config>),
Stats(commands::stats::Config),
Server(commands::server::Config),
Writer(commands::writer::Config),
Operation(commands::operations::Config),
}
下面经过打印出来的例子来对应structopt中的内容。
$ ./influxdb_iox -vvvv run
Config { verbose: 4, host: "http://127.0.0.1:8082", num_threads: None, command: Run(Config { rust_log: None, log_format: None, verbose_count: 0, writer_id: None, http_bind_address: 127.0.0.1:8080, grpc_bind_address: 127.0.0.1:8082, database_directory: None, object_store: None, bucket: None, aws_access_key_id: None, aws_secret_access_key: None, aws_default_region: "us-east-1", google_service_account: None, azure_storage_account: None, azure_storage_access_key: None, jaeger_host: None }) }
能够看到，咱们执行了Run这个变体的Subcommand，而且指定了Config结构体中的verbose 4 次，IOx也成功的识别了。
后面继续学习程序的启动过程，祝玩儿的开心！
欢迎关注微信公众号：
或添加微信好友： liutaohua001
