跳到主要内容

Sender

搜索
tokio::sync::mpsc

结构体 Sender 

Source 
pub struct Sender<T> { /* private fields */ }
展开描述

向关联的 Receiver 发送值。

实例 由 channel 函数创建。

要将 Sender 转换为 Sink 或在 poll 函数 中使用它， 你可以 使用 PollSender 工具。

实现§

Source§

impl<T> Sender<T>

Source

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

发送一个值，等待直到有可用容量。

当确定通道的另一端尚未挂起时，发送成功。发送失败的情况是对应的接收者已被关闭。请注意，返回 Err 意味着数据将永远不会被接收，但返回 Ok 并不意味着数据一定会被接收。对应的接收者可能在此函数返回 Ok 之后立即挂起。

§Errors

如果通道的接收半部已关闭（无论是因为调用了 close 还是 Receiver 句柄已被丢弃），此函数返回错误。错误包含传递给 send 的值。

§Cancel safety

如果 send 在 tokio::select! 语句中作为事件使用，并且其他分支首先完成，则可以保证该消息未被发送。但是，在这种情况下，消息会被丢弃并丢失。

为避免丢失消息，请使用 reserve 预留容量，然后使用返回的 Permit 发送消息。

此通道使用队列来确保 send 和 reserve 调用按请求顺序完成。取消对 send 的调用会丢失在队列中的位置。

§示例

在下面的示例中，每次调用 send 都会阻塞，直到先前发送的值被接收。

use tokio::sync::mpsc;

let (tx, mut rx) = mpsc::channel(1);

tokio::spawn(async move {
    for i in 0..10 {
        if let Err(_) = tx.send(i).await {
            println!("receiver dropped");
            return;
        }
    }
});

while let Some(i) = rx.recv().await {
    println!("got = {}", i);
}
Source

pub async fn closed(&self)

当接收者已丢弃时完成。

这允许生产者在对所产生值的兴趣被取消时收到通知，并立即停止工作。

§Cancel safety

此方法是取消安全的。一旦通道关闭，它将永远保持关闭状态，所有后续对 closed 的调用都将立即返回。

§示例
use tokio::sync::mpsc;

let (tx1, rx) = mpsc::channel::<()>(1);
let tx2 = tx1.clone();
let tx3 = tx1.clone();
let tx4 = tx1.clone();
let tx5 = tx1.clone();
tokio::spawn(async move {
    drop(rx);
});

futures::join!(
    tx1.closed(),
    tx2.closed(),
    tx3.closed(),
    tx4.closed(),
    tx5.closed()
);
println!("Receiver dropped");
Source

pub fn try_send(&self, message: T) -> Result<(), TrySendError<T>>

尝试立即通过此 Sender 发送一条消息。

此方法与 send 的不同之处在于，如果通道缓冲区已满或没有接收者等待获取数据，则会立即返回。与 send 相比，此函数有两种失败情况（一种表示断开连接，一种表示缓冲区已满）。

§Errors

如果已达到通道容量，即通道已缓冲 n 个值（其中 n 是传递给 channel 的参数），则返回错误。

如果通道的接收半部已关闭（无论是因为调用了 close 还是 Receiver 句柄已被丢弃），此函数返回错误。错误包含传递给 send 的值。

§示例
use tokio::sync::mpsc;

// Create a channel with buffer size 1
let (tx1, mut rx) = mpsc::channel(1);
let tx2 = tx1.clone();

tokio::spawn(async move {
    tx1.send(1).await.unwrap();
    tx1.send(2).await.unwrap();
    // task waits until the receiver receives a value.
});

tokio::spawn(async move {
    // This will return an error and send
    // no message if the buffer is full
    let _ = tx2.try_send(3);
});

let mut msg;
msg = rx.recv().await.unwrap();
println!("message {} received", msg);

msg = rx.recv().await.unwrap();
println!("message {} received", msg);

// Third message may have never been sent
match rx.recv().await {
    Some(msg) => println!("message {} received", msg),
    None => println!("the third message was never sent"),
}
Source

