结构体 Sender

源代码

pub struct Sender<T> { /* 私有字段 */ }

展开描述

broadcast channel 的发送端。

可以从多个线程使用。可以使用 send 发送消息。

§示例

use tokio::sync::broadcast;

let (tx, mut rx1) = broadcast::channel(16);
let mut rx2 = tx.subscribe();

tokio::spawn(async move {
    assert_eq!(rx1.recv().await.unwrap(), 10);
    assert_eq!(rx1.recv().await.unwrap(), 20);
});

tokio::spawn(async move {
    assert_eq!(rx2.recv().await.unwrap(), 10);
    assert_eq!(rx2.recv().await.unwrap(), 20);
});

tx.send(10).unwrap();
tx.send(20).unwrap();

实现§

源代码 §

impl<T> Sender<T>

源代码

pub fn new(capacity: usize) -> Self

创建 broadcast channel 的发送端。

有关此方法的更多信息，请参阅 broadcast::channel 的文档。

源代码

pub fn send(&self, value: T) -> Result<usize, SendError<T>>

尝试向所有活动的 Receiver 句柄发送一个值，如果无法发送则将其返回。

当至少有一个活动的 Receiver 句柄时，发送成功。发送失败是指所有关联的 Receiver 句柄都已被 drop。

§返回值

成功时，返回已订阅的 Receiver 句柄数。这并不意味着这么多接收者将看到该消息，因为接收者可能在接收消息之前 drop 或滞后（参见滞后）。

§注意

Ok 的返回值并不意味着所发送的值将被所有或任何活动的 Receiver 句柄观察到。Receiver 句柄可能在接收到发送的消息之前被 drop。

Err 的返回值并不意味着未来对 send 的调用会失败。可以通过调用 subscribe 来创建新的 Receiver 句柄。

§示例

use tokio::sync::broadcast;

let (tx, mut rx1) = broadcast::channel(16);
let mut rx2 = tx.subscribe();

tokio::spawn(async move {
    assert_eq!(rx1.recv().await.unwrap(), 10);
    assert_eq!(rx1.recv().await.unwrap(), 20);
});

tokio::spawn(async move {
    assert_eq!(rx2.recv().await.unwrap(), 10);
    assert_eq!(rx2.recv().await.unwrap(), 20);
});

tx.send(10).unwrap();
tx.send(20).unwrap();

创建一个新的 Receiver 句柄，该句柄将接收在此次调用 subscribe 之后发送的值。

§示例

use tokio::sync::broadcast;

let (tx, _rx) = broadcast::channel(16);

// Will not be seen
tx.send(10).unwrap();

let mut rx = tx.subscribe();

tx.send(20).unwrap();

let value = rx.recv().await.unwrap();
assert_eq!(20, value);

源代码

pub fn downgrade(&self) -> WeakSender<T>

将 Sender 转换为 WeakSender，它不计入 RAII 语义，即如果 channel 的所有 Sender 实例都已被 drop，并且只剩下 WeakSender 实例，则 channel 已关闭。

源代码

pub fn len(&self) -> usize

返回已排队值的数量。

一个值在发送时被所有当时活动的接收者看到之前，或被超出队列容量的后续发送从队列中驱逐之前，一直处于排队状态。

§注意

与 Receiver::len 相比，此方法仅报告已排队的值，而不报告在被所有接收者看到之前已从队列中驱逐的值。

§示例

use tokio::sync::broadcast;

let (tx, mut rx1) = broadcast::channel(16);
let mut rx2 = tx.subscribe();

tx.send(10).unwrap();
tx.send(20).unwrap();
tx.send(30).unwrap();

assert_eq!(tx.len(), 3);

rx1.recv().await.unwrap();

// The len is still 3 since rx2 hasn't seen the first value yet.
assert_eq!(tx.len(), 3);

rx2.recv().await.unwrap();

assert_eq!(tx.len(), 2);

源代码

pub fn is_empty(&self) -> bool

如果没有排队的值，则返回 true。

§示例

use tokio::sync::broadcast;

let (tx, mut rx1) = broadcast::channel(16);
let mut rx2 = tx.subscribe();

assert!(tx.is_empty());

tx.send(10).unwrap();

assert!(!tx.is_empty());

rx1.recv().await.unwrap();

// The queue is still not empty since rx2 hasn't seen the value.
assert!(!tx.is_empty());

rx2.recv().await.unwrap();

assert!(tx.is_empty());

源代码

pub fn receiver_count(&self) -> usize

返回活动接收者的数量。

活动接收者是从 channel 或 subscribe 返回的 Receiver 句柄。这些句柄将接收在此 Sender 上发送的值。

§注意

不能保证发送的消息会到达此数量的接收者。活动接收者在 drop 之前可能再也不会调用 recv。

§示例

use tokio::sync::broadcast;

let (tx, _rx1) = broadcast::channel(16);

assert_eq!(1, tx.receiver_count());

let mut _rx2 = tx.subscribe();

assert_eq!(2, tx.receiver_count());

tx.send(10).unwrap();

源代码

pub fn same_channel(&self, other: &Self) -> bool

如果这些 sender 属于同一个通道，则返回 true。

§示例

use tokio::sync::broadcast;

let (tx, _rx) = broadcast::channel::<()>(16);
let tx2 = tx.clone();

assert!(tx.same_channel(&tx2));

let (tx3, _rx3) = broadcast::channel::<()>(16);

assert!(!tx3.same_channel(&tx2));

源代码

pub async fn closed(&self)

当订阅此 Sender 的 Receiver 数量达到零时完成的 future。

§示例

use futures::FutureExt;
use tokio::sync::broadcast;

let (tx, mut rx1) = broadcast::channel::<u32>(16);
let mut rx2 = tx.subscribe();

let _ = tx.send(10);

assert_eq!(rx1.recv().await.unwrap(), 10);
drop(rx1);
assert!(tx.closed().now_or_never().is_none());

assert_eq!(rx2.recv().await.unwrap(), 10);
drop(rx2);
assert!(tx.closed().now_or_never().is_some());