79个版本 (22个破坏性更新)
0.26.2 | 2024年4月2日 |
---|---|
0.26.1 | 2024年2月3日 |
0.26.0 | 2023年10月11日 |
0.22.6 | 2023年7月3日 |
0.3.0 | 2020年7月17日 |
#3 in #message-broker
1MB
29K SLoC
什么是Netidx
Netidx是一种中间件,它允许在一个程序中发布一个值(如42),并在另一个程序中消费它,无论是在同一台机器上还是在网络上。
值在层次命名空间中具有全局唯一的名称。例如,我们发布的42可能被命名为/the-ultimate-answer(通常我们不会在根目录下直接放置值,但在此情况下这是合适的)。网络上的任何其他程序都可以通过该名称引用42,并在/the-ultimate-answer更改的情况下接收更新。
与其他系统的比较
-
类似于LDAP
- Netidx跟踪值的层次目录
- Netidx在一定程度上可浏览和可查询
- Netidx支持身份验证、授权和加密
- Netidx值既可以写入也可以读取。
- 可以通过在较小系统之间添加引用来构建更大的Netidx系统。解析器服务器集群可以有父级和子级。
-
与LDAP不同
- 在Netidx中,解析器服务器(如slapd)只跟踪拥有数据的发布者的位置,而不是数据本身。
- 没有'条目'、'属性'、'ldif记录'等。系统中的每个名称要么是结构性的,要么是单个值。使用层次结构创建类似条目的结构。因此,也没有模式检查。
- 可以订阅一个值,如果它更改,则会立即通知。
- 没有对数据的全局过滤器,例如,您不能查询(&(cn=bob)(uid=foo)),因为Netidx不是数据库。是否存在以及查询机制取决于发布者。然而,您可以查询结构,例如,/foo/**/bar将返回任何以bar结尾的foo下的路径。
-
类似于MQTT
- Netidx值是发布/订阅
- 单个Netidx值可以有多个订阅者
- 当订阅的值更改时,所有Netidx订阅者都会收到更新。
- Netidx消息传递是可靠和有序的。
-
与MQTT不同
- Netidx中没有集中的消息代理。消息直接通过TCP从发布者流向订阅者。解析服务器仅存储发布者/发布值的地址。
有关更多详细信息,请参阅netidx书籍
以下是一个示例服务,它发布CPU温度,以及相应的消费者,用于消费数据。
发布者
use netidx::{
publisher::{Publisher, Value, BindCfg},
config::Config,
resolver::Auth,
path::Path,
};
use tokio::time;
use std::time::Duration;
use anyhow::Result;
fn get_cpu_temp() -> f32 { 42. }
async fn run() -> Result<()> {
// load the site cluster config. You can also just use a file.
let cfg = Config::load_default()?;
// no authentication (kerberos v5 is the other option)
// listen on any unique address matching 192.168.0.0/16
let publisher = Publisher::new(cfg, Auth::Anonymous, "192.168.0.0/16".parse()?).await?;
let temp = publisher.publish(
Path::from("/hw/washu-chan/cpu-temp"),
Value::F32(get_cpu_temp())
)?;
loop {
time::sleep(Duration::from_millis(500)).await;
let mut batch = publisher.start_batch();
temp.update(&mut batch, Value::F32(get_cpu_temp()));
batch.commit(None).await;
}
Ok(())
}
订阅者
use netidx::{
subscriber::{Subscriber, UpdatesFlags},
config::Config,
resolver::Auth,
path::Path,
};
use futures::{prelude::*, channel::mpsc};
use anyhow::Result;
async fn run() -> Result<()> {
let cfg = Config::load_default()?;
let subscriber = Subscriber::new(cfg, Auth::Anonymous)?;
let path = Path::from("/hw/washu-chan/cpu-temp");
let temp = subscriber.subscribe_one(path, None).await?;
println!("washu-chan cpu temp is: {:?}", temp.last());
let (tx, mut rx) = mpsc::channel(10);
temp.updates(UpdatesFlags::empty(), tx);
while let Some(mut batch) = rx.next().await {
for (_, v) in batch.drain(..) {
println!("washu-chan cpu temp is: {:?}", v);
}
}
Ok(())
}
发布的内容始终具有值,新订阅者最初接收这些值。此后,订阅成为一个无损失的有序流,就像TCP连接一样,只是传输的单位不是字节,而是publisher::Value
。由于订阅者可以将值写回发布者,因此连接是双向的,就像Tcp流一样。
值包括许多有用的基本类型,包括零拷贝的字节缓冲区(使用出色的bytes crate),因此您可以使用netidx高效地发送任何类型的消息。但是,建议坚持使用基本类型,并使用多个发布值在层次结构中表达结构,因为这可以使您的系统更易于发现,并且也非常高效。
netidx包含对kerberos v5(包括Active Directory)的可选支持。如果启用,所有组件将在解析器、订阅者和发布者之间执行相互认证,以及加密所有传输中的数据。
在krb5模式下,解析服务器维护并强制执行整个命名空间的一组授权权限。系统管理员可以集中执行谁可以在哪里发布,以及谁可以订阅什么。
依赖项
~37–70MB
~1M SLoC