Samsa

Rust 原生 Kafka/Redpanda 协议和客户端实现。

此 crate 提供了 Rust 原生的消费者和生产者，以及 Apache Kafka 协议的低级绑定。与使用 FFI 的 librdkafka 的 crate 不同，此 crate 的用户 不需要 C 库，并且将受益于 Rust 的所有优点；这意味着内存安全、安全的并发、低资源使用，以及当然，闪电般的速度。

文档

目标

• 易于理解的代码
• 利用像 Tokio、Nom 这样的最佳类库来完成繁重的工作
• 从坚实的基础开始，并随着时间的推移添加更多高级功能
• 提供 Kafka 协议的纯 Rust 实现
• 成为基于 Kafka 的未来工作的良好构建块

目录

• 入门
  • 生产者
  • 消费者
  • 消费者组
  • TLS 支持
  • 压缩支持
  • SASL 支持
• 示例
• 开发设置
  • 测试
• 基准测试
• 资源

入门

使用 samsa 将 cargo add samsa 将 Cargo.toml 依赖项中的以下片段包含到您的 Rust 项目中

samsa = "0.1"

此项目包含 Docker Compose 文件，以帮助设置 Redpanda 和 Kafka 集群以简化测试。最简单的方法是运行 docker-compose up 以启动一个 2 个代理的 Redpanda 集群。如果您想使用不同的 Kafka 版本，请参阅 DockerCompose.README.md

生产者

生产者向指定的主题和分区发送消息。

它是缓存的，具有超时和体积阈值，当达到时清除缓冲区。这就是如何调整 letency 和 throughout 以实现所需的速率。

要创建一个实例，最简单的方法是使用Stream和ProducerBuilder。

use samsa::prelude::*;

let bootstrap_addrs = vec![BrokerAddress {
        host: "127.0.0.1".to_owned(),
        port: 9092,
    }];
let topic_name = "my-topic".to_string();
let partition_id = 0;

// create a stream of 5k messages in batches of 100
let stream = iter(0..5000).map(|_| ProduceMessage {
    topic: topic_name.to_string(),
    partition_id,
    key: Some(bytes::Bytes::from_static(b"Tester")),
    value: Some(bytes::Bytes::from_static(b"Value")),
    headers: vec![
        samsa::prelude::Header::new(String::from("Key"), bytes::Bytes::from("Value"))
    ],
}).chunks(100);

let output_stream =
ProducerBuilder::<TcpConnection>::new(bootstrap_addrs, vec![topic_name.to_string()])
    .await?
    .batch_timeout_ms(100)
    .max_batch_size(100)
    .clone()
    .build_from_stream(stream)
    .await;

tokio::pin!(output_stream);
while (output_stream.next().await).is_some() {}

消费者

Consumer用于从代理中获取消息。它是一个可以配置为自动提交的异步迭代器。要创建一个实例，从ConsumerBuilder开始。

use samsa::prelude::*;

let bootstrap_addrs = vec![BrokerAddress {
        host: "127.0.0.1".to_owned(),
        port: 9092,
    }];
let partitions = vec![0];
let topic_name = "my-topic".to_string();
let assignment = TopicPartitionsBuilder::new()
    .assign(topic_name, partitions)
    .build();

let consumer = ConsumerBuilder::<TcpConnection>::new(
        bootstrap_addrs,
        assignment,
    )
    .await?
    .build();

let stream = consumer.into_stream();
// have to pin streams before iterating
tokio::pin!(stream);

// Stream will do nothing unless consumed.
while let Some(Ok((batch, offsets))) = stream.next().await {
    println!("{:?}", batch);
}

消费者组

您可以使用组ID和分配来设置一个消费者组。组成员的偏移量会自动提交。

use samsa::prelude::*;

let bootstrap_addrs = vec![BrokerAddress {
        host: "127.0.0.1".to_owned(),
        port: 9092,
    }];
let partitions = vec![0];
let topic_name = "my-topic".to_string();
let assignment = TopicPartitionsBuilder::new()
    .assign(topic_name, partitions)
    .build();
let group_id = "The Data Engineering Team".to_string();

let consumer_group_member = ConsumerGroupBuilder::<TcpConnection>::new(
        bootstrap_addrs,
        group_id,
        assignment,
    ).await?
    .build()
    .await?;

let stream = consumer_group_member.into_stream();
// have to pin streams before iterating
tokio::pin!(stream);
 
// Stream will do nothing unless consumed.
while let Some(batch) = stream.next().await {
    println!("{:?}", batch);
}

TLS 支持

您可以为消费者或生产者添加TLS支持以实现安全通信。要启用此功能，从指定TlsConnectionOptions开始，并将其传递给ProducerBuilder或ConsumerBuilder的一个实例。

TLS支持的生产者示例

use samsa::prelude::*;

