nats 的消息调度

nats 2.12.2 支持消息调度了，但是目前实现上并不是很全，已经可以支持一些玩法了，以下简单说明下

特点

• 首先是基于jetstream 的，我们需要创建一个jetstream
• jetstream 需要开启--allow-schedules，--allow-msg-ttl 推荐也开启
• publish 消息的时候需要制定策略，以及配置source 以及target，当前似乎都应该是存在以及stream 模式的（待详细确定）
• 调度策略暂时不如规划上的完整，已经支持简单的时间点了@at 模式
• 如果时间比计划执行的早（或者说就是临时消息），调度会首先执行一次
• 调度实际是在特定时间，从source 获取数据，然后到target，对于接受者可以收到数据
• 从测试上source 实际上并不是必须的，如果我们发布的时候指定了一个subject 实际会获取此数据到target

参考代码

• 消费者

import {connect} from "@nats-io/transport-node"
import { jetstream, jetstreamManager } from "@nats-io/jetstream";const nc = await connect({servers:"nats://127.0.0.1:4222"
})const jsm = await jetstreamManager(nc);await jsm.streams.add({name:"appdemov2",subjects:["ordersv3.>"],retention:"limits",max_consumers:-1,max_msgs:-1,allow_msg_schedules:true,allow_msg_ttl:true
})await jsm.consumers.add("appdemov2",{durable_name:"appdemo_consumer",ack_policy:"explicit",deliver_policy:"all"
})const jsc = await jetstream(nc);const c = await jsc.consumers.get("appdemov2","appdemo_consumer");const messages = await c.consume();for await (const m of messages) {console.log(`received message: ${m.subject} - ${m.data}`);m.ack();
}

• 发布

import {connect} from "@nats-io/transport-node"
import { jetstream } from "@nats-io/jetstream";const nc = await connect({servers:"nats://127.0.0.1:4222"
})const jsc = await jetstream(nc,{
});jsc.publish("ordersv3.appdemo222",Buffer.from("demo"),{schedule: {specification: "@at 2025-11-23T03:38:00Z",target: "ordersv3.appdemov2",}
}); 

说明

目前来说nats 的消息调度是一个很不错的选择，但是目前还是欠缺一些能力，是一个值得期待完善的功能

参考资料

https://github.com/nats-io/nats-architecture-and-design/blob/main/adr/ADR-51.md

https://docs.nats.io/release-notes/whats_new/whats_new_212