spring-stream插件发布了

spring-stream插件是实时流式处理消息的编程工具，可以大大简化基于文件、redis stream、kafka的消息处理

下面是一个简单的生产者：

#[auto_config(WebConfigurator)] #[tokio::main] async fn main() {     App::new()         .add_plugin(StreamPlugin)         .add_plugin(WebPlugin)         .run()         .await }  #[get("/")] async fn send_msg(Component(producer): Component<Producer>) -> Result<impl IntoResponse> {     let now = SystemTime::now();     let json = json!({         "success": true,         "msg": format!("This message was sent at {:?}", now),     });     let resp = producer         .send_json("topic", json)         .await         .context("send msg failed")?;      let seq = resp.sequence();     Ok(Json(json!({"seq":seq}))) } 

Producer用于向消息存储发送消息。spring-stream底层使用sea-streamer实现，它抽象了file、stdio、redis、kafka消息存储层，使开发者可以用统一的接口来发送和处理消息。

下面是一个简单的消费者的代码：

#[tokio::main] async fn main() {     App::new()         .add_plugin(StreamPlugin)         .add_consumer(consumers())         .run()         .await }  fn consumers() -> Consumers {     Consumers::new().typed_consumer(listen_topic_do_something) }  #[stream_listener("topic")] async fn listen_topic_do_something(Json(payload): Json<Payload>) {     tracing::info!("{:#?}", payload);     // do something } 

点击这里可以查看相关文档。