结构体 Receiver 

pub struct Receiver<T> { /* private fields */ }

从关联的 Sender 接收值。

实例 由 channel 函数创建。

此 receiver 可以使用 ReceiverStream 转换为 Stream。

实现

impl<T> Receiver<T>

pub async fn recv(&mut self) -> Option<T>

接收此接收者的下一个值。

如果通道已关闭且通道的缓冲区中没有剩余消息，此方法返回 None。这表示再也无法从此 Receiver 接收到任何值。当所有发送者都已被丢弃或调用了 close 时，通道被关闭。

如果通道的缓冲区中没有消息，但通道尚未关闭，此方法将休眠直到发送消息或通道被关闭。请注意，如果调用了 close，但关闭前还有未完成的 Permit，则在 Permit 被释放之前，recv 不会认为通道已关闭。

Cancel safety

此方法是取消安全的。如果 recv 在 tokio::select! 语句中作为事件使用，并且其他分支首先完成，则可以保证此通道上没有接收到消息。

示例

use tokio::sync::mpsc;

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

tokio::spawn(async move {
    tx.send("hello").await.unwrap();
});

assert_eq!(Some("hello"), rx.recv().await);
assert_eq!(None, rx.recv().await);

值已缓冲：

use tokio::sync::mpsc;

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

tx.send("hello").await.unwrap();
tx.send("world").await.unwrap();

assert_eq!(Some("hello"), rx.recv().await);
assert_eq!(Some("world"), rx.recv().await);

pub async fn recv_many(&mut self, buffer: &mut Vec<T>, limit: usize) -> usize

接收此接收者的下一些值并扩展缓冲区。

此方法最多将缓冲区扩展 limit 指定数量的值。如果 limit 为零，函数立即返回 0。返回值是已添加到缓冲区的值数量。

对于 limit > 0，如果通道的队列中没有消息，但通道尚未关闭，此方法将休眠直到发送消息或通道被关闭。请注意，如果调用了 close，但关闭前还有未完成的 Permit，则在 Permit 被释放之前，recv_many 不会认为通道已关闭。

对于非零的 limit 值，此方法永远不会返回 0，除非通道已关闭且通道的队列中没有剩余消息。这表示再也无法从此 Receiver 接收到任何值。当所有发送者都已被丢弃或调用了 close 时，通道被关闭。

buffer 的容量按需增加。

Cancel safety

此方法是取消安全的。如果 recv_many 在 tokio::select! 语句中作为事件使用，并且其他分支首先完成，则可以保证此通道上没有接收到消息。

示例

use tokio::sync::mpsc;

let mut buffer: Vec<&str> = Vec::with_capacity(2);
let limit = 2;
let (tx, mut rx) = mpsc::channel(100);
let tx2 = tx.clone();
tx2.send("first").await.unwrap();
tx2.send("second").await.unwrap();
tx2.send("third").await.unwrap();

// Call `recv_many` to receive up to `limit` (2) values.
assert_eq!(2, rx.recv_many(&mut buffer, limit).await);
assert_eq!(vec!["first", "second"], buffer);

// If the buffer is full, the next call to `recv_many`
// reserves additional capacity.
assert_eq!(1, rx.recv_many(&mut buffer, 1).await);

tokio::spawn(async move {
    tx.send("fourth").await.unwrap();
});

// 'tx' is dropped, but `recv_many`
// is guaranteed not to return 0 as the channel
// is not yet closed.
assert_eq!(1, rx.recv_many(&mut buffer, 1).await);
assert_eq!(vec!["first", "second", "third", "fourth"], buffer);

// Once the last sender is dropped, the channel is
// closed and `recv_many` returns 0, capacity unchanged.
drop(tx2);
assert_eq!(0, rx.recv_many(&mut buffer, limit).await);
assert_eq!(vec!["first", "second", "third", "fourth"], buffer);

pub fn try_recv(&mut self) -> Result<T, TryRecvError>

尝试接收此接收者的下一个值。

如果通道当前为空但仍有未完成的 sender 或 permit，此方法返回 Empty 错误。

如果通道当前为空且没有未完成的 sender 或 permit，此方法返回 Disconnected 错误。

与 poll_recv 方法不同，此方法绝不会虚假地返回 Empty 错误。

示例

use tokio::sync::mpsc;
use tokio::sync::mpsc::error::TryRecvError;

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

tx.send("hello").await.unwrap();

assert_eq!(Ok("hello"), rx.try_recv());
assert_eq!(Err(TryRecvError::Empty), rx.try_recv());

tx.send("hello").await.unwrap();
// Drop the last sender, closing the channel.
drop(tx);

assert_eq!(Ok("hello"), rx.try_recv());
assert_eq!(Err(TryRecvError::Disconnected), rx.try_recv());

pub fn blocking_recv(&mut self) -> Option<T>

在异步上下文之外调用的阻塞接收。

如果通道已关闭且通道的缓冲区中没有剩余消息，此方法返回 None。这表示再也无法从此 Receiver 接收到任何值。当所有发送者都已被丢弃或调用了 close 时，通道被关闭。

如果通道的缓冲区中没有消息，但通道尚未关闭，此方法将阻塞直到发送消息或通道被关闭。

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

请注意，如果调用了 close，但关闭前还有未完成的 Permit，则在 Permit 被释放之前，blocking_recv 不会认为通道已关闭。

Panics

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

示例

use std::thread;
use tokio::runtime::Runtime;
use tokio::sync::mpsc;

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

    let sync_code = thread::spawn(move || {
        assert_eq!(Some(10), rx.blocking_recv());
    });

    Runtime::new()
        .unwrap()
        .block_on(async move {
            let _ = tx.send(10).await;
        });
    sync_code.join().unwrap()
}