pub async fn send_timeout( &self, value: T, timeout: Duration, ) -> Result<(), SendTimeoutError<T>>

发送一个值，等待直到有可用容量，但仅限一段有限的时间。

与 send 共享相同的成功和错误条件，但增加了一个额外的失败条件：提供的超时已过且仍没有可用容量。

§Errors

如果通道的接收半部已关闭（无论是因为调用了 close 还是 Receiver 已被丢弃），此函数返回错误。错误包含传递给 send 的值。

§Panics

如果在未启用 time 特性的 Tokio 运行时上下文之外调用此函数，会触发 panic。

§示例

在下面的示例中，每次调用 send_timeout 都会阻塞，直到先前发送的值被接收（除非超时已过）。

use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

let (tx, mut rx) = mpsc::channel(1);

tokio::spawn(async move {
    for i in 0..10 {
        if let Err(e) = tx.send_timeout(i, Duration::from_millis(100)).await {
            println!("send error: #{:?}", e);
            return;
        }
    }
});

while let Some(i) = rx.recv().await {
    println!("got = {}", i);
    sleep(Duration::from_millis(200)).await;
}
Source

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

在异步上下文之外调用的阻塞发送。

此方法用于从同步代码向异步代码发送的场景，即使接收者未使用 blocking_recv 接收消息也能工作。

§Panics

如果在异步执行上下文中调用此函数会触发 panic。

§示例
use std::thread;
use tokio::runtime::Runtime;
use tokio::sync::mpsc;

fn main() {
    let (tx, mut rx) = mpsc::channel::<u8>(1);

    let sync_code = thread::spawn(move || {
        tx.blocking_send(10).unwrap();
    });

    Runtime::new().unwrap().block_on(async move {
        assert_eq!(Some(10), rx.recv().await);
    });
    sync_code.join().unwrap()
}
Source

pub fn is_closed(&self) -> bool

检查通道是否已关闭。当 Receiver 被丢弃时或调用 Receiver::close 方法时会发生这种情况。

let (tx, rx) = tokio::sync::mpsc::channel::<()>(42);
assert!(!tx.is_closed());

let tx2 = tx.clone();
assert!(!tx2.is_closed());

drop(rx);
assert!(tx.is_closed());
assert!(tx2.is_closed());
Source

pub async fn reserve(&self) -> Result<Permit<'_, T>, SendError<()>>

等待通道容量。一旦有可发送一条消息的容量，就为调用方预留。

如果通道已满，此函数将等待未接收消息的数量小于通道容量。为调用方预留一条消息的容量。返回 Permit 以跟踪预留的容量。Permit 上的 send 函数会消费预留的容量。

丢弃 Permit 而不发送消息会将容量释放回通道。

§Cancel safety

此通道使用队列来确保 send 和 reserve 调用按请求顺序完成。取消对 reserve 的调用会丢失在队列中的位置。

§示例
use tokio::sync::mpsc;

let (tx, mut rx) = mpsc::channel(1);

// Reserve capacity
let permit = tx.reserve().await.unwrap();

// Trying to send directly on the `tx` will fail due to no
// available capacity.
assert!(tx.try_send(123).is_err());

// Sending on the permit succeeds
permit.send(456);

// The value sent on the permit is received
assert_eq!(rx.recv().await.unwrap(), 456);
Source

pub async fn reserve_many( &self, n: usize, ) -> Result<PermitIterator<'_, T>, SendError<()>>

等待通道容量。一旦有可发送 n 条消息的容量，就为调用方预留。

如果通道已满或可用许可证少于 n 个，此函数将等待未接收消息的数量小于通道容量 n。然后为调用方预留 n 条消息的容量。

返回 PermitIterator 以跟踪预留的容量。可以调用此 Iterator 直到耗尽以获得 Permit，然后调用 Permit::send。此函数与 try_reserve_many 类似，但它会等待槽位变为可用。

如果通道已关闭，此函数返回 SendError。

丢弃 PermitIterator 而不将其完全消费会将剩余许可证释放回通道。

§Cancel safety

