Redis Stream

一个用于从Redis流中读取数据的Rust高级库。

本项目是将Elixir的Redix.Stream库稍作修改后移植到Rust，并作为redis-rs的扩展。

我们在Klaxit中使用它来处理来自Heroku的Logplex的合并日志流，并在用户注意到之前自动修复一些在Heroku上偶尔出现的问题。

当需要时，我们还用它来扩展一些服务。

它已成功运行超过6个月。

安装

该Crate名为redis-stream，您可以通过Cargo依赖它

[dependencies]
redis-stream = "0.1.2"

文档

该库的文档可以在docs.rs/redis-stream找到。

基本用法

use redis_stream::consumer::{Consumer, ConsumerOpts, Message};

let redis_url =
  std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string());

let mut redis = redis::Client::open(redis_url)
  .expect("client")
  .get_connection()
  .expect("connection");

// Message handler
let handler = |_id: &str, message: &Message| {
  // do something
  Ok(())
};

// Consumer config
let opts = ConsumerOpts::default();
let mut consumer = Consumer::init(&mut redis, "my-stream", handler, opts).expect("consumer");

// Consume some messages through handler.
consumer.consume().expect("consume messages");

// Clean up redis
use redis::Commands;
redis.del::<&str, bool>("my-stream").expect("del");

消费者组用法

use redis_stream::consumer::{Consumer, ConsumerOpts, Message};

let redis_url =
  std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string());

let mut redis = redis::Client::open(redis_url)
  .expect("client")
  .get_connection()
  .expect("connection");

// Message handler
let handler = |_id: &str, message: &Message| {
  // do something
  Ok(())
};

// Consumer config
let opts = ConsumerOpts::default().group("my-group", "worker.1");
let mut consumer = Consumer::init(&mut redis, "my-stream-2", handler, opts).unwrap();

// Consume some messages through handler.
consumer.consume().expect("consume messages");

// Clean up redis
use redis::Commands;
redis.xgroup_destroy::<&str, &str, bool>("my-stream-2", "my-group").expect("xgroup destroy");
redis.del::<&str, bool>("my-stream-2").expect("del");

开发

如果您想开发该库，makefile提供了一些命令。

运行make help获取更多信息。

对于测试，如果需要启动本地Redis实例，也提供了一个docker-compose.yml文件

$ docker-compose up -d
$ make test

许可证

请参阅LICENSE