rust - 无限循环异步将客户端重新连接到服务器

喜欢ヅ旅行 2022-11-04 00:59 17阅读 0赞

我无法创建尝试连接到服务器的客户端，并且:

*  如果服务器关闭，则必须在无限循环中重试
 *  (如果服务器已启动且连接成功)，则在连接丢失(即服务器断开客户端连接)时，客户端必须重新启动无限循环才能尝试连接到服务器
 *  这是连接服务器的代码；当前，当连接断开时，程序退出。我不确定实现它的最佳方法是什么；也许我必须创建一个无限循环的`Future`？
    
        extern crate tokio_line; use tokio_line::LineCodec; fn get_connection(handle: &Handle) -> Box<Future<Item = (), Error = io::Error>> { let remote_addr = "127.0.0.1:9876".parse().unwrap(); let tcp = TcpStream::connect(&remote_addr, handle); let client = tcp.and_then(|stream| { let (sink, from_server) = stream.framed(LineCodec).split(); let reader = from_server.for_each(|message| { println!("{}", message); Ok(()) }); reader.map(|_| { println!("CLIENT DISCONNECTED"); () }).map_err(|err| err) }); let client = client.map_err(|_| { panic!()}); Box::new(client) } fn main() { let mut core = Core::new().unwrap(); let handle = core.handle(); let client = get_connection(&handle); let client = client.and_then(|c| { println!("Try to reconnect"); get_connection(&handle); Ok(()) }); core.run(client).unwrap(); }
    
      
    添加带有以下内容的tokio-line crate :  
     
    
        <span style="color:#212529"><code style="margin-left:2px">tokio-line = { git = "https://github.com/tokio-rs/tokio-line" }
        </code></span>
    
    **最佳答案**
    
    关键问题似乎是:如何使用Tokio实现无限循环？通过回答这个问题，我们可以解决在断开连接时无限重新连接的问题。根据我编写异步代码的经验，递归似乎是解决此问题的直接方法。  
      
    **更新**:正如Shepmaster(以及Tokio Gitter的人)所指出的那样，由于我们构建了随着每次迭代而增长的 future 链，所以我的原始答案会泄漏内存。下面是一个新的:  
      
    更新答案:使用`loop_fn``futures` crate 中有一个函数可以完全满足您的需求。它称为[ `loop_fn` ][loop_fn]。您可以通过将主要功能更改为以下内容来使用它:
    
        fn main() { let mut core = Core::new().unwrap(); let handle = core.handle(); let client = future::loop_fn((), |_| { // Run the get_connection function and loop again regardless of its result get_connection(&handle).map(|_| -> Loop<(), ()> { Loop::Continue(()) }) }); core.run(client).unwrap(); }
    
      
    该函数类似于for循环，可以根据`get_connection`的结果继续或中断(请参见[ `Loop` ][Loop]枚举的文档)。在这种情况下，我们选择始终继续，因此它将无限地保持重新连接。  
      
    请注意，如果出现错误(例如，如果客户端无法连接到服务器)，您的`get_connection`版本将出现恐慌。如果您还想在发生错误后重试，则应删除对`panic!`的调用。  
      
    旧答案:使用递归  
      
    如果有人发现它有趣，请遵循我的旧答案。  
      
    **警告**:使用下面的代码会导致无限的内存增长。  
      
    使`get_connection`无限循环  
      
    我们希望在每次客户端断开连接时都调用`get_connection`函数，这正是我们要做的(请查看`reader.and_then`之后的注释):
    
        fn get_connection(handle: &Handle) -> Box<Future<Item = (), Error = io::Error>> { let remote_addr = "127.0.0.1:9876".parse().unwrap(); let tcp = TcpStream::connect(&remote_addr, handle); let handle_clone = handle.clone(); let client = tcp.and_then(|stream| { let (sink, from_server) = stream.framed(LineCodec).split(); let reader = from_server.for_each(|message| { println!("{}", message); Ok(()) }); reader.and_then(move |_| { println!("CLIENT DISCONNECTED"); // Attempt to reconnect in the future get_connection(&handle_clone) }) }); let client = client.map_err(|_| { panic!()}); Box::new(client) }
    
      
    记住`get_connection`是非阻塞的。它只是构造一个`Box<Future>`。这意味着当递归调用它时，我们仍然不会阻塞。相反，我们得到了一个新的 future ，可以使用`and_then`将其链接到上一个。如您所见，这与普通递归不同，因为堆栈不会在每次迭代中增长。  
      
    请注意，我们需要克隆`handle`(请参阅`handle_clone`)，并将其移至传递给`reader.and_then`的闭包中。这是必需的，因为闭包的寿命比该函数的寿命长(它将在以后的返回中包含)。  
      
    处理错误  
      
    您提供的代码无法处理客户端无法连接到服务器的情况(也不存在任何其他错误)。按照上面显示的相同原理，我们可以通过将`get_connection`的结尾更改为以下内容来处理错误:
    
        let handle_clone = handle.clone(); let client = client.or_else(move |err| { // Note: this code will infinitely retry, but you could pattern match on the error // to retry only on certain kinds of error println!("Error connecting to server: {}", err); get_connection(&handle_clone) }); Box::new(client)
    
      
    请注意，`or_else`类似于`and_then`，但是它对将来产生的错误进行操作。  
      
    从`main`删除不必要的代码  
      
    最后，不需要在`and_then`函数中使用`main`。您可以将`main`替换为以下代码:
    
        fn main() { let mut core = Core::new().unwrap(); let handle = core.handle(); let client = get_connection(&handle); core.run(client).unwrap(); }

[loop_fn]: https://docs.rs/futures/0.1.10/futures/future/fn.loop_fn.html
[Loop]: https://docs.rs/futures/0.1.10/futures/future/enum.Loop.html