此通道使用队列来确保 send 和 reserve_many 调用按请求顺序完成。取消对 reserve_many 的调用会丢失在队列中的位置。

§示例
use tokio::sync::mpsc;

let (tx, mut rx) = mpsc::channel(2);

// Reserve capacity
let mut permit = tx.reserve_many(2).await.unwrap();

// Trying to send directly on the `tx` will fail due to no
// available capacity.
assert!(tx.try_send(123).is_err());

// Sending with the permit iterator succeeds
permit.next().unwrap().send(456);
permit.next().unwrap().send(457);

// The iterator should now be exhausted
assert!(permit.next().is_none());

// The value sent on the permit is received
assert_eq!(rx.recv().await.unwrap(), 456);
assert_eq!(rx.recv().await.unwrap(), 457);
Source

pub async fn reserve_owned(self) -> Result<OwnedPermit<T>, SendError<()>>

等待通道容量，移动 Sender 并返回拥有的许可证。一旦有可发送一条消息的容量，就为调用方预留。

此方法按值移动 sender，并返回可用于向通道发送消息的拥有的许可证。与 Sender::reserve 不同，此方法可用于许可证必须对 'static 生命周期有效的情况。Sender 可以廉价地克隆（Sender::clone 本质上是引用计数递增，类似于 Arc::clone），因此当需要多个 OwnedPermit 或无法移动 Sender 时，可以在调用 reserve_owned 之前对其进行克隆。

如果通道已满，此函数将等待未接收消息的数量小于通道容量。为调用方预留一条消息的容量。返回 OwnedPermit 以跟踪预留的容量。OwnedPermit 上的 send 函数会消费预留的容量。

丢弃 OwnedPermit 而不发送消息会将容量释放回通道。

§Cancel safety

此通道使用队列来确保 send 和 reserve 调用按请求顺序完成。取消对 reserve_owned 的调用会丢失在队列中的位置。

§示例

使用 OwnedPermit 发送消息：

use tokio::sync::mpsc;

let (tx, mut rx) = mpsc::channel(1);

// Reserve capacity, moving the sender.
let permit = tx.reserve_owned().await.unwrap();

// Send a message, consuming the permit and returning
// the moved sender.
let tx = permit.send(123);

// The value sent on the permit is received.
assert_eq!(rx.recv().await.unwrap(), 123);

// The sender can now be used again.
tx.send(456).await.unwrap();

当需要多个 OwnedPermit 或无法按值移动 sender 时，可以在调用 reserve_owned 之前廉价地克隆它：

use tokio::sync::mpsc;

let (tx, mut rx) = mpsc::channel(1);

// Clone the sender and reserve capacity.
let permit = tx.clone().reserve_owned().await.unwrap();

// Trying to send directly on the `tx` will fail due to no
// available capacity.
assert!(tx.try_send(123).is_err());

// Sending on the permit succeeds.
permit.send(456);

// The value sent on the permit is received
assert_eq!(rx.recv().await.unwrap(), 456);
Source

pub fn try_reserve(&self) -> Result<Permit<'_, T>, TrySendError<()>>

尝试在通道中获取一个槽位，而无需等待该槽位变为可用。

如果通道已满，此函数返回 TrySendError；否则如果有可用槽位，它将返回一个 Permit，使您能够在保证有槽位的情况下向通道发送消息。此函数与 reserve 类似，只是它不会等待槽位变为可用。

丢弃 Permit 而不发送消息会将容量释放回通道。

§示例
use tokio::sync::mpsc;

let (tx, mut rx) = mpsc::channel(1);

// Reserve capacity
let permit = tx.try_reserve().unwrap();

// Trying to send directly on the `tx` will fail due to no
// available capacity.
assert!(tx.try_send(123).is_err());

// Trying to reserve an additional slot on the `tx` will
// fail because there is no capacity.
assert!(tx.try_reserve().is_err());

// Sending on the permit succeeds
permit.send(456);

// The value sent on the permit is received
assert_eq!(rx.recv().await.unwrap(), 456);
Source

pub fn try_reserve_many( &self, n: usize, ) -> Result<PermitIterator<'_, T>, TrySendError<()>>