let tls_option = TlsConnectionOptions {
        broker_options: vec![BrokerAddress {
          host: "127.0.0.1".to_owned(),
          port: 9092,
        }],
        key: "/path_to_key_file".into(),
        cert: "/path_to_cert_file".into(),
        cafile: Some("/path_to_ca_file".into()),
    };
let topic_name = "my-topic".to_string();
let partition_id = 0;

let message = ProduceMessage {
        topic: topic_name.to_string(),
        partition_id,
        key: Some(bytes::Bytes::from_static(b"Tester")),
        value: Some(bytes::Bytes::from_static(b"Value")),
        headers: vec![
            Header::new(String::from("Key"), bytes::Bytes::from("Value"))
        ],
    };

let producer_client = ProducerBuilder::<TlsConnection>::new(
        tls_option, 
        vec![topic_name.to_string()]
    )
    .await?
    .batch_timeout_ms(1)
    .max_batch_size(2)
    .clone()
    .build()
    .await;

producer_client
    .produce(message)
    .await;

TLS支持的消费者示例

use samsa::prelude::*;

let tls_option = TlsConnectionOptions {
        broker_options: vec![BrokerAddress {
            host: "127.0.0.1".to_owned(),
            port: 9092,
        }],
        key: "/path_to_key_file".into(),
        cert: "/path_to_cert_file".into(),
        cafile: Some("/path_to_ca_file".into()),
    };
let partitions = vec![0];
let topic_name = "my-topic".to_string();
let assignment = TopicPartitionsBuilder::new()
    .assign(topic_name, partitions)
    .build();

let consumer = ConsumerBuilder::<TlsConnection>::new(
        tls_option,
        assignment,
    )
    .await?
    .build();

let stream = consumer.into_stream();
// have to pin streams before iterating
tokio::pin!(stream);

// Stream will do nothing unless consumed.
while let Some(batch) = stream.next().await {
    println!("{:?} messages read", batch.unwrap().count());
}

压缩支持

我们为生产者提供压缩支持，使用Compression枚举。枚举允许指定要使用的压缩类型。消费者将自动知道解压缩消息。

TLS和GZIP压缩支持的生产者示例

use samsa::prelude::*;

let tls_option = TlsConnectionOptions {
        broker_options: vec![BrokerAddress {
            host: "127.0.0.1".to_owned(),
            port: 9092,
        }],
        key: "/path_to_key_file".into(),
        cert: "/path_to_cert_file".into(),
        cafile: Some("/path_to_ca_file".into()),
    };
let topic_name = "my-topic".to_string();
let partition_id = 0;

let message = ProduceMessage {
        topic: topic_name.to_string(),
        partition_id,
        key: Some(bytes::Bytes::from_static(b"Tester")),
        value: Some(bytes::Bytes::from_static(b"Value")),
        headers: vec![
            Header::new(String::from("Key"), bytes::Bytes::from("Value"))
        ],
    };

let producer_client = ProducerBuilder::new(tls_option, vec![topic_name.to_string()])
    .await?
    .compression(Compression::Gzip)
    .batch_timeout_ms(1)
    .max_batch_size(2)
    .clone()
    .build()
    .await;

producer_client
    .produce(message)
    .await;

SASL 支持

我们包括支持SASL的所有典型机制：PLAIN、SCRAM-SHA-256、SCRAM-SHA-512。这表示为另一种类型的BrokerConnection，我们的消费者和生产者将其作为泛型参数接收。所需的一切只是提供凭证。

同时使用TLS和SASL的生产者示例

use samsa::prelude::*;

let tls_config = TlsConnectionOptions {
    broker_options: vec![BrokerAddress {
        host: "127.0.0.1".to_owned(),
        port: 9092,
    }],
    key: "/path_to_key_file".into(),
    cert: "/path_to_cert_file".into(),
    cafile: Some("/path_to_ca_file".into()),
};

let sasl_config = SaslConfig::new(String::from("myuser"), String::from("pass1234"), None, None);

let options = SaslTlsConfig {
    tls_config,
    sasl_config,
};

let topic_name = "atopic";

let s = ConsumerBuilder::<SaslTlsConnection>::new(
    options,
    TopicPartitionsBuilder::new()
        .assign(topic_name.to_owned(), vec![0])
        .build(),
)
.await
.unwrap()
.build()
.into_stream();

tokio::pin!(s);

while let Some(m) = s.next().await {
    tracing::info!("{:?} messages read", m.unwrap().count());
}

示例

我们为有兴趣尝试samsa的人提供高级示例。设置如下

1. 使用docker-compose up启动您的redpanda集群。
2. 转到https://127.0.0.1:8080/topics以查看redpanda控制台。
3. 创建一个名为my-topic的具有4个分区的主题。

time:   [1.5852 s 1.6119 s 1.6406 s]

time:   [1.8013 s 1.8112 s 1.8225 s]