pub fn blocking_recv_many(&mut self, buffer: &mut Vec<T>, limit: usize) -> usize

用于阻塞上下文的 Self::recv_many 变体。

适用与 Self::blocking_recv 相同的条件。

pub fn close(&mut self)

关闭通道的接收半部而不丢弃它。

这会阻止通过此通道发送更多消息，同时仍允许接收者排空已缓冲的消息。任何未完成的 Permit 值仍能发送消息。

为了保证不丢失消息，调用 close() 后必须反复调用 recv() 直到返回 None。如果存在未完成的 Permit 或 OwnedPermit 值，则在它们被释放之前，recv 方法不会返回 None。

示例

use tokio::sync::mpsc;

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

tokio::spawn(async move {
    let mut i = 0;
    while let Ok(permit) = tx.reserve().await {
        permit.send(i);
        i += 1;
    }
});

rx.close();

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

// Channel closed and no messages are lost.

pub fn is_closed(&self) -> bool

检查通道是否已关闭。

如果通道已关闭，此方法返回 true。当所有 Sender 都已被丢弃或调用了 Receiver::close 时，通道被关闭。

示例

use tokio::sync::mpsc;

let (_tx, mut rx) = mpsc::channel::<()>(10);
assert!(!rx.is_closed());

rx.close();

assert!(rx.is_closed());

pub fn is_empty(&self) -> bool

检查通道是否为空。

如果通道没有消息，此方法返回 true。

示例

use tokio::sync::mpsc;

let (tx, rx) = mpsc::channel(10);
assert!(rx.is_empty());

tx.send(0).await.unwrap();
assert!(!rx.is_empty());

pub fn len(&self) -> usize

返回通道中的消息数量。

示例

use tokio::sync::mpsc;

let (tx, rx) = mpsc::channel(10);
assert_eq!(0, rx.len());

tx.send(0).await.unwrap();
assert_eq!(1, rx.len());

pub fn capacity(&self) -> usize

返回通道的当前容量。

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

示例

use tokio::sync::mpsc;

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

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

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

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

// Directly sending a message drops the capacity by one.
tx.send(()).await.unwrap();
assert_eq!(rx.capacity(), 4);
assert_eq!(rx.len(), 1);

// Receiving the message increases the capacity by one.
rx.recv().await.unwrap();
assert_eq!(rx.capacity(), 5);
assert_eq!(rx.len(), 0);

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!(rx.max_capacity(), 5);
assert_eq!(rx.capacity(), 5);

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

pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>>

poll 以接收此通道上的下一条消息。

此方法返回：

• Poll::Pending if no messages are available but the channel is not closed, or if a spurious failure happens.
• Poll::Ready(Some(message)) if a message is available.
• Poll::Ready(None) if the channel has been closed and all messages sent before it was closed have been received.

当方法返回 Poll::Pending 时，提供的 Context 中的 Waker 被调度为在任一接收者上发送消息时或通道关闭时接收唤醒。请注意，对 poll_recv 或 poll_recv_many 的多次调用，只有最近一次调用传递的 Context 中的 Waker 会被调度为接收唤醒。

如果此方法因虚假失败而返回 Poll::Pending，则当导致虚假失败的情况得到解决时，Waker 将被通知。请注意，收到这样的唤醒并不保证下一次调用会成功 —— 它可能会以另一个虚假失败而失败。

pub fn poll_recv_many( &mut self, cx: &mut Context<'_>, buffer: &mut Vec<T>, limit: usize, ) -> Poll<usize>

poll 以接收此通道上的多条消息，并扩展提供的缓冲区。

此方法返回：

• Poll::Pending if no messages are available but the channel is not closed, or if a spurious failure happens.
• Poll::Ready(count) where count is the number of messages successfully received and stored in buffer. This can be less than, or equal to, limit.
• Poll::Ready(0) if limit is set to zero or when the channel is closed.

当方法返回 Poll::Pending 时，提供的 Context 中的 Waker 被调度为在任一接收者上发送消息时或通道关闭时接收唤醒。请注意，对 poll_recv 或 poll_recv_many 的多次调用，只有最近一次调用传递的 Context 中的 Waker 会被调度为接收唤醒。

请注意，此方法不保证恰好接收 limit 条消息。而是如果至少有一条消息可用，它会尽可能返回多达 limit 条消息。仅当通道已关闭（或 limit 为零）时，此方法才返回零。

示例

use std::task::{Context, Poll};
use std::pin::Pin;
use tokio::sync::mpsc;
use futures::Future;

struct MyReceiverFuture<'a> {
    receiver: mpsc::Receiver<i32>,
    buffer: &'a mut Vec<i32>,
    limit: usize,
}

impl<'a> Future for MyReceiverFuture<'a> {
    type Output = usize; // Number of messages received

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let MyReceiverFuture { receiver, buffer, limit } = &mut *self;

        // Now `receiver` and `buffer` are mutable references, and `limit` is copied
        match receiver.poll_recv_many(cx, *buffer, *limit) {
            Poll::Pending => Poll::Pending,
            Poll::Ready(count) => Poll::Ready(count),
        }
    }
}

let (tx, rx) = mpsc::channel(32);
let mut buffer = Vec::new();

let my_receiver_future = MyReceiverFuture {
    receiver: rx,
    buffer: &mut buffer,
    limit: 3,
};

for i in 0..10 {
    tx.send(i).await.unwrap();
}

let count = my_receiver_future.await;
assert_eq!(count, 3);
assert_eq!(buffer, vec![0,1,2])

pub fn sender_strong_count(&self) -> usize

返回 Sender 句柄的数量。

pub fn sender_weak_count(&self) -> usize

返回 WeakSender 句柄的数量。

Trait 实现

impl<T> Debug for Receiver<T>

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

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

impl<T> Unpin for Receiver<T>