尝试在通道中获取 n 个槽位，而无需等待这些槽位变为可用。

返回 PermitIterator 以跟踪预留的容量。可以调用此 Iterator 直到耗尽以获得 Permit，然后调用 Permit::send。此函数与 reserve_many 类似，只是它不会等待槽位变为可用。

如果通道上可用许可证少于 n 个，则此函数返回 TrySendError::Full。如果通道已关闭，此函数返回 TrySendError::Closed。

丢弃 PermitIterator 而不将其完全消费会将剩余许可证释放回通道。

§示例
use tokio::sync::mpsc;

let (tx, mut rx) = mpsc::channel(2);

// Reserve capacity
let mut permit = tx.try_reserve_many(2).unwrap();

// Trying to send directly on the `tx` will fail due to no
// available capacity.
assert!(tx.try_send(123).is_err());

// Trying to reserve an additional slot on the `tx` will
// fail because there is no capacity.
assert!(tx.try_reserve().is_err());

// Sending with the permit iterator succeeds
permit.next().unwrap().send(456);
permit.next().unwrap().send(457);

// The iterator should now be exhausted
assert!(permit.next().is_none());

// The value sent on the permit is received
assert_eq!(rx.recv().await.unwrap(), 456);
assert_eq!(rx.recv().await.unwrap(), 457);

// Trying to call try_reserve_many with 0 will return an empty iterator
let mut permit = tx.try_reserve_many(0).unwrap();
assert!(permit.next().is_none());

// Trying to call try_reserve_many with a number greater than the channel
// capacity will return an error
let permit = tx.try_reserve_many(3);
assert!(permit.is_err());

// Trying to call try_reserve_many on a closed channel will return an error
drop(rx);
let permit = tx.try_reserve_many(1);
assert!(permit.is_err());

let permit = tx.try_reserve_many(0);
assert!(permit.is_err());
Source

pub fn try_reserve_owned(self) -> Result<OwnedPermit<T>, TrySendError<Self>>

尝试在通道中获取一个槽位，而无需等待该槽位变为可用，并返回拥有的许可证。

此方法按值移动 sender，并返回可用于向通道发送消息的拥有的许可证。与 Sender::try_reserve 不同，此方法可用于许可证必须对 'static 生命周期有效的情况。Sender 可以廉价地克隆（Sender::clone 本质上是引用计数递增，类似于 Arc::clone），因此当需要多个 OwnedPermit 或无法移动 Sender 时，可以在调用 try_reserve_owned 之前对其进行克隆。

如果通道已满，此函数返回 TrySendError。由于 sender 是按值获取的，此情况下返回的 TrySendError 包含 sender，以便可以再次使用。否则，如果有可用槽位，此方法将返回一个 OwnedPermit，可在保证有槽位的情况下向通道发送消息。此函数与 reserve_owned 类似，只是它不会等待槽位变为可用。

丢弃 OwnedPermit 而不发送消息会将容量释放回通道。

§示例
use tokio::sync::mpsc;

let (tx, mut rx) = mpsc::channel(1);

// Reserve capacity
let permit = tx.clone().try_reserve_owned().unwrap();

// Trying to send directly on the `tx` will fail due to no
// available capacity.
assert!(tx.try_send(123).is_err());

// Trying to reserve an additional slot on the `tx` will
// fail because there is no capacity.
assert!(tx.try_reserve().is_err());

// Sending on the permit succeeds
permit.send(456);

// The value sent on the permit is received
assert_eq!(rx.recv().await.unwrap(), 456);
Source

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

如果发送者属于同一通道则返回 true。

§示例
let (tx, rx) = tokio::sync::mpsc::channel::<()>(1);
let  tx2 = tx.clone();
assert!(tx.same_channel(&tx2));

let (tx3, rx3) = tokio::sync::mpsc::channel::<()>(1);
assert!(!tx3.same_channel(&tx2));
Source

pub fn capacity(&self) -> usize

返回通道的当前容量。

