pub struct Receiver<T> { /* 私有字段 */ }展开描述
实现§
源代码§impl<T> Receiver<T>
impl<T> Receiver<T>
源代码pub fn borrow(&self) -> Ref<'_, T>
pub fn borrow(&self) -> Ref<'_, T>
返回对最近发送值的引用。
此方法不会将返回的值标记为已读,因此对 changed 的后续调用可能会立即返回,即使你已经通过调用 borrow 看到了该值。
未释放的借用会对内部值保持一个读锁。这意味着长时间存活的借用可能会导致生产者一半被阻塞。建议尽量缩短借用的生命周期。此外,如果运行在允许 !Send future 的环境中,必须确保返回的 Ref 类型不会跨越 .await 点存活,否则可能会导致死锁。
锁的优先级策略取决于底层锁实现,本类型不保证使用任何特定策略。特别是,正在等待通过 send 获取锁的生产者可能会阻塞对 borrow 的并发调用,也可能不会,例如:
Potential deadlock example
// Task 1 (on thread A) | // Task 2 (on thread B)
let _ref1 = rx.borrow(); |
| // will block
| let _ = tx.send(());
// may deadlock |
let _ref2 = rx.borrow(); |有关何时使用此方法与 borrow_and_update 的更多信息,请参阅此处。
§示例
use tokio::sync::watch;
let (_, rx) = watch::channel("hello");
assert_eq!(*rx.borrow(), "hello");源代码pub fn borrow_and_update(&mut self) -> Ref<'_, T>
pub fn borrow_and_update(&mut self) -> Ref<'_, T>
返回对最近发送值的引用,并将该值标记为已读。
此方法会将当前值标记为已读。在 Sender 再次修改共享值之前,对 changed 的后续调用不会立即返回。
未释放的借用会对内部值保持一个读锁。这意味着长时间存活的借用可能会导致生产者一半被阻塞。建议尽量缩短借用的生命周期。此外,如果运行在允许 !Send future 的环境中,必须确保返回的 Ref 类型不会跨越 .await 点存活,否则可能会导致死锁。
锁的优先级策略取决于底层锁实现,本类型不保证使用任何特定策略。特别是,正在等待通过 send 获取锁的生产者可能会阻塞对 borrow 的并发调用,也可能不会,例如:
Potential deadlock example
// Task 1 (on thread A) | // Task 2 (on thread B)
let _ref1 = rx1.borrow_and_update(); |
| // will block
| let _ = tx.send(());
// may deadlock |
let _ref2 = rx2.borrow_and_update(); |源代码pub fn has_changed(&self) -> Result<bool, RecvError>
pub fn has_changed(&self) -> Result<bool, RecvError>
检查此通道是否包含此接收者尚未看到的消息。当前值将不会被标记为已读。
虽然此方法称为 has_changed,但它不会检查消息的相等性,因此即使当前消息等于上一条消息,此调用也会返回 true。
§错误
当且仅当通道已关闭时返回 RecvError。
§示例
§Basic usage
use tokio::sync::watch;
let (tx, mut rx) = watch::channel("hello");
tx.send("goodbye").unwrap();
assert!(rx.has_changed().unwrap());
assert_eq!(*rx.borrow_and_update(), "goodbye");
// The value has been marked as seen
assert!(!rx.has_changed().unwrap());§Closed channel example
use tokio::sync::watch;
let (tx, rx) = watch::channel("hello");
tx.send("goodbye").unwrap();
drop(tx);
// The channel is closed
assert!(rx.has_changed().is_err());源代码pub fn mark_changed(&mut self)
pub fn mark_changed(&mut self)
源代码pub fn mark_unchanged(&mut self)
pub fn mark_unchanged(&mut self)
将状态标记为未更改。
当前值将被接收者视为已读。
如果你对接收者中可见的当前值不感兴趣,这将很有用。
源代码pub async fn changed(&mut self) -> Result<(), RecvError>
pub async fn changed(&mut self) -> Result<(), RecvError>
等待更改通知,然后将当前值标记为已读。
如果调用此方法时通道中的当前值尚未标记为已读,则该方法会将该值标记为已读并立即返回。如果最新值已标记为已读,则该方法将休眠,直到与此 Receiver 连接的 Sender 发送新消息,或者直到所有 Sender 都已 drop。
有关更多信息,请参阅模块级文档中的 更改通知。
§错误
如果通道已关闭并且当前值已读,则返回 RecvError。
§取消安全性
此方法是可安全取消的。如果在 tokio::select! 语句中将此方法用作事件,并且其他某个分支先完成,则可以保证对 changed 的此次调用未将任何值标记为已读。
§示例
use tokio::sync::watch;
let (tx, mut rx) = watch::channel("hello");
tokio::spawn(async move {
tx.send("goodbye").unwrap();
});
assert!(rx.changed().await.is_ok());
assert_eq!(*rx.borrow_and_update(), "goodbye");
// The `tx` handle has been dropped
assert!(rx.changed().await.is_err());源代码pub async fn wait_for(
&mut self,
f: impl FnMut(&T) -> bool,
) -> Result<Ref<'_, T>, RecvError>
pub async fn wait_for( &mut self, f: impl FnMut(&T) -> bool, ) -> Result<Ref<'_, T>, RecvError>
等待满足所提供条件的值。
每当通道上发送了某些内容时,此方法都会调用所提供的闭包。一旦闭包返回 true,此方法将返回传递给闭包的值的引用。
在 wait_for 开始等待更改之前,它将对当前值调用该闭包。如果对当前值调用时闭包返回 true,则 wait_for 将立即返回对当前值的引用。即使当前值已被视为已读,也是如此。
watch 通道仅跟踪最近的值,因此如果发送多条消息的速度快于 wait_for 调用闭包的速度,则它可能会跳过某些更新。每当调用闭包时,都会使用最近的值来调用它。
当此函数返回时,当闭包返回 true 时传递给闭包的值将被视为已读。
如果通道已关闭,则 wait_for 将返回 RecvError。一旦发生这种情况,通道上就不会再发送任何消息。当返回错误时,可以保证已对最后一个值调用了闭包,并且它对该值返回了 false。(如果闭包返回了 true,那么将返回最后一个值而不是错误。)
与 borrow 方法一样,返回的借用对内部值持有读锁。这意味着长时间的借用可能会导致生产者一半阻塞。建议将借用的生命周期保持尽可能短。有关这方面的更多信息,请参阅 borrow 的文档。
§取消安全性
此方法是可安全取消的。如果在 tokio::select! 语句中将此方法用作事件,并且其他某个分支先完成,则可以保证最后看到的值 val(如果有)满足 f(val) == false。
§恐慌
当且仅当闭包 f 发生 panic。在这种情况下,此 Receiver 拥有或共享的任何资源都不会被污染。
§示例
use tokio::sync::watch;
use tokio::time::{sleep, Duration};
#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
let (tx, mut rx) = watch::channel("hello");
tokio::spawn(async move {
sleep(Duration::from_secs(1)).await;
tx.send("goodbye").unwrap();
});
assert!(rx.wait_for(|val| *val == "goodbye").await.is_ok());
assert_eq!(*rx.borrow(), "goodbye");
}源代码pub fn same_channel(&self, other: &Self) -> bool
pub fn same_channel(&self, other: &Self) -> bool
如果这些 receiver 属于同一个通道,则返回 true。
§示例
let (tx, rx) = tokio::sync::watch::channel(true);
let rx2 = rx.clone();
assert!(rx.same_channel(&rx2));
let (tx3, rx3) = tokio::sync::watch::channel(true);
assert!(!rx3.same_channel(&rx2));