当通过调用 send 或通过 reserve 预留容量来发送值时，容量下降。当值被 Receiver 接收时，容量上升。这与 max_capacity 不同，后者总是返回最初调用 channel 时指定的缓冲区容量。

§示例
use tokio::sync::mpsc;

let (tx, mut rx) = mpsc::channel::<()>(5);

assert_eq!(tx.capacity(), 5);

// Making a reservation drops the capacity by one.
let permit = tx.reserve().await.unwrap();
assert_eq!(tx.capacity(), 4);

// Sending and receiving a value increases the capacity by one.
permit.send(());
rx.recv().await.unwrap();
assert_eq!(tx.capacity(), 5);
Source

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

将 Sender 转换为 WeakSender，它不计入 RAII 语义——也就是说，如果通道的所有 Sender 实例都已被丢弃且仅剩 WeakSender 实例，则通道被关闭。

Source

pub fn max_capacity(&self) -> usize

返回通道的最大缓冲区容量。

最大容量是最初调用 channel 时指定的缓冲区容量。这与 capacity 不同，后者返回当前可用的缓冲区容量：随着消息的发送和接收，capacity 返回的值会上升或下降，而 max_capacity 返回的值将保持不变。

§示例
use tokio::sync::mpsc;

let (tx, _rx) = mpsc::channel::<()>(5);

// both max capacity and capacity are the same at first
assert_eq!(tx.max_capacity(), 5);
assert_eq!(tx.capacity(), 5);

// Making a reservation doesn't change the max capacity.
let permit = tx.reserve().await.unwrap();
assert_eq!(tx.max_capacity(), 5);
// but drops the capacity by one
assert_eq!(tx.capacity(), 4);
Source

pub fn strong_count(&self) -> usize

返回 Sender 句柄的数量。

Source

pub fn weak_count(&self) -> usize

返回 WeakSender 句柄的数量。

Trait 实现§

Source§

impl<T> Clone for Sender<T>

Source§

fn clone(&self) -> Self

返回值的副本。 更多信息
1.0.0 · Source§

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source. 更多信息
Source§

impl<T> Debug for Sender<T>

Source§

fn fmt(&self, fmt: &mut Formatter<'_>) -> Result

使用给定的格式化器格式化此值。 更多信息

自动 Trait 实现§

§

impl<T> Freeze for Sender<T>

§

impl<T> RefUnwindSafe for Sender<T>

§

impl<T> Send for Sender<T>
where T: Send,

§

impl<T> Sync for Sender<T>
where T: Send,

§

impl<T> Unpin for Sender<T>

§

impl<T> UnsafeUnpin for Sender<T>

§

impl<T> UnwindSafe for Sender<T>

Blanket 实现§

Source§

impl<T> Any for T
where T: 'static + ?Sized,

Source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. 更多信息
Source§

impl<T> Borrow<T> for T
where T: ?Sized,

Source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. 更多信息
Source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

Source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. 更多信息
Source§

impl<T> CloneToUninit for T
where T: Clone,

Source§

unsafe fn clone_to_uninit(&self, dest: *mut u8)

🔬This is a nightly-only experimental API. (clone_to_uninit)
Performs copy-assignment from self to dest. 更多信息
Source§

impl<T> From<T> for T

Source§

fn from(t: T) -> T

原样返回传入的参数。

Source§

impl<T, U> Into<U> for T
where U: From<T>,

Source§

fn into(self) -> U

调用 U::from(self)。

也就是说，此转换的具体行为取决于 From<T> for U 的实现方式。

Source§

impl<T> ToOwned for T
where T: Clone,

Source§

type Owned = T

获得所有权后的类型。
Source§

fn to_owned(&self) -> T

Creates owned data from borrowed data, usually by cloning. 更多信息
Source§

fn clone_into(&self, target: &mut T)

Uses borrowed data to replace owned data, usually by cloning. 更多信息
Source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

Source§

type Error = Infallible

转换出错时返回的类型。
Source§

fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>

执行转换。
Source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

Source§

type Error = <U as TryFrom<T>>::Error

转换出错时返回的类型。
Source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

执行转换。