|
|
|
|
|
<!DOCTYPE HTML>
|
|
|
|
|
|
<html lang="zh-CN" class="light" dir="ltr">
|
|
|
|
|
|
<head>
|
|
|
|
|
|
<!-- Book generated using mdBook -->
|
|
|
|
|
|
<meta charset="UTF-8">
|
|
|
|
|
|
<title>将单线程 server 变为多线程 server - Rust 程序设计语言 简体中文版</title>
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
<!-- Custom HTML head -->
|
|
|
|
|
|
|
|
|
|
|
|
<meta name="description" content="Rust 程序设计语言 简体中文版">
|
|
|
|
|
|
<meta name="viewport" content="width=device-width, initial-scale=1">
|
|
|
|
|
|
<meta name="theme-color" content="#ffffff">
|
|
|
|
|
|
|
|
|
|
|
|
<link rel="icon" href="favicon.svg">
|
|
|
|
|
|
<link rel="shortcut icon" href="favicon.png">
|
|
|
|
|
|
<link rel="stylesheet" href="css/variables.css">
|
|
|
|
|
|
<link rel="stylesheet" href="css/general.css">
|
|
|
|
|
|
<link rel="stylesheet" href="css/chrome.css">
|
|
|
|
|
|
<link rel="stylesheet" href="css/print.css" media="print">
|
|
|
|
|
|
|
|
|
|
|
|
<!-- Fonts -->
|
|
|
|
|
|
<link rel="stylesheet" href="FontAwesome/css/font-awesome.css">
|
|
|
|
|
|
<link rel="stylesheet" href="fonts/fonts.css">
|
|
|
|
|
|
|
|
|
|
|
|
<!-- Highlight.js Stylesheets -->
|
|
|
|
|
|
<link rel="stylesheet" href="highlight.css">
|
|
|
|
|
|
<link rel="stylesheet" href="tomorrow-night.css">
|
|
|
|
|
|
<link rel="stylesheet" href="ayu-highlight.css">
|
|
|
|
|
|
|
|
|
|
|
|
<!-- Custom theme stylesheets -->
|
|
|
|
|
|
<link rel="stylesheet" href="ferris.css">
|
|
|
|
|
|
<link rel="stylesheet" href="theme/2018-edition.css">
|
|
|
|
|
|
|
|
|
|
|
|
</head>
|
|
|
|
|
|
<body class="sidebar-visible no-js">
|
|
|
|
|
|
<div id="body-container">
|
|
|
|
|
|
<!-- Provide site root to javascript -->
|
|
|
|
|
|
<script>
|
|
|
|
|
|
var path_to_root = "";
|
|
|
|
|
|
var default_theme = window.matchMedia("(prefers-color-scheme: dark)").matches ? "navy" : "light";
|
|
|
|
|
|
</script>
|
|
|
|
|
|
|
|
|
|
|
|
<!-- Work around some values being stored in localStorage wrapped in quotes -->
|
|
|
|
|
|
<script>
|
|
|
|
|
|
try {
|
|
|
|
|
|
var theme = localStorage.getItem('mdbook-theme');
|
|
|
|
|
|
var sidebar = localStorage.getItem('mdbook-sidebar');
|
|
|
|
|
|
|
|
|
|
|
|
if (theme.startsWith('"') && theme.endsWith('"')) {
|
|
|
|
|
|
localStorage.setItem('mdbook-theme', theme.slice(1, theme.length - 1));
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
if (sidebar.startsWith('"') && sidebar.endsWith('"')) {
|
|
|
|
|
|
localStorage.setItem('mdbook-sidebar', sidebar.slice(1, sidebar.length - 1));
|
|
|
|
|
|
}
|
|
|
|
|
|
} catch (e) { }
|
|
|
|
|
|
</script>
|
|
|
|
|
|
|
|
|
|
|
|
<!-- Set the theme before any content is loaded, prevents flash -->
|
|
|
|
|
|
<script>
|
|
|
|
|
|
var theme;
|
|
|
|
|
|
try { theme = localStorage.getItem('mdbook-theme'); } catch(e) { }
|
|
|
|
|
|
if (theme === null || theme === undefined) { theme = default_theme; }
|
|
|
|
|
|
var html = document.querySelector('html');
|
|
|
|
|
|
html.classList.remove('light')
|
|
|
|
|
|
html.classList.add(theme);
|
|
|
|
|
|
var body = document.querySelector('body');
|
|
|
|
|
|
body.classList.remove('no-js')
|
|
|
|
|
|
body.classList.add('js');
|
|
|
|
|
|
</script>
|
|
|
|
|
|
|
|
|
|
|
|
<input type="checkbox" id="sidebar-toggle-anchor" class="hidden">
|
|
|
|
|
|
|
|
|
|
|
|
<!-- Hide / unhide sidebar before it is displayed -->
|
|
|
|
|
|
<script>
|
|
|
|
|
|
var body = document.querySelector('body');
|
|
|
|
|
|
var sidebar = null;
|
|
|
|
|
|
var sidebar_toggle = document.getElementById("sidebar-toggle-anchor");
|
|
|
|
|
|
if (document.body.clientWidth >= 1080) {
|
|
|
|
|
|
try { sidebar = localStorage.getItem('mdbook-sidebar'); } catch(e) { }
|
|
|
|
|
|
sidebar = sidebar || 'visible';
|
|
|
|
|
|
} else {
|
|
|
|
|
|
sidebar = 'hidden';
|
|
|
|
|
|
}
|
|
|
|
|
|
sidebar_toggle.checked = sidebar === 'visible';
|
|
|
|
|
|
body.classList.remove('sidebar-visible');
|
|
|
|
|
|
body.classList.add("sidebar-" + sidebar);
|
|
|
|
|
|
</script>
|
|
|
|
|
|
|
|
|
|
|
|
<nav id="sidebar" class="sidebar" aria-label="Table of contents">
|
|
|
|
|
|
<div class="sidebar-scrollbox">
|
|
|
|
|
|
<ol class="chapter"><li class="chapter-item expanded affix "><a href="title-page.html">Rust 程序设计语言</a></li><li class="chapter-item expanded affix "><a href="foreword.html">前言</a></li><li class="chapter-item expanded affix "><a href="ch00-00-introduction.html">简介</a></li><li class="chapter-item expanded "><a href="ch01-00-getting-started.html"><strong aria-hidden="true">1.</strong> 入门指南</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="ch01-01-installation.html"><strong aria-hidden="true">1.1.</strong> 安装</a></li><li class="chapter-item expanded "><a href="ch01-02-hello-world.html"><strong aria-hidden="true">1.2.</strong> Hello, World!</a></li><li class="chapter-item expanded "><a href="ch01-03-hello-cargo.html"><strong aria-hidden="true">1.3.</strong> Hello, Cargo!</a></li></ol></li><li class="chapter-item expanded "><a href="ch02-00-guessing-game-tutorial.html"><strong aria-hidden="true">2.</strong> 写个猜数字游戏</a></li><li class="chapter-item expanded "><a href="ch03-00-common-programming-concepts.html"><strong aria-hidden="true">3.</strong> 常见编程概念</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="ch03-01-variables-and-mutability.html"><strong aria-hidden="true">3.1.</strong> 变量与可变性</a></li><li class="chapter-item expanded "><a href="ch03-02-data-types.html"><strong aria-hidden="true">3.2.</strong> 数据类型</a></li><li class="chapter-item expanded "><a href="ch03-03-how-functions-work.html"><strong aria-hidden="true">3.3.</strong> 函数</a></li><li class="chapter-item expanded "><a href="ch03-04-comments.html"><strong aria-hidden="true">3.4.</strong> 注释</a></li><li class="chapter-item expanded "><a href="ch03-05-control-flow.html"><strong aria-hidden="true">3.5.</strong> 控制流</a></li></ol></li><li class="chapter-item expanded "><a href="ch04-00-understanding-ownership.html"><strong aria-hidden="true">4.</strong> 认识所有权</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="ch04-01-what-is-ownership.html"><strong aria-hidden="true">4.1.</strong> 什么是所有权？</a></li><li class="chapter-item expanded "><a href="ch04-02-references-and-borrowing.html"><strong aria-hidden="true">4.2.</strong> 引用与借用</a></li><li class="chapter-item expanded "><a href="ch04-03-slices.html"><strong aria-hidden="true">4.3.</strong> Slice 类型</a></li></ol></li><li class="chapter-item expanded "><a href="ch05-00-structs.html"><strong aria-hidden="true">5.</strong> 使用结构体组织相关联的数据</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="ch05-01-defining-structs.html"><strong aria-hidden="true">5.1.</strong> 结构体的定义和实例化</a></li><li class="chapter-item expanded "><a href="ch05-02-example-structs.html"><strong aria-hidden="true">5.2.</strong> 结构体示例程序</a></li><li class="chapter-item expanded "><a href="ch05-03-method-syntax.html"><strong aria-hidden="true">5.3.</strong> 方法语法</a></li></ol></li><li class="chapter-item expanded "><a href="ch06-00-enums.html"><strong aria-hidden="true">6.</strong> 枚举和模式匹配</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="ch06-01-defining-an-enum.html"><strong aria-hidden="true">6.1.</strong> 枚举的定义</a></li><li class="chapter-item expanded "><a href="ch06-02-match.html"><strong aria-hidden="true">6.2.</strong> match 控制流结构</a></li><li class="chapter-item expanded "><a href="ch06-03-if-let.html"><strong aria-hidden="true">6.3.</strong> if let 简洁控制流</a></li></ol></li><li class="chapter-item expanded "><a href="ch07-00-managing-growing-projects-with-packages-crates-and-modules.html"><strong aria-hidden="true">7.</strong> 使用包、Crate 和模块管理不断增长的项目</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="ch07-01-packages-and-crates.html"><strong aria-hidden="true">7.1.</strong> 包和 Crate</a></li><li class="chapter-item expanded "><a h
|
|
|
|
|
|
</div>
|
|
|
|
|
|
<div id="sidebar-resize-handle" class="sidebar-resize-handle">
|
|
|
|
|
|
<div class="sidebar-resize-indicator"></div>
|
|
|
|
|
|
</div>
|
|
|
|
|
|
</nav>
|
|
|
|
|
|
|
|
|
|
|
|
<!-- Track and set sidebar scroll position -->
|
|
|
|
|
|
<script>
|
|
|
|
|
|
var sidebarScrollbox = document.querySelector('#sidebar .sidebar-scrollbox');
|
|
|
|
|
|
sidebarScrollbox.addEventListener('click', function(e) {
|
|
|
|
|
|
if (e.target.tagName === 'A') {
|
|
|
|
|
|
sessionStorage.setItem('sidebar-scroll', sidebarScrollbox.scrollTop);
|
|
|
|
|
|
}
|
|
|
|
|
|
}, { passive: true });
|
|
|
|
|
|
var sidebarScrollTop = sessionStorage.getItem('sidebar-scroll');
|
|
|
|
|
|
sessionStorage.removeItem('sidebar-scroll');
|
|
|
|
|
|
if (sidebarScrollTop) {
|
|
|
|
|
|
// preserve sidebar scroll position when navigating via links within sidebar
|
|
|
|
|
|
sidebarScrollbox.scrollTop = sidebarScrollTop;
|
|
|
|
|
|
} else {
|
|
|
|
|
|
// scroll sidebar to current active section when navigating via "next/previous chapter" buttons
|
|
|
|
|
|
var activeSection = document.querySelector('#sidebar .active');
|
|
|
|
|
|
if (activeSection) {
|
|
|
|
|
|
activeSection.scrollIntoView({ block: 'center' });
|
|
|
|
|
|
}
|
|
|
|
|
|
}
|
|
|
|
|
|
</script>
|
|
|
|
|
|
|
|
|
|
|
|
<div id="page-wrapper" class="page-wrapper">
|
|
|
|
|
|
|
|
|
|
|
|
<div class="page">
|
|
|
|
|
|
<div id="menu-bar-hover-placeholder"></div>
|
|
|
|
|
|
<div id="menu-bar" class="menu-bar sticky">
|
|
|
|
|
|
<div class="left-buttons">
|
|
|
|
|
|
<label id="sidebar-toggle" class="icon-button" for="sidebar-toggle-anchor" title="Toggle Table of Contents" aria-label="Toggle Table of Contents" aria-controls="sidebar">
|
|
|
|
|
|
<i class="fa fa-bars"></i>
|
|
|
|
|
|
</label>
|
|
|
|
|
|
<button id="theme-toggle" class="icon-button" type="button" title="Change theme" aria-label="Change theme" aria-haspopup="true" aria-expanded="false" aria-controls="theme-list">
|
|
|
|
|
|
<i class="fa fa-paint-brush"></i>
|
|
|
|
|
|
</button>
|
|
|
|
|
|
<ul id="theme-list" class="theme-popup" aria-label="Themes" role="menu">
|
|
|
|
|
|
<li role="none"><button role="menuitem" class="theme" id="light">Light</button></li>
|
|
|
|
|
|
<li role="none"><button role="menuitem" class="theme" id="rust">Rust</button></li>
|
|
|
|
|
|
<li role="none"><button role="menuitem" class="theme" id="coal">Coal</button></li>
|
|
|
|
|
|
<li role="none"><button role="menuitem" class="theme" id="navy">Navy</button></li>
|
|
|
|
|
|
<li role="none"><button role="menuitem" class="theme" id="ayu">Ayu</button></li>
|
|
|
|
|
|
</ul>
|
|
|
|
|
|
<button id="search-toggle" class="icon-button" type="button" title="Search. (Shortkey: s)" aria-label="Toggle Searchbar" aria-expanded="false" aria-keyshortcuts="S" aria-controls="searchbar">
|
|
|
|
|
|
<i class="fa fa-search"></i>
|
|
|
|
|
|
</button>
|
|
|
|
|
|
</div>
|
|
|
|
|
|
|
|
|
|
|
|
<h1 class="menu-title">Rust 程序设计语言 简体中文版</h1>
|
|
|
|
|
|
|
|
|
|
|
|
<div class="right-buttons">
|
|
|
|
|
|
<a href="print.html" title="Print this book" aria-label="Print this book">
|
|
|
|
|
|
<i id="print-button" class="fa fa-print"></i>
|
|
|
|
|
|
</a>
|
|
|
|
|
|
<a href="https://github.com/KaiserY/trpl-zh-cn/tree/main" title="Git repository" aria-label="Git repository">
|
|
|
|
|
|
<i id="git-repository-button" class="fa fa-github"></i>
|
|
|
|
|
|
</a>
|
|
|
|
|
|
<a href="https://github.com/KaiserY/trpl-zh-cn/edit/main/src/ch20-02-multithreaded.md" title="Suggest an edit" aria-label="Suggest an edit">
|
|
|
|
|
|
<i id="git-edit-button" class="fa fa-edit"></i>
|
|
|
|
|
|
</a>
|
|
|
|
|
|
|
|
|
|
|
|
</div>
|
|
|
|
|
|
</div>
|
|
|
|
|
|
|
|
|
|
|
|
<div id="search-wrapper" class="hidden">
|
|
|
|
|
|
<form id="searchbar-outer" class="searchbar-outer">
|
|
|
|
|
|
<input type="search" id="searchbar" name="searchbar" placeholder="Search this book ..." aria-controls="searchresults-outer" aria-describedby="searchresults-header">
|
|
|
|
|
|
</form>
|
|
|
|
|
|
<div id="searchresults-outer" class="searchresults-outer hidden">
|
|
|
|
|
|
<div id="searchresults-header" class="searchresults-header"></div>
|
|
|
|
|
|
<ul id="searchresults">
|
|
|
|
|
|
</ul>
|
|
|
|
|
|
</div>
|
|
|
|
|
|
</div>
|
|
|
|
|
|
|
|
|
|
|
|
<!-- Apply ARIA attributes after the sidebar and the sidebar toggle button are added to the DOM -->
|
|
|
|
|
|
<script>
|
|
|
|
|
|
document.getElementById('sidebar-toggle').setAttribute('aria-expanded', sidebar === 'visible');
|
|
|
|
|
|
document.getElementById('sidebar').setAttribute('aria-hidden', sidebar !== 'visible');
|
|
|
|
|
|
Array.from(document.querySelectorAll('#sidebar a')).forEach(function(link) {
|
|
|
|
|
|
link.setAttribute('tabIndex', sidebar === 'visible' ? 0 : -1);
|
|
|
|
|
|
});
|
|
|
|
|
|
</script>
|
|
|
|
|
|
|
|
|
|
|
|
<div id="content" class="content">
|
|
|
|
|
|
<main>
|
|
|
|
|
|
<h2 id="将单线程-server-变为多线程-server"><a class="header" href="#将单线程-server-变为多线程-server">将单线程 server 变为多线程 server</a></h2>
|
|
|
|
|
|
<blockquote>
|
|
|
|
|
|
<p><a href="https://github.com/rust-lang/book/blob/main/src/ch20-02-multithreaded.md">ch20-02-multithreaded.md</a>
|
|
|
|
|
|
<br>
|
|
|
|
|
|
commit 98c6225e5fb8255349ec0dc235433530ed3fb534</p>
|
|
|
|
|
|
</blockquote>
|
|
|
|
|
|
<p>目前 server 会依次处理每一个请求，意味着它在完成第一个连接的处理之前不会处理第二个连接。如果 server 正接收越来越多的请求，这类串行操作会使性能越来越差。如果一个请求花费很长时间来处理，随后而来的请求则不得不等待这个长请求结束，即便这些新请求可以很快就处理完。我们需要修复这种情况，不过首先让我们实际尝试一下这个问题。</p>
|
|
|
|
|
|
<h3 id="在当前-server-实现中模拟慢请求"><a class="header" href="#在当前-server-实现中模拟慢请求">在当前 server 实现中模拟慢请求</a></h3>
|
|
|
|
|
|
<p>让我们看看一个慢请求如何影响当前 server 实现中的其他请求。示例 20-10 通过模拟慢响应实现了 <em>/sleep</em> 请求处理，它会使 server 在响应之前休眠五秒。</p>
|
|
|
|
|
|
<p><span class="filename">文件名：src/main.rs</span></p>
|
|
|
|
|
|
<pre><pre class="playground"><code class="language-rust no_run">use std::{
|
|
|
|
|
|
fs,
|
|
|
|
|
|
io::{prelude::*, BufReader},
|
|
|
|
|
|
net::{TcpListener, TcpStream},
|
|
|
|
|
|
thread,
|
|
|
|
|
|
time::Duration,
|
|
|
|
|
|
};
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
<span class="boring">
|
|
|
|
|
|
</span><span class="boring">fn main() {
|
|
|
|
|
|
</span><span class="boring"> let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> for stream in listener.incoming() {
|
|
|
|
|
|
</span><span class="boring"> let stream = stream.unwrap();
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> handle_connection(stream);
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span><span class="boring">}
|
|
|
|
|
|
</span>
|
|
|
|
|
|
fn handle_connection(mut stream: TcpStream) {
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
|
|
|
|
|
|
<span class="boring"> let buf_reader = BufReader::new(&mut stream);
|
|
|
|
|
|
</span><span class="boring"> let request_line = buf_reader.lines().next().unwrap().unwrap();
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span> let (status_line, filename) = match &request_line[..] {
|
|
|
|
|
|
"GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
|
|
|
|
|
|
"GET /sleep HTTP/1.1" => {
|
|
|
|
|
|
thread::sleep(Duration::from_secs(5));
|
|
|
|
|
|
("HTTP/1.1 200 OK", "hello.html")
|
|
|
|
|
|
}
|
|
|
|
|
|
_ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
|
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
<span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let contents = fs::read_to_string(filename).unwrap();
|
|
|
|
|
|
</span><span class="boring"> let length = contents.len();
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let response =
|
|
|
|
|
|
</span><span class="boring"> format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> stream.write_all(response.as_bytes()).unwrap();
|
|
|
|
|
|
</span>}</code></pre></pre>
|
|
|
|
|
|
<p><span class="caption">示例 20-10: 通过休眠五秒来模拟慢请求</span></p>
|
|
|
|
|
|
<p>从 <code>if</code> 切换到 <code>match</code> 后现在有三个分支了。我们需要显式匹配一个 slice 的 <code>request_line</code> 以匹配字符串字面值的模式。<code>match</code> 不会像相等方法那样自动引用和解引用。</p>
|
|
|
|
|
|
<p>第一个分支与示例 20-9 中的 <code>if</code> 代码块相同。第二个分支匹配一个 <em>/sleep</em> 请求。当接收到这个请求时，server 在渲染成功 HTML 页面之前会先休眠五秒。第三个分支与示例 20-9 中的 <code>else</code> 代码块相同。</p>
|
|
|
|
|
|
<p>现在就可以真切的看出我们的 server 有多么的原始：真实的库将会以更简洁的方式处理多请求识别问题！</p>
|
|
|
|
|
|
<p>使用 <code>cargo run</code> 启动 server，并接着打开两个浏览器窗口：一个请求 <em>http://127.0.0.1:7878/</em> 而另一个请求 <em>http://127.0.0.1:7878/sleep</em> 。如果像之前一样多次请求 <em>/</em>，会发现响应的比较快速。不过如果请求 <em>/sleep</em> 之后再请求 <em>/</em>，就会看到 <em>/</em> 会等待直到 <code>sleep</code> 休眠完五秒之后才出现。</p>
|
|
|
|
|
|
<p>有多种技术可以用来避免所有请求都排在慢请求之后；我们将要实现的一个便是线程池。</p>
|
|
|
|
|
|
<h3 id="使用线程池改善吞吐量"><a class="header" href="#使用线程池改善吞吐量">使用线程池改善吞吐量</a></h3>
|
|
|
|
|
|
<p><strong>线程池</strong>（<em>thread pool</em>）是一组预先分配的等待或准备处理任务的线程。当程序收到一个新任务，线程池中的一个线程会被分配任务，这个线程会离开并处理任务。其余的线程则可用于处理在第一个线程处理任务的同时处理其他接收到的任务。当第一个线程处理完任务时，它会返回空闲线程池中等待处理新任务。线程池允许我们并发处理连接，增加 server 的吞吐量。</p>
|
|
|
|
|
|
<p>我们会将池中线程限制为较少的数量，以防拒绝服务（Denial of Service，DoS）攻击；如果程序为每一个接收的请求都新建一个线程，某人向 server 发起千万级的请求时会耗尽服务器的资源并导致所有请求的处理都被终止。</p>
|
|
|
|
|
|
<p>不同于分配无限的线程，线程池中将有固定数量的等待线程。当新进请求时，将请求发送到线程池中做处理。线程池会维护一个接收请求的队列。每一个线程会从队列中取出一个请求，处理请求，接着向队列索取另一个请求。通过这种设计，则可以并发处理 <code>N</code> 个请求，其中 <code>N</code> 为线程数。如果每一个线程都在响应慢请求，之后的请求仍然会阻塞队列，不过相比之前增加了能处理的慢请求的数量。</p>
|
|
|
|
|
|
<p>这个设计仅仅是多种改善 web server 吞吐量的方法之一。其他可供探索的方法有 <strong>fork/join 模型</strong>（<em>fork/join model</em>）、<strong>单线程异步 I/O 模型</strong>（<em>single-threaded async I/O model</em>）或者<strong>多线程异步 I/O 模型</strong>（<em>multi-threaded async I/O model</em>）。如果你对这个主题感兴趣，则可以阅读更多关于其他解决方案的内容并尝试实现它们；对于一个像 Rust 这样的底层语言，所有这些方法都是可能的。</p>
|
|
|
|
|
|
<p>在开始之前，让我们讨论一下线程池应用看起来怎样。当尝试设计代码时，首先编写客户端接口确实有助于指导代码设计。以期望的调用方式来构建 API 代码的结构，接着在这个结构之内实现功能，而不是先实现功能再设计公有 API。</p>
|
|
|
|
|
|
<p>类似于第十二章项目中使用的测试驱动开发。这里将要使用编译器驱动开发（compiler-driven development）。我们将编写调用所期望的函数的代码，接着观察编译器错误告诉我们接下来需要修改什么使得代码可以工作。不过在开始之前，我们将探索不会作为起点的技术。</p>
|
|
|
|
|
|
<h4 id="为每一个请求分配线程"><a class="header" href="#为每一个请求分配线程">为每一个请求分配线程</a></h4>
|
|
|
|
|
|
<p>首先，让我们探索一下为每一个连接都创建一个线程的代码看起来如何。这并不是最终方案，因为正如之前讲到的它会潜在的分配无限的线程，不过这是一个可用的多线程 server 的起点。接着我们会增加线程池作为改进，这样比较两个方案将会更容易。示例 20-11 展示了 <code>main</code> 的改变，它在 <code>for</code> 循环中为每一个流分配了一个新线程进行处理：</p>
|
|
|
|
|
|
<p><span class="filename">文件名：src/main.rs</span></p>
|
|
|
|
|
|
<pre><pre class="playground"><code class="language-rust no_run"><span class="boring">use std::{
|
|
|
|
|
|
</span><span class="boring"> fs,
|
|
|
|
|
|
</span><span class="boring"> io::{prelude::*, BufReader},
|
|
|
|
|
|
</span><span class="boring"> net::{TcpListener, TcpStream},
|
|
|
|
|
|
</span><span class="boring"> thread,
|
|
|
|
|
|
</span><span class="boring"> time::Duration,
|
|
|
|
|
|
</span><span class="boring">};
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span>fn main() {
|
|
|
|
|
|
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
|
|
|
|
|
|
|
|
|
|
|
|
for stream in listener.incoming() {
|
|
|
|
|
|
let stream = stream.unwrap();
|
|
|
|
|
|
|
|
|
|
|
|
thread::spawn(|| {
|
|
|
|
|
|
handle_connection(stream);
|
|
|
|
|
|
});
|
|
|
|
|
|
}
|
|
|
|
|
|
}
|
|
|
|
|
|
<span class="boring">
|
|
|
|
|
|
</span><span class="boring">fn handle_connection(mut stream: TcpStream) {
|
|
|
|
|
|
</span><span class="boring"> let buf_reader = BufReader::new(&mut stream);
|
|
|
|
|
|
</span><span class="boring"> let request_line = buf_reader.lines().next().unwrap().unwrap();
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let (status_line, filename) = match &request_line[..] {
|
|
|
|
|
|
</span><span class="boring"> "GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
|
|
|
|
|
|
</span><span class="boring"> "GET /sleep HTTP/1.1" => {
|
|
|
|
|
|
</span><span class="boring"> thread::sleep(Duration::from_secs(5));
|
|
|
|
|
|
</span><span class="boring"> ("HTTP/1.1 200 OK", "hello.html")
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span><span class="boring"> _ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
|
|
|
|
|
|
</span><span class="boring"> };
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let contents = fs::read_to_string(filename).unwrap();
|
|
|
|
|
|
</span><span class="boring"> let length = contents.len();
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let response =
|
|
|
|
|
|
</span><span class="boring"> format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> stream.write_all(response.as_bytes()).unwrap();
|
|
|
|
|
|
</span><span class="boring">}</span></code></pre></pre>
|
|
|
|
|
|
<p><span class="caption">示例 20-11: 为每一个流新建一个线程</span></p>
|
|
|
|
|
|
<p>正如第十六章讲到的，<code>thread::spawn</code> 会创建一个新线程并在其中运行闭包中的代码。如果运行这段代码并在在浏览器中加载 <em>/sleep</em>，接着在另两个浏览器标签页中加载 <em>/</em>，确实会发现 <em>/</em> 请求不必等待 <em>/sleep</em> 结束。不过正如之前提到的，这最终会使系统崩溃因为我们无限制的创建新线程。</p>
|
|
|
|
|
|
<h4 id="创建有限数量的线程"><a class="header" href="#创建有限数量的线程">创建有限数量的线程</a></h4>
|
|
|
|
|
|
<p>我们期望线程池以类似且熟悉的方式工作，以便从线程切换到线程池并不会对使用该 API 的代码做出较大的修改。示例 20-12 展示我们希望用来替换 <code>thread::spawn</code> 的 <code>ThreadPool</code> 结构体的假想接口：</p>
|
|
|
|
|
|
<p><span class="filename">文件名：src/main.rs</span></p>
|
|
|
|
|
|
<pre><code class="language-rust ignore does_not_compile"><span class="boring">use std::{
|
|
|
|
|
|
</span><span class="boring"> fs,
|
|
|
|
|
|
</span><span class="boring"> io::{prelude::*, BufReader},
|
|
|
|
|
|
</span><span class="boring"> net::{TcpListener, TcpStream},
|
|
|
|
|
|
</span><span class="boring"> thread,
|
|
|
|
|
|
</span><span class="boring"> time::Duration,
|
|
|
|
|
|
</span><span class="boring">};
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span>fn main() {
|
|
|
|
|
|
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
|
|
|
|
|
|
let pool = ThreadPool::new(4);
|
|
|
|
|
|
|
|
|
|
|
|
for stream in listener.incoming() {
|
|
|
|
|
|
let stream = stream.unwrap();
|
|
|
|
|
|
|
|
|
|
|
|
pool.execute(|| {
|
|
|
|
|
|
handle_connection(stream);
|
|
|
|
|
|
});
|
|
|
|
|
|
}
|
|
|
|
|
|
}
|
|
|
|
|
|
<span class="boring">
|
|
|
|
|
|
</span><span class="boring">fn handle_connection(mut stream: TcpStream) {
|
|
|
|
|
|
</span><span class="boring"> let buf_reader = BufReader::new(&mut stream);
|
|
|
|
|
|
</span><span class="boring"> let request_line = buf_reader.lines().next().unwrap().unwrap();
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let (status_line, filename) = match &request_line[..] {
|
|
|
|
|
|
</span><span class="boring"> "GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
|
|
|
|
|
|
</span><span class="boring"> "GET /sleep HTTP/1.1" => {
|
|
|
|
|
|
</span><span class="boring"> thread::sleep(Duration::from_secs(5));
|
|
|
|
|
|
</span><span class="boring"> ("HTTP/1.1 200 OK", "hello.html")
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span><span class="boring"> _ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
|
|
|
|
|
|
</span><span class="boring"> };
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let contents = fs::read_to_string(filename).unwrap();
|
|
|
|
|
|
</span><span class="boring"> let length = contents.len();
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let response =
|
|
|
|
|
|
</span><span class="boring"> format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> stream.write_all(response.as_bytes()).unwrap();
|
|
|
|
|
|
</span><span class="boring">}</span></code></pre>
|
|
|
|
|
|
<p><span class="caption">示例 20-12: 假想的 <code>ThreadPool</code> 接口</span></p>
|
|
|
|
|
|
<p>这里使用 <code>ThreadPool::new</code> 来创建一个新的线程池，它有一个可配置的线程数的参数，在这里是四。这样在 <code>for</code> 循环中，<code>pool.execute</code> 有着类似 <code>thread::spawn</code> 的接口，它获取一个线程池运行于每一个流的闭包。<code>pool.execute</code> 需要实现为获取闭包并传递给池中的线程运行。这段代码还不能编译，不过通过尝试，编译器会指导我们如何修复它。</p>
|
|
|
|
|
|
<h4 id="采用编译器驱动构建-threadpool"><a class="header" href="#采用编译器驱动构建-threadpool">采用编译器驱动构建 <code>ThreadPool</code></a></h4>
|
|
|
|
|
|
<p>继续并对示例 20-12 中的 <em>src/main.rs</em> 做出修改，并利用来自 <code>cargo check</code> 的编译器错误来驱动开发。下面是我们得到的第一个错误：</p>
|
|
|
|
|
|
<pre><code class="language-console">$ cargo check
|
|
|
|
|
|
Checking hello v0.1.0 (file:///projects/hello)
|
|
|
|
|
|
error[E0433]: failed to resolve: use of undeclared type `ThreadPool`
|
|
|
|
|
|
--> src/main.rs:11:16
|
|
|
|
|
|
|
|
|
|
|
|
|
11 | let pool = ThreadPool::new(4);
|
|
|
|
|
|
| ^^^^^^^^^^ use of undeclared type `ThreadPool`
|
|
|
|
|
|
|
|
|
|
|
|
For more information about this error, try `rustc --explain E0433`.
|
|
|
|
|
|
error: could not compile `hello` (bin "hello") due to 1 previous error
|
|
|
|
|
|
</code></pre>
|
|
|
|
|
|
<p>好的，这告诉我们需要一个 <code>ThreadPool</code> 类型或模块，所以我们将构建一个。<code>ThreadPool</code> 的实现会与 web server 的特定工作相独立，所以让我们从 <code>hello</code> crate 切换到存放 <code>ThreadPool</code> 实现的新库 crate。这也意味着可以在任何工作中使用这个单独的线程池库，而不仅仅是处理网络请求。</p>
|
|
|
|
|
|
<p>创建 <em>src/lib.rs</em> 文件，它包含了目前可用的最简单的 <code>ThreadPool</code> 定义：</p>
|
|
|
|
|
|
<p><span class="filename">文件名：src/lib.rs</span></p>
|
|
|
|
|
|
<pre><code class="language-rust noplayground">pub struct ThreadPool;</code></pre>
|
|
|
|
|
|
<p>接着编辑 <em>main.rs</em> 文件通过在 <em>src/main.rs</em> 的开头增加如下代码将 <code>ThreadPool</code> 从库 crate 引入作用域：</p>
|
|
|
|
|
|
<p><span class="filename">文件名：src/main.rs</span></p>
|
|
|
|
|
|
<pre><code class="language-rust ignore">use hello::ThreadPool;
|
|
|
|
|
|
<span class="boring">use std::{
|
|
|
|
|
|
</span><span class="boring"> fs,
|
|
|
|
|
|
</span><span class="boring"> io::{prelude::*, BufReader},
|
|
|
|
|
|
</span><span class="boring"> net::{TcpListener, TcpStream},
|
|
|
|
|
|
</span><span class="boring"> thread,
|
|
|
|
|
|
</span><span class="boring"> time::Duration,
|
|
|
|
|
|
</span><span class="boring">};
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring">fn main() {
|
|
|
|
|
|
</span><span class="boring"> let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
|
|
|
|
|
|
</span><span class="boring"> let pool = ThreadPool::new(4);
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> for stream in listener.incoming() {
|
|
|
|
|
|
</span><span class="boring"> let stream = stream.unwrap();
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> pool.execute(|| {
|
|
|
|
|
|
</span><span class="boring"> handle_connection(stream);
|
|
|
|
|
|
</span><span class="boring"> });
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span><span class="boring">}
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring">fn handle_connection(mut stream: TcpStream) {
|
|
|
|
|
|
</span><span class="boring"> let buf_reader = BufReader::new(&mut stream);
|
|
|
|
|
|
</span><span class="boring"> let request_line = buf_reader.lines().next().unwrap().unwrap();
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let (status_line, filename) = match &request_line[..] {
|
|
|
|
|
|
</span><span class="boring"> "GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
|
|
|
|
|
|
</span><span class="boring"> "GET /sleep HTTP/1.1" => {
|
|
|
|
|
|
</span><span class="boring"> thread::sleep(Duration::from_secs(5));
|
|
|
|
|
|
</span><span class="boring"> ("HTTP/1.1 200 OK", "hello.html")
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span><span class="boring"> _ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
|
|
|
|
|
|
</span><span class="boring"> };
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let contents = fs::read_to_string(filename).unwrap();
|
|
|
|
|
|
</span><span class="boring"> let length = contents.len();
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let response =
|
|
|
|
|
|
</span><span class="boring"> format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> stream.write_all(response.as_bytes()).unwrap();
|
|
|
|
|
|
</span><span class="boring">}</span></code></pre>
|
|
|
|
|
|
<p>这仍然不能工作，再次尝试运行来得到下一个需要解决的错误：</p>
|
|
|
|
|
|
<pre><code class="language-console">$ cargo check
|
|
|
|
|
|
Checking hello v0.1.0 (file:///projects/hello)
|
|
|
|
|
|
error[E0599]: no function or associated item named `new` found for struct `ThreadPool` in the current scope
|
|
|
|
|
|
--> src/main.rs:12:28
|
|
|
|
|
|
|
|
|
|
|
|
|
12 | let pool = ThreadPool::new(4);
|
|
|
|
|
|
| ^^^ function or associated item not found in `ThreadPool`
|
|
|
|
|
|
|
|
|
|
|
|
For more information about this error, try `rustc --explain E0599`.
|
|
|
|
|
|
error: could not compile `hello` (bin "hello") due to 1 previous error
|
|
|
|
|
|
</code></pre>
|
|
|
|
|
|
<p>这告诉我们下一步是为 <code>ThreadPool</code> 创建一个叫做 <code>new</code> 的关联函数。我们还知道 <code>new</code> 需要有一个参数可以接受 <code>4</code>，而且 <code>new</code> 应该返回 <code>ThreadPool</code> 实例。让我们实现拥有此特征的最小化 <code>new</code> 函数：</p>
|
|
|
|
|
|
<p><span class="filename">文件夹：src/lib.rs</span></p>
|
|
|
|
|
|
<pre><code class="language-rust noplayground">pub struct ThreadPool;
|
|
|
|
|
|
|
|
|
|
|
|
impl ThreadPool {
|
|
|
|
|
|
pub fn new(size: usize) -> ThreadPool {
|
|
|
|
|
|
ThreadPool
|
|
|
|
|
|
}
|
|
|
|
|
|
}</code></pre>
|
|
|
|
|
|
<p>这里选择 <code>usize</code> 作为 <code>size</code> 参数的类型，因为我们知道为负的线程数没有意义。我们还知道将使用 4 作为线程集合的元素数量，这也就是使用 <code>usize</code> 类型的原因，如第三章 <a href="ch03-02-data-types.html#%E6%95%B4%E5%9E%8B">“整型”</a> 部分所讲。</p>
|
|
|
|
|
|
<p>再次编译检查这段代码：</p>
|
|
|
|
|
|
<pre><code class="language-console">$ cargo check
|
|
|
|
|
|
Checking hello v0.1.0 (file:///projects/hello)
|
|
|
|
|
|
error[E0599]: no method named `execute` found for struct `ThreadPool` in the current scope
|
|
|
|
|
|
--> src/main.rs:17:14
|
|
|
|
|
|
|
|
|
|
|
|
|
17 | pool.execute(|| {
|
|
|
|
|
|
| -----^^^^^^^ method not found in `ThreadPool`
|
|
|
|
|
|
|
|
|
|
|
|
For more information about this error, try `rustc --explain E0599`.
|
|
|
|
|
|
error: could not compile `hello` (bin "hello") due to 1 previous error
|
|
|
|
|
|
</code></pre>
|
|
|
|
|
|
<p>现在有了一个警告和一个错误。暂时先忽略警告，发生错误是因为并没有 <code>ThreadPool</code> 上的 <code>execute</code> 方法。回忆 <a href="#%E5%88%9B%E5%BB%BA%E6%9C%89%E9%99%90%E6%95%B0%E9%87%8F%E7%9A%84%E7%BA%BF%E7%A8%8B">“创建有限数量的线程”</a> 部分我们决定线程池应该有与 <code>thread::spawn</code> 类似的接口，同时我们将实现 <code>execute</code> 函数来获取传递的闭包并将其传递给池中的空闲线程执行。</p>
|
|
|
|
|
|
<p>我们会在 <code>ThreadPool</code> 上定义 <code>execute</code> 函数来获取一个闭包参数。回忆第十三章的 <a href="ch13-01-closures.html#%E5%B0%86%E8%A2%AB%E6%8D%95%E8%8E%B7%E7%9A%84%E5%80%BC%E7%A7%BB%E5%87%BA%E9%97%AD%E5%8C%85%E5%92%8C-fn-trait">“将被捕获的值移出闭包和 <code>Fn</code> trait”</a> 部分，闭包作为参数时可以使用三个不同的 trait：<code>Fn</code>、<code>FnMut</code> 和 <code>FnOnce</code>。我们需要决定这里应该使用哪种闭包。最终需要实现的类似于标准库的 <code>thread::spawn</code>，所以我们可以观察 <code>thread::spawn</code> 的签名在其参数中使用了何种 bound。查看文档会发现：</p>
|
|
|
|
|
|
<pre><code class="language-rust ignore">pub fn spawn<F, T>(f: F) -> JoinHandle<T>
|
|
|
|
|
|
where
|
|
|
|
|
|
F: FnOnce() -> T,
|
|
|
|
|
|
F: Send + 'static,
|
|
|
|
|
|
T: Send + 'static,</code></pre>
|
|
|
|
|
|
<p><code>F</code> 是这里我们关心的参数；<code>T</code> 与返回值有关所以我们并不关心。考虑到 <code>spawn</code> 使用 <code>FnOnce</code> 作为 <code>F</code> 的 trait bound，这可能也是我们需要的，因为最终会将传递给 <code>execute</code> 的参数传给 <code>spawn</code>。因为处理请求的线程只会执行闭包一次，这也进一步确认了 <code>FnOnce</code> 是我们需要的 trait，这里符合 <code>FnOnce</code> 中 <code>Once</code> 的意思。</p>
|
|
|
|
|
|
<p><code>F</code> 还有 trait bound <code>Send</code> 和生命周期绑定 <code>'static</code>，这对我们的情况也是有意义的：需要 <code>Send</code> 来将闭包从一个线程转移到另一个线程，而 <code>'static</code> 是因为并不知道线程会执行多久。让我们编写一个使用带有这些 bound 的泛型参数 <code>F</code> 的 <code>ThreadPool</code> 的 <code>execute</code> 方法：</p>
|
|
|
|
|
|
<p><span class="filename">文件名：src/lib.rs</span></p>
|
|
|
|
|
|
<pre><code class="language-rust noplayground"><span class="boring">pub struct ThreadPool;
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span>impl ThreadPool {
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
<span class="boring"> pub fn new(size: usize) -> ThreadPool {
|
|
|
|
|
|
</span><span class="boring"> ThreadPool
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span> pub fn execute<F>(&self, f: F)
|
|
|
|
|
|
where
|
|
|
|
|
|
F: FnOnce() + Send + 'static,
|
|
|
|
|
|
{
|
|
|
|
|
|
}
|
|
|
|
|
|
}</code></pre>
|
|
|
|
|
|
<p><code>FnOnce</code> trait 仍然需要之后的 <code>()</code>，因为这里的 <code>FnOnce</code> 代表一个没有参数也没有返回值的闭包。正如函数的定义，返回值类型可以从签名中省略，不过即便没有参数也需要括号。</p>
|
|
|
|
|
|
<p>这里再一次增加了 <code>execute</code> 方法的最小化实现：它没有做任何工作，只是尝试让代码能够编译。再次进行检查：</p>
|
|
|
|
|
|
<pre><code class="language-console">$ cargo check
|
|
|
|
|
|
Checking hello v0.1.0 (file:///projects/hello)
|
|
|
|
|
|
Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.24s
|
|
|
|
|
|
</code></pre>
|
|
|
|
|
|
<p>现在就只有警告了！这意味着能够编译了！注意如果尝试 <code>cargo run</code> 运行程序并在浏览器中发起请求，仍会在浏览器中出现在本章开始时那样的错误。这个库实际上还没有调用传递给 <code>execute</code> 的闭包！</p>
|
|
|
|
|
|
<blockquote>
|
|
|
|
|
|
<p>一个你可能听说过的关于像 Haskell 和 Rust 这样有严格编译器的语言的说法是 “如果代码能够编译，它就能工作”。这是一个提醒大家的好时机，实际上这并不是普适的。我们的项目可以编译，不过它完全没有做任何工作！如果构建一个真实且功能完整的项目，则需花费大量的时间来开始编写单元测试来检查代码能否编译 <strong>并且</strong> 拥有期望的行为。</p>
|
|
|
|
|
|
</blockquote>
|
|
|
|
|
|
<h4 id="在-new-中验证池中线程数量"><a class="header" href="#在-new-中验证池中线程数量">在 <code>new</code> 中验证池中线程数量</a></h4>
|
|
|
|
|
|
<p>这里仍然存在警告是因为其并没有对 <code>new</code> 和 <code>execute</code> 的参数做任何操作。让我们用期望的行为来实现这些函数。以考虑 <code>new</code> 作为开始。之前选择使用无符号类型作为 <code>size</code> 参数的类型，因为线程数为负的线程池没有意义。然而，线程数为零的线程池同样没有意义，不过零是一个完全有效的 <code>usize</code> 值。让我们增加在返回 <code>ThreadPool</code> 实例之前检查 <code>size</code> 是否大于零的代码，并使用 <code>assert!</code> 宏在得到零时 panic，如示例 20-13 所示：</p>
|
|
|
|
|
|
<p><span class="filename">文件名：src/lib.rs</span></p>
|
|
|
|
|
|
<pre><code class="language-rust noplayground"><span class="boring">pub struct ThreadPool;
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span>impl ThreadPool {
|
|
|
|
|
|
/// 创建线程池。
|
|
|
|
|
|
///
|
|
|
|
|
|
/// 线程池中线程的数量。
|
|
|
|
|
|
///
|
|
|
|
|
|
/// # Panics
|
|
|
|
|
|
///
|
|
|
|
|
|
/// `new` 函数在 size 为 0 时会 panic。
|
|
|
|
|
|
pub fn new(size: usize) -> ThreadPool {
|
|
|
|
|
|
assert!(size > 0);
|
|
|
|
|
|
|
|
|
|
|
|
ThreadPool
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
<span class="boring"> pub fn execute<F>(&self, f: F)
|
|
|
|
|
|
</span><span class="boring"> where
|
|
|
|
|
|
</span><span class="boring"> F: FnOnce() + Send + 'static,
|
|
|
|
|
|
</span><span class="boring"> {
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span>}</code></pre>
|
|
|
|
|
|
<p><span class="caption">示例 20-13: 实现 <code>ThreadPool::new</code> 在 <code>size</code> 为零时 panic</span></p>
|
|
|
|
|
|
<p>这里也用文档注释为 <code>ThreadPool</code> 增加了一些文档。注意这里遵循了良好的文档实践并增加了一个部分来提示函数会 panic 的情况，正如第十四章所讨论的。尝试运行 <code>cargo doc --open</code> 并点击 <code>ThreadPool</code> 结构体来查看生成的 <code>new</code> 的文档看起来如何！</p>
|
|
|
|
|
|
<p>相比像这里使用 <code>assert!</code> 宏，也可以让 <code>new</code> 像之前 I/O 项目中示例 12-9 中 <code>Config::build</code> 那样将 <code>new</code> 更改为 <code>build</code> 并返回一个 <code>Result</code>，不过在这里我们选择创建一个没有任何线程的线程池应该是不可恢复的错误。如果你想做的更好，尝试编写一个采用如下签名的名为 <code>build</code> 的函数来对比一下 <code>new</code> 函数：</p>
|
|
|
|
|
|
<pre><code class="language-rust ignore">pub fn build(size: usize) -> Result<ThreadPool, PoolCreationError> {</code></pre>
|
|
|
|
|
|
<h4 id="分配空间以储存线程"><a class="header" href="#分配空间以储存线程">分配空间以储存线程</a></h4>
|
|
|
|
|
|
<p>现在有了一个有效的线程池线程数，就可以实际创建这些线程并在返回结构体之前将它们储存在 <code>ThreadPool</code> 结构体中。不过如何 “储存” 一个线程？让我们再看看 <code>thread::spawn</code> 的签名：</p>
|
|
|
|
|
|
<pre><code class="language-rust ignore">pub fn spawn<F, T>(f: F) -> JoinHandle<T>
|
|
|
|
|
|
where
|
|
|
|
|
|
F: FnOnce() -> T,
|
|
|
|
|
|
F: Send + 'static,
|
|
|
|
|
|
T: Send + 'static,</code></pre>
|
|
|
|
|
|
<p><code>spawn</code> 返回 <code>JoinHandle<T></code>，其中 <code>T</code> 是闭包返回的类型。尝试使用 <code>JoinHandle</code> 来看看会发生什么。在我们的情况中，传递给线程池的闭包会处理连接并不返回任何值，所以 <code>T</code> 将会是单元类型 <code>()</code>。</p>
|
|
|
|
|
|
<p>示例 20-14 中的代码可以编译，不过实际上还并没有创建任何线程。我们改变了 <code>ThreadPool</code> 的定义来存放一个 <code>thread::JoinHandle<()></code> 的 vector 实例，使用 <code>size</code> 容量来初始化，并设置一个 <code>for</code> 循环来运行创建线程的代码，并返回包含这些线程的 <code>ThreadPool</code> 实例：</p>
|
|
|
|
|
|
<p><span class="filename">文件名：src/lib.rs</span></p>
|
|
|
|
|
|
<pre><code class="language-rust ignore not_desired_behavior">use std::thread;
|
|
|
|
|
|
|
|
|
|
|
|
pub struct ThreadPool {
|
|
|
|
|
|
threads: Vec<thread::JoinHandle<()>>,
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
impl ThreadPool {
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
<span class="boring"> /// Create a new ThreadPool.
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// The size is the number of threads in the pool.
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// # Panics
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// The `new` function will panic if the size is zero.
|
|
|
|
|
|
</span> pub fn new(size: usize) -> ThreadPool {
|
|
|
|
|
|
assert!(size > 0);
|
|
|
|
|
|
|
|
|
|
|
|
let mut threads = Vec::with_capacity(size);
|
|
|
|
|
|
|
|
|
|
|
|
for _ in 0..size {
|
|
|
|
|
|
// create some threads and store them in the vector
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
ThreadPool { threads }
|
|
|
|
|
|
}
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
<span class="boring">
|
|
|
|
|
|
</span><span class="boring"> pub fn execute<F>(&self, f: F)
|
|
|
|
|
|
</span><span class="boring"> where
|
|
|
|
|
|
</span><span class="boring"> F: FnOnce() + Send + 'static,
|
|
|
|
|
|
</span><span class="boring"> {
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span>}</code></pre>
|
|
|
|
|
|
<p><span class="caption">示例 20-14: 为 <code>ThreadPool</code> 创建一个 vector 来存放线程</span></p>
|
|
|
|
|
|
<p>这里将 <code>std::thread</code> 引入库 crate 的作用域，因为使用了 <code>thread::JoinHandle</code> 作为 <code>ThreadPool</code> 中 vector 元素的类型。</p>
|
|
|
|
|
|
<p>在得到了有效的数量之后，<code>ThreadPool</code> 新建一个存放 <code>size</code> 个元素的 vector。<code>with_capacity</code> 函数与 <code>Vec::new</code> 做了同样的工作，不过有一个重要的区别：它为 vector 预先分配空间。因为已经知道了 vector 中需要 <code>size</code> 个元素，预先进行分配比仅仅 <code>Vec::new</code> 要稍微有效率一些，因为 <code>Vec::new</code> 随着插入元素而重新改变大小。</p>
|
|
|
|
|
|
<p>如果再次运行 <code>cargo check</code>，它应该会成功。</p>
|
|
|
|
|
|
<h4 id="worker-结构体负责从-threadpool-中将代码传递给线程"><a class="header" href="#worker-结构体负责从-threadpool-中将代码传递给线程"><code>Worker</code> 结构体负责从 <code>ThreadPool</code> 中将代码传递给线程</a></h4>
|
|
|
|
|
|
<p>示例 20-14 的 <code>for</code> 循环中留下了一个关于创建线程的注释。如何实际创建线程呢？这是一个难题。标准库提供的创建线程的方法，<code>thread::spawn</code>，它期望获取一些一旦创建线程就应该执行的代码。然而，我们希望开始线程并使其等待稍后传递的代码。标准库的线程实现并没有包含这么做的方法；我们必须自己实现。</p>
|
|
|
|
|
|
<p>我们将要实现的行为是创建线程并稍后发送代码，这会在 <code>ThreadPool</code> 和线程间引入一个新数据类型来管理这种新行为。这个数据结构称为 <em>Worker</em>，这是一个池实现中的常见概念。想象一下在餐馆厨房工作的员工：员工等待来自客户的订单，他们负责接受这些订单并完成它们。</p>
|
|
|
|
|
|
<p>不同于在线程池中储存一个 <code>JoinHandle<()></code> 实例的 vector，我们会储存 <code>Worker</code> 结构体的实例。每一个 <code>Worker</code> 会储存一个单独的 <code>JoinHandle<()></code> 实例。接着会在 <code>Worker</code> 上实现一个方法，它会获取需要允许代码的闭包并将其发送给已经运行的线程执行。我们还会赋予每一个 worker <code>id</code>，这样就可以在日志和调试中区别线程池中的不同 worker。</p>
|
|
|
|
|
|
<p>如下是创建 <code>ThreadPool</code> 时会发生的新过程。在通过如下方式设置完 <code>Worker</code> 之后，我们会实现向线程发送闭包的代码：</p>
|
|
|
|
|
|
<ol>
|
|
|
|
|
|
<li>定义 <code>Worker</code> 结构体存放 <code>id</code> 和 <code>JoinHandle<()></code></li>
|
|
|
|
|
|
<li>修改 <code>ThreadPool</code> 存放一个 <code>Worker</code> 实例的 vector</li>
|
|
|
|
|
|
<li>定义 <code>Worker::new</code> 函数，它获取一个 <code>id</code> 数字并返回一个带有 <code>id</code> 和用空闭包分配的线程的 <code>Worker</code> 实例</li>
|
|
|
|
|
|
<li>在 <code>ThreadPool::new</code> 中，使用 <code>for</code> 循环计数生成 <code>id</code>，使用这个 <code>id</code> 新建 <code>Worker</code>，并储存进 vector 中</li>
|
|
|
|
|
|
</ol>
|
|
|
|
|
|
<p>如果你渴望挑战，在查示例 20-15 中的代码之前尝试自己实现这些修改。</p>
|
|
|
|
|
|
<p>准备好了吗？示例 20-15 就是一个做出了这些修改的例子：</p>
|
|
|
|
|
|
<p><span class="filename">文件名：src/lib.rs</span></p>
|
|
|
|
|
|
<pre><code class="language-rust noplayground">use std::thread;
|
|
|
|
|
|
|
|
|
|
|
|
pub struct ThreadPool {
|
|
|
|
|
|
workers: Vec<Worker>,
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
impl ThreadPool {
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
<span class="boring"> /// Create a new ThreadPool.
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// The size is the number of threads in the pool.
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// # Panics
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// The `new` function will panic if the size is zero.
|
|
|
|
|
|
</span> pub fn new(size: usize) -> ThreadPool {
|
|
|
|
|
|
assert!(size > 0);
|
|
|
|
|
|
|
|
|
|
|
|
let mut workers = Vec::with_capacity(size);
|
|
|
|
|
|
|
|
|
|
|
|
for id in 0..size {
|
|
|
|
|
|
workers.push(Worker::new(id));
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
ThreadPool { workers }
|
|
|
|
|
|
}
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
<span class="boring">
|
|
|
|
|
|
</span><span class="boring"> pub fn execute<F>(&self, f: F)
|
|
|
|
|
|
</span><span class="boring"> where
|
|
|
|
|
|
</span><span class="boring"> F: FnOnce() + Send + 'static,
|
|
|
|
|
|
</span><span class="boring"> {
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span>}
|
|
|
|
|
|
|
|
|
|
|
|
struct Worker {
|
|
|
|
|
|
id: usize,
|
|
|
|
|
|
thread: thread::JoinHandle<()>,
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
impl Worker {
|
|
|
|
|
|
fn new(id: usize) -> Worker {
|
|
|
|
|
|
let thread = thread::spawn(|| {});
|
|
|
|
|
|
|
|
|
|
|
|
Worker { id, thread }
|
|
|
|
|
|
}
|
|
|
|
|
|
}</code></pre>
|
|
|
|
|
|
<p><span class="caption">示例 20-15: 修改 <code>ThreadPool</code> 存放 <code>Worker</code> 实例而不是直接存放线程</span></p>
|
|
|
|
|
|
<p>这里将 <code>ThreadPool</code> 中字段名从 <code>threads</code> 改为 <code>workers</code>，因为它现在储存 <code>Worker</code> 而不是 <code>JoinHandle<()></code>。使用 <code>for</code> 循环中的计数作为 <code>Worker::new</code> 的参数，并将每一个新建的 <code>Worker</code> 储存在叫做 <code>workers</code> 的 vector 中。</p>
|
|
|
|
|
|
<p><code>Worker</code> 结构体和其 <code>new</code> 函数是私有的，因为外部代码（比如 <em>src/main.rs</em> 中的 server）并不需要知道关于 <code>ThreadPool</code> 中使用 <code>Worker</code> 结构体的实现细节。<code>Worker::new</code> 函数使用 <code>id</code> 参数并储存了使用一个空闭包创建的 <code>JoinHandle<()></code>。</p>
|
|
|
|
|
|
<blockquote>
|
|
|
|
|
|
<p>注意：如果操作系统因为没有足够的系统资源而无法创建线程时，<code>thread::spawn</code> 会 panic。这会导致整个 server panic，即使一些线程可能创建成功了。出于简单的考虑，这个行为是可行的，不过在一个生产级别的线程池实现中，你可能会希望使用 <a href="https://doc.rust-lang.org/std/thread/struct.Builder.html"><code>std::thread::Builder</code></a> 和其 <a href="https://doc.rust-lang.org/std/thread/struct.Builder.html#method.spawn"><code>spawn</code></a> 方法来返回一个 <code>Result</code>。</p>
|
|
|
|
|
|
</blockquote>
|
|
|
|
|
|
<p>这段代码能够编译并用指定给 <code>ThreadPool::new</code> 的参数创建储存了一系列的 <code>Worker</code> 实例，不过 <strong>仍然</strong> 没有处理 <code>execute</code> 中得到的闭包。让我们聊聊接下来怎么做。</p>
|
|
|
|
|
|
<h4 id="使用信道向线程发送请求"><a class="header" href="#使用信道向线程发送请求">使用信道向线程发送请求</a></h4>
|
|
|
|
|
|
<p>下一个需要解决的问题是传递给 <code>thread::spawn</code> 的闭包完全没有做任何工作。目前，我们在 <code>execute</code> 方法中获得期望执行的闭包，不过在创建 <code>ThreadPool</code> 的过程中创建每一个 <code>Worker</code> 时需要向 <code>thread::spawn</code> 传递一个闭包。</p>
|
|
|
|
|
|
<p>我们希望刚创建的 <code>Worker</code> 结构体能够从 <code>ThreadPool</code> 的队列中获取需要执行的代码，并发送到线程中执行它们。</p>
|
|
|
|
|
|
<p>在第十六章，我们学习了 <strong>信道</strong> —— 一个沟通两个线程的简单手段 —— 对于这个例子来说则是绝佳的。这里信道将充当任务队列的作用，<code>execute</code> 将通过 <code>ThreadPool</code> 向其中线程正在寻找工作的 <code>Worker</code> 实例发送任务。如下是这个计划：</p>
|
|
|
|
|
|
<ol>
|
|
|
|
|
|
<li><code>ThreadPool</code> 会创建一个信道并充当发送者。</li>
|
|
|
|
|
|
<li>每个 <code>Worker</code> 将会充当接收者。</li>
|
|
|
|
|
|
<li>新建一个 <code>Job</code> 结构体来存放用于向信道中发送的闭包。</li>
|
|
|
|
|
|
<li><code>execute</code> 方法会在发送者发出期望执行的任务。</li>
|
|
|
|
|
|
<li>在线程中，<code>Worker</code> 会遍历接收者并执行任何接收到的任务。</li>
|
|
|
|
|
|
</ol>
|
|
|
|
|
|
<p>让我们以在 <code>ThreadPool::new</code> 中创建信道并让 <code>ThreadPool</code> 实例充当发送者开始，如示例 20-16 所示。<code>Job</code> 是将在信道中发出的类型，目前它是一个没有任何内容的结构体：</p>
|
|
|
|
|
|
<p><span class="filename">文件名：src/lib.rs</span></p>
|
|
|
|
|
|
<pre><code class="language-rust noplayground">use std::{sync::mpsc, thread};
|
|
|
|
|
|
|
|
|
|
|
|
pub struct ThreadPool {
|
|
|
|
|
|
workers: Vec<Worker>,
|
|
|
|
|
|
sender: mpsc::Sender<Job>,
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
struct Job;
|
|
|
|
|
|
|
|
|
|
|
|
impl ThreadPool {
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
<span class="boring"> /// Create a new ThreadPool.
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// The size is the number of threads in the pool.
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// # Panics
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// The `new` function will panic if the size is zero.
|
|
|
|
|
|
</span> pub fn new(size: usize) -> ThreadPool {
|
|
|
|
|
|
assert!(size > 0);
|
|
|
|
|
|
|
|
|
|
|
|
let (sender, receiver) = mpsc::channel();
|
|
|
|
|
|
|
|
|
|
|
|
let mut workers = Vec::with_capacity(size);
|
|
|
|
|
|
|
|
|
|
|
|
for id in 0..size {
|
|
|
|
|
|
workers.push(Worker::new(id));
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
ThreadPool { workers, sender }
|
|
|
|
|
|
}
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
<span class="boring">
|
|
|
|
|
|
</span><span class="boring"> pub fn execute<F>(&self, f: F)
|
|
|
|
|
|
</span><span class="boring"> where
|
|
|
|
|
|
</span><span class="boring"> F: FnOnce() + Send + 'static,
|
|
|
|
|
|
</span><span class="boring"> {
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span>}
|
|
|
|
|
|
<span class="boring">
|
|
|
|
|
|
</span><span class="boring">struct Worker {
|
|
|
|
|
|
</span><span class="boring"> id: usize,
|
|
|
|
|
|
</span><span class="boring"> thread: thread::JoinHandle<()>,
|
|
|
|
|
|
</span><span class="boring">}
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring">impl Worker {
|
|
|
|
|
|
</span><span class="boring"> fn new(id: usize) -> Worker {
|
|
|
|
|
|
</span><span class="boring"> let thread = thread::spawn(|| {});
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> Worker { id, thread }
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span><span class="boring">}</span></code></pre>
|
|
|
|
|
|
<p><span class="caption">示例 20-16: 修改 <code>ThreadPool</code> 来储存一个传输 <code>Job</code> 实例的发送者</span></p>
|
|
|
|
|
|
<p>在 <code>ThreadPool::new</code> 中，新建了一个信道，并接着让线程池在接收端等待。这段代码能够成功编译。</p>
|
|
|
|
|
|
<p>让我们尝试在线程池创建每个 worker 时将接收者传递给它们。须知我们希望在 worker 所分配的线程中使用接收者，所以将在闭包中引用 <code>receiver</code> 参数。示例 20-17 中展示的代码还不能编译：</p>
|
|
|
|
|
|
<p><span class="filename">文件名：src/lib.rs</span></p>
|
|
|
|
|
|
<pre><code class="language-rust ignore does_not_compile"><span class="boring">use std::{sync::mpsc, thread};
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring">pub struct ThreadPool {
|
|
|
|
|
|
</span><span class="boring"> workers: Vec<Worker>,
|
|
|
|
|
|
</span><span class="boring"> sender: mpsc::Sender<Job>,
|
|
|
|
|
|
</span><span class="boring">}
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring">struct Job;
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span>impl ThreadPool {
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
<span class="boring"> /// Create a new ThreadPool.
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// The size is the number of threads in the pool.
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// # Panics
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// The `new` function will panic if the size is zero.
|
|
|
|
|
|
</span> pub fn new(size: usize) -> ThreadPool {
|
|
|
|
|
|
assert!(size > 0);
|
|
|
|
|
|
|
|
|
|
|
|
let (sender, receiver) = mpsc::channel();
|
|
|
|
|
|
|
|
|
|
|
|
let mut workers = Vec::with_capacity(size);
|
|
|
|
|
|
|
|
|
|
|
|
for id in 0..size {
|
|
|
|
|
|
workers.push(Worker::new(id, receiver));
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
ThreadPool { workers, sender }
|
|
|
|
|
|
}
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
<span class="boring">
|
|
|
|
|
|
</span><span class="boring"> pub fn execute<F>(&self, f: F)
|
|
|
|
|
|
</span><span class="boring"> where
|
|
|
|
|
|
</span><span class="boring"> F: FnOnce() + Send + 'static,
|
|
|
|
|
|
</span><span class="boring"> {
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span>}
|
|
|
|
|
|
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
|
|
|
|
|
|
<span class="boring">
|
|
|
|
|
|
</span><span class="boring">struct Worker {
|
|
|
|
|
|
</span><span class="boring"> id: usize,
|
|
|
|
|
|
</span><span class="boring"> thread: thread::JoinHandle<()>,
|
|
|
|
|
|
</span><span class="boring">}
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span>impl Worker {
|
|
|
|
|
|
fn new(id: usize, receiver: mpsc::Receiver<Job>) -> Worker {
|
|
|
|
|
|
let thread = thread::spawn(|| {
|
|
|
|
|
|
receiver;
|
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
|
|
Worker { id, thread }
|
|
|
|
|
|
}
|
|
|
|
|
|
}</code></pre>
|
|
|
|
|
|
<p><span class="caption">示例 20-17: 将信道的接收端传递给 worker</span></p>
|
|
|
|
|
|
<p>这是一些小而直观的修改：将接收者传递进了 <code>Worker::new</code>，并接着在闭包中使用它。</p>
|
|
|
|
|
|
<p>如果尝试 check 代码，会得到这个错误：</p>
|
|
|
|
|
|
<pre><code class="language-console">$ cargo check
|
|
|
|
|
|
Checking hello v0.1.0 (file:///projects/hello)
|
|
|
|
|
|
error[E0382]: use of moved value: `receiver`
|
|
|
|
|
|
--> src/lib.rs:26:42
|
|
|
|
|
|
|
|
|
|
|
|
|
21 | let (sender, receiver) = mpsc::channel();
|
|
|
|
|
|
| -------- move occurs because `receiver` has type `std::sync::mpsc::Receiver<Job>`, which does not implement the `Copy` trait
|
|
|
|
|
|
...
|
|
|
|
|
|
25 | for id in 0..size {
|
|
|
|
|
|
| ----------------- inside of this loop
|
|
|
|
|
|
26 | workers.push(Worker::new(id, receiver));
|
|
|
|
|
|
| ^^^^^^^^ value moved here, in previous iteration of loop
|
|
|
|
|
|
|
|
|
|
|
|
|
note: consider changing this parameter type in method `new` to borrow instead if owning the value isn't necessary
|
|
|
|
|
|
--> src/lib.rs:47:33
|
|
|
|
|
|
|
|
|
|
|
|
|
47 | fn new(id: usize, receiver: mpsc::Receiver<Job>) -> Worker {
|
|
|
|
|
|
| --- in this method ^^^^^^^^^^^^^^^^^^^ this parameter takes ownership of the value
|
|
|
|
|
|
|
|
|
|
|
|
For more information about this error, try `rustc --explain E0382`.
|
|
|
|
|
|
error: could not compile `hello` (lib) due to 1 previous error
|
|
|
|
|
|
</code></pre>
|
|
|
|
|
|
<p>这段代码尝试将 <code>receiver</code> 传递给多个 <code>Worker</code> 实例。这是不行的，回忆第十六章：Rust 所提供的信道实现是多 <strong>生产者</strong>，单 <strong>消费者</strong> 的。这意味着不能简单的克隆信道的消费端来解决问题。我们也不希望将一个消息向多个消费者发送多次；我们希望有一个消息列表和多个 worker 这样每个消息就只会处理一次。</p>
|
|
|
|
|
|
<p>另外，从信道队列中取出任务涉及到修改 <code>receiver</code>，所以这些线程需要一个能安全的共享和修改 <code>receiver</code> 的方式，否则可能导致竞争状态（参考第十六章）。</p>
|
|
|
|
|
|
<p>回忆一下第十六章讨论的线程安全智能指针，为了在多个线程间共享所有权并允许线程修改其值，需要使用 <code>Arc<Mutex<T>></code>。<code>Arc</code> 使得多个 worker 拥有接收端，而 <code>Mutex</code> 则确保一次只有一个 worker 能从接收端得到任务。示例 20-18 展示了所需的修改：</p>
|
|
|
|
|
|
<p><span class="filename">文件名：src/lib.rs</span></p>
|
|
|
|
|
|
<pre><code class="language-rust noplayground">use std::{
|
|
|
|
|
|
sync::{mpsc, Arc, Mutex},
|
|
|
|
|
|
thread,
|
|
|
|
|
|
};
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
|
|
|
|
|
|
<span class="boring">pub struct ThreadPool {
|
|
|
|
|
|
</span><span class="boring"> workers: Vec<Worker>,
|
|
|
|
|
|
</span><span class="boring"> sender: mpsc::Sender<Job>,
|
|
|
|
|
|
</span><span class="boring">}
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring">struct Job;
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span>impl ThreadPool {
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
<span class="boring"> /// Create a new ThreadPool.
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// The size is the number of threads in the pool.
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// # Panics
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// The `new` function will panic if the size is zero.
|
|
|
|
|
|
</span> pub fn new(size: usize) -> ThreadPool {
|
|
|
|
|
|
assert!(size > 0);
|
|
|
|
|
|
|
|
|
|
|
|
let (sender, receiver) = mpsc::channel();
|
|
|
|
|
|
|
|
|
|
|
|
let receiver = Arc::new(Mutex::new(receiver));
|
|
|
|
|
|
|
|
|
|
|
|
let mut workers = Vec::with_capacity(size);
|
|
|
|
|
|
|
|
|
|
|
|
for id in 0..size {
|
|
|
|
|
|
workers.push(Worker::new(id, Arc::clone(&receiver)));
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
ThreadPool { workers, sender }
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
<span class="boring">
|
|
|
|
|
|
</span><span class="boring"> pub fn execute<F>(&self, f: F)
|
|
|
|
|
|
</span><span class="boring"> where
|
|
|
|
|
|
</span><span class="boring"> F: FnOnce() + Send + 'static,
|
|
|
|
|
|
</span><span class="boring"> {
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span>}
|
|
|
|
|
|
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
|
|
|
|
|
|
<span class="boring">struct Worker {
|
|
|
|
|
|
</span><span class="boring"> id: usize,
|
|
|
|
|
|
</span><span class="boring"> thread: thread::JoinHandle<()>,
|
|
|
|
|
|
</span><span class="boring">}
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span>impl Worker {
|
|
|
|
|
|
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
<span class="boring"> let thread = thread::spawn(|| {
|
|
|
|
|
|
</span><span class="boring"> receiver;
|
|
|
|
|
|
</span><span class="boring"> });
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> Worker { id, thread }
|
|
|
|
|
|
</span> }
|
|
|
|
|
|
}</code></pre>
|
|
|
|
|
|
<p><span class="caption">示例 20-18: 使用 <code>Arc</code> 和 <code>Mutex</code> 在 worker 间共享接收者</span></p>
|
|
|
|
|
|
<p>在 <code>ThreadPool::new</code> 中，将接收者放入一个 <code>Arc</code> 和一个 <code>Mutex</code> 中。对于每一个新 worker，克隆 <code>Arc</code> 来增加引用计数，如此这些 worker 就可以共享接收者的所有权了。</p>
|
|
|
|
|
|
<p>通过这些修改，代码可以编译了！我们做到了！</p>
|
|
|
|
|
|
<h4 id="实现-execute-方法"><a class="header" href="#实现-execute-方法">实现 <code>execute</code> 方法</a></h4>
|
|
|
|
|
|
<p>最后让我们实现 <code>ThreadPool</code> 上的 <code>execute</code> 方法。同时也要修改 <code>Job</code> 结构体：它将不再是结构体，<code>Job</code> 将是一个有着 <code>execute</code> 接收到的闭包类型的 trait 对象的类型别名。第十九章 <a href="ch19-04-advanced-types.html#%E7%B1%BB%E5%9E%8B%E5%88%AB%E5%90%8D%E7%94%A8%E6%9D%A5%E5%88%9B%E5%BB%BA%E7%B1%BB%E5%9E%8B%E5%90%8C%E4%B9%89%E8%AF%8D">“类型别名用来创建类型同义词”</a> 部分提到过，类型别名允许将长的类型变短。观察示例 20-19：</p>
|
|
|
|
|
|
<p><span class="filename">文件名：src/lib.rs</span></p>
|
|
|
|
|
|
<pre><code class="language-rust noplayground"><span class="boring">use std::{
|
|
|
|
|
|
</span><span class="boring"> sync::{mpsc, Arc, Mutex},
|
|
|
|
|
|
</span><span class="boring"> thread,
|
|
|
|
|
|
</span><span class="boring">};
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring">pub struct ThreadPool {
|
|
|
|
|
|
</span><span class="boring"> workers: Vec<Worker>,
|
|
|
|
|
|
</span><span class="boring"> sender: mpsc::Sender<Job>,
|
|
|
|
|
|
</span><span class="boring">}
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span>// --snip--
|
|
|
|
|
|
|
|
|
|
|
|
type Job = Box<dyn FnOnce() + Send + 'static>;
|
|
|
|
|
|
|
|
|
|
|
|
impl ThreadPool {
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
<span class="boring"> /// Create a new ThreadPool.
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// The size is the number of threads in the pool.
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// # Panics
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// The `new` function will panic if the size is zero.
|
|
|
|
|
|
</span><span class="boring"> pub fn new(size: usize) -> ThreadPool {
|
|
|
|
|
|
</span><span class="boring"> assert!(size > 0);
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let (sender, receiver) = mpsc::channel();
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let receiver = Arc::new(Mutex::new(receiver));
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let mut workers = Vec::with_capacity(size);
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> for id in 0..size {
|
|
|
|
|
|
</span><span class="boring"> workers.push(Worker::new(id, Arc::clone(&receiver)));
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> ThreadPool { workers, sender }
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span>
|
|
|
|
|
|
pub fn execute<F>(&self, f: F)
|
|
|
|
|
|
where
|
|
|
|
|
|
F: FnOnce() + Send + 'static,
|
|
|
|
|
|
{
|
|
|
|
|
|
let job = Box::new(f);
|
|
|
|
|
|
|
|
|
|
|
|
self.sender.send(job).unwrap();
|
|
|
|
|
|
}
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
// --snip--
|
|
|
|
|
|
<span class="boring">
|
|
|
|
|
|
</span><span class="boring">struct Worker {
|
|
|
|
|
|
</span><span class="boring"> id: usize,
|
|
|
|
|
|
</span><span class="boring"> thread: thread::JoinHandle<()>,
|
|
|
|
|
|
</span><span class="boring">}
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring">impl Worker {
|
|
|
|
|
|
</span><span class="boring"> fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
|
|
|
|
|
|
</span><span class="boring"> let thread = thread::spawn(|| {
|
|
|
|
|
|
</span><span class="boring"> receiver;
|
|
|
|
|
|
</span><span class="boring"> });
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> Worker { id, thread }
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span><span class="boring">}</span></code></pre>
|
|
|
|
|
|
<p><span class="caption">示例 20-19: 为存放每一个闭包的 <code>Box</code> 创建一个 <code>Job</code> 类型别名，接着在信道中发出任务</span></p>
|
|
|
|
|
|
<p>在使用 <code>execute</code> 得到的闭包新建 <code>Job</code> 实例之后，将这些任务从信道的发送端发出。这里调用 <code>send</code> 上的 <code>unwrap</code>，因为发送可能会失败，这可能发生于例如停止了所有线程执行的情况，这意味着接收端停止接收新消息了。不过目前我们无法停止线程执行；只要线程池存在它们就会一直执行。使用 <code>unwrap</code> 是因为我们知道失败不可能发生，即便编译器不这么认为。</p>
|
|
|
|
|
|
<p>不过到此事情还没有结束！在 worker 中，传递给 <code>thread::spawn</code> 的闭包仍然还只是 <strong>引用</strong> 了信道的接收端。相反我们需要闭包一直循环，向信道的接收端请求任务，并在得到任务时执行它们。如示例 20-20 对 <code>Worker::new</code> 做出修改：</p>
|
|
|
|
|
|
<p><span class="filename">文件名：src/lib.rs</span></p>
|
|
|
|
|
|
<pre><code class="language-rust noplayground"><span class="boring">use std::{
|
|
|
|
|
|
</span><span class="boring"> sync::{mpsc, Arc, Mutex},
|
|
|
|
|
|
</span><span class="boring"> thread,
|
|
|
|
|
|
</span><span class="boring">};
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring">pub struct ThreadPool {
|
|
|
|
|
|
</span><span class="boring"> workers: Vec<Worker>,
|
|
|
|
|
|
</span><span class="boring"> sender: mpsc::Sender<Job>,
|
|
|
|
|
|
</span><span class="boring">}
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring">type Job = Box<dyn FnOnce() + Send + 'static>;
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring">impl ThreadPool {
|
|
|
|
|
|
</span><span class="boring"> /// Create a new ThreadPool.
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// The size is the number of threads in the pool.
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// # Panics
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// The `new` function will panic if the size is zero.
|
|
|
|
|
|
</span><span class="boring"> pub fn new(size: usize) -> ThreadPool {
|
|
|
|
|
|
</span><span class="boring"> assert!(size > 0);
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let (sender, receiver) = mpsc::channel();
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let receiver = Arc::new(Mutex::new(receiver));
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let mut workers = Vec::with_capacity(size);
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> for id in 0..size {
|
|
|
|
|
|
</span><span class="boring"> workers.push(Worker::new(id, Arc::clone(&receiver)));
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> ThreadPool { workers, sender }
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> pub fn execute<F>(&self, f: F)
|
|
|
|
|
|
</span><span class="boring"> where
|
|
|
|
|
|
</span><span class="boring"> F: FnOnce() + Send + 'static,
|
|
|
|
|
|
</span><span class="boring"> {
|
|
|
|
|
|
</span><span class="boring"> let job = Box::new(f);
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> self.sender.send(job).unwrap();
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span><span class="boring">}
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring">struct Worker {
|
|
|
|
|
|
</span><span class="boring"> id: usize,
|
|
|
|
|
|
</span><span class="boring"> thread: thread::JoinHandle<()>,
|
|
|
|
|
|
</span><span class="boring">}
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span>// --snip--
|
|
|
|
|
|
|
|
|
|
|
|
impl Worker {
|
|
|
|
|
|
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
|
|
|
|
|
|
let thread = thread::spawn(move || loop {
|
|
|
|
|
|
let job = receiver.lock().unwrap().recv().unwrap();
|
|
|
|
|
|
|
|
|
|
|
|
println!("Worker {id} got a job; executing.");
|
|
|
|
|
|
|
|
|
|
|
|
job();
|
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
|
|
Worker { id, thread }
|
|
|
|
|
|
}
|
|
|
|
|
|
}</code></pre>
|
|
|
|
|
|
<p><span class="caption">示例 20-20: 在 worker 线程中接收并执行任务</span></p>
|
|
|
|
|
|
<p>这里，首先在 <code>receiver</code> 上调用了 <code>lock</code> 来获取互斥器，接着 <code>unwrap</code> 在出现任何错误时 panic。如果互斥器处于一种叫做 <strong>被污染</strong>（<em>poisoned</em>）的状态时获取锁可能会失败，这可能发生于其他线程在持有锁时 panic 了且没有释放锁。在这种情况下，调用 <code>unwrap</code> 使其 panic 是正确的行为。请随意将 <code>unwrap</code> 改为包含有意义错误信息的 <code>expect</code>。</p>
|
|
|
|
|
|
<p>如果锁定了互斥器，接着调用 <code>recv</code> 从信道中接收 <code>Job</code>。最后的 <code>unwrap</code> 也绕过了一些错误，这可能发生于持有信道发送端的线程停止的情况，类似于如果接收端关闭时 <code>send</code> 方法如何返回 <code>Err</code> 一样。</p>
|
|
|
|
|
|
<p>调用 <code>recv</code> 会阻塞当前线程，所以如果还没有任务，其会等待直到有可用的任务。<code>Mutex<T></code> 确保一次只有一个 <code>Worker</code> 线程尝试请求任务。</p>
|
|
|
|
|
|
<p>现在线程池处于可以运行的状态了！执行 <code>cargo run</code> 并发起一些请求：</p>
|
|
|
|
|
|
<pre><code class="language-console">$ cargo run
|
|
|
|
|
|
Compiling hello v0.1.0 (file:///projects/hello)
|
|
|
|
|
|
warning: field is never read: `workers`
|
|
|
|
|
|
--> src/lib.rs:7:5
|
|
|
|
|
|
|
|
|
|
|
|
|
7 | workers: Vec<Worker>,
|
|
|
|
|
|
| ^^^^^^^^^^^^^^^^^^^^
|
|
|
|
|
|
|
|
|
|
|
|
|
= note: `#[warn(dead_code)]` on by default
|
|
|
|
|
|
|
|
|
|
|
|
warning: field is never read: `id`
|
|
|
|
|
|
--> src/lib.rs:48:5
|
|
|
|
|
|
|
|
|
|
|
|
|
48 | id: usize,
|
|
|
|
|
|
| ^^^^^^^^^
|
|
|
|
|
|
|
|
|
|
|
|
warning: field is never read: `thread`
|
|
|
|
|
|
--> src/lib.rs:49:5
|
|
|
|
|
|
|
|
|
|
|
|
|
49 | thread: thread::JoinHandle<()>,
|
|
|
|
|
|
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
|
|
|
|
|
|
|
|
|
|
|
warning: `hello` (lib) generated 3 warnings
|
|
|
|
|
|
Finished dev [unoptimized + debuginfo] target(s) in 1.40s
|
|
|
|
|
|
Running `target/debug/hello`
|
|
|
|
|
|
Worker 0 got a job; executing.
|
|
|
|
|
|
Worker 2 got a job; executing.
|
|
|
|
|
|
Worker 1 got a job; executing.
|
|
|
|
|
|
Worker 3 got a job; executing.
|
|
|
|
|
|
Worker 0 got a job; executing.
|
|
|
|
|
|
Worker 2 got a job; executing.
|
|
|
|
|
|
Worker 1 got a job; executing.
|
|
|
|
|
|
Worker 3 got a job; executing.
|
|
|
|
|
|
Worker 0 got a job; executing.
|
|
|
|
|
|
Worker 2 got a job; executing.
|
|
|
|
|
|
</code></pre>
|
|
|
|
|
|
<p>成功了！现在我们有了一个可以异步执行连接的线程池！它绝不会创建超过四个线程，所以当 server 收到大量请求时系统也不会负担过重。如果请求 <em>/sleep</em>，server 也能够通过另外一个线程处理其他请求。</p>
|
|
|
|
|
|
<blockquote>
|
|
|
|
|
|
<p>注意如果同时在多个浏览器窗口打开 <em>/sleep</em>，它们可能会彼此间隔地加载 5 秒，因为一些浏览器出于缓存的原因会顺序执行相同请求的多个实例。这些限制并不是由于我们的 web server 造成的。</p>
|
|
|
|
|
|
</blockquote>
|
|
|
|
|
|
<p>在学习了第十八章的 <code>while let</code> 循环之后，你可能会好奇为何不能如此编写 worker 线程，如示例 20-21 所示：</p>
|
|
|
|
|
|
<p><span class="filename">文件名：src/lib.rs</span></p>
|
|
|
|
|
|
<pre><code class="language-rust ignore not_desired_behavior"><span class="boring">use std::{
|
|
|
|
|
|
</span><span class="boring"> sync::{mpsc, Arc, Mutex},
|
|
|
|
|
|
</span><span class="boring"> thread,
|
|
|
|
|
|
</span><span class="boring">};
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring">pub struct ThreadPool {
|
|
|
|
|
|
</span><span class="boring"> workers: Vec<Worker>,
|
|
|
|
|
|
</span><span class="boring"> sender: mpsc::Sender<Job>,
|
|
|
|
|
|
</span><span class="boring">}
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring">type Job = Box<dyn FnOnce() + Send + 'static>;
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring">impl ThreadPool {
|
|
|
|
|
|
</span><span class="boring"> /// Create a new ThreadPool.
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// The size is the number of threads in the pool.
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// # Panics
|
|
|
|
|
|
</span><span class="boring"> ///
|
|
|
|
|
|
</span><span class="boring"> /// The `new` function will panic if the size is zero.
|
|
|
|
|
|
</span><span class="boring"> pub fn new(size: usize) -> ThreadPool {
|
|
|
|
|
|
</span><span class="boring"> assert!(size > 0);
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let (sender, receiver) = mpsc::channel();
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let receiver = Arc::new(Mutex::new(receiver));
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> let mut workers = Vec::with_capacity(size);
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> for id in 0..size {
|
|
|
|
|
|
</span><span class="boring"> workers.push(Worker::new(id, Arc::clone(&receiver)));
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> ThreadPool { workers, sender }
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> pub fn execute<F>(&self, f: F)
|
|
|
|
|
|
</span><span class="boring"> where
|
|
|
|
|
|
</span><span class="boring"> F: FnOnce() + Send + 'static,
|
|
|
|
|
|
</span><span class="boring"> {
|
|
|
|
|
|
</span><span class="boring"> let job = Box::new(f);
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring"> self.sender.send(job).unwrap();
|
|
|
|
|
|
</span><span class="boring"> }
|
|
|
|
|
|
</span><span class="boring">}
|
|
|
|
|
|
</span><span class="boring">
|
|
|
|
|
|
</span><span class="boring">struct Worker {
|
|
|
|
|
|
</span><span class="boring"> id: usize,
|
|
|
|
|
|
</span><span class="boring"> thread: thread::JoinHandle<()>,
|
|
|
|
|
|
</span><span class="boring">}
|
|
|
|
|
|
</span>// --snip--
|
|
|
|
|
|
|
|
|
|
|
|
impl Worker {
|
|
|
|
|
|
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
|
|
|
|
|
|
let thread = thread::spawn(move || {
|
|
|
|
|
|
while let Ok(job) = receiver.lock().unwrap().recv() {
|
|
|
|
|
|
println!("Worker {id} got a job; executing.");
|
|
|
|
|
|
|
|
|
|
|
|
job();
|
|
|
|
|
|
}
|
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
|
|
Worker { id, thread }
|
|
|
|
|
|
}
|
|
|
|
|
|
}</code></pre>
|
|
|
|
|
|
<p><span class="caption">示例 20-21: 一个使用 <code>while let</code> 的 <code>Worker::new</code> 替代实现</span></p>
|
|
|
|
|
|
<p>这段代码可以编译和运行，但是并不会产生所期望的线程行为：一个慢请求仍然会导致其他请求等待执行。其原因有些微妙：<code>Mutex</code> 结构体没有公有 <code>unlock</code> 方法，因为锁的所有权依赖 <code>lock</code> 方法返回的 <code>LockResult<MutexGuard<T>></code> 中 <code>MutexGuard<T></code> 的生命周期。这允许借用检查器在编译时确保绝不会在没有持有锁的情况下访问由 <code>Mutex</code> 守护的资源，不过如果没有认真的思考 <code>MutexGuard<T></code> 的生命周期的话，也可能会导致比预期更久的持有锁。</p>
|
|
|
|
|
|
<p>示例 20-20 中的代码使用的 <code>let job = receiver.lock().unwrap().recv().unwrap();</code> 之所以可以工作是因为对于 <code>let</code> 来说，当 <code>let</code> 语句结束时任何表达式中等号右侧使用的临时值都会立即被丢弃。然而 <code>while let</code>（<code>if let</code> 和 <code>match</code>）直到相关的代码块结束都不会丢弃临时值。在示例 20-21 中，<code>job()</code> 调用期间锁一直持续，这也意味着其他的 worker 无法接受任务。</p>
|
|
|
|
|
|
|
|
|
|
|
|
</main>
|
|
|
|
|
|
|
|
|
|
|
|
<nav class="nav-wrapper" aria-label="Page navigation">
|
|
|
|
|
|
<!-- Mobile navigation buttons -->
|
|
|
|
|
|
<a rel="prev" href="ch20-01-single-threaded.html" class="mobile-nav-chapters previous" title="Previous chapter" aria-label="Previous chapter" aria-keyshortcuts="Left">
|
|
|
|
|
|
<i class="fa fa-angle-left"></i>
|
|
|
|
|
|
</a>
|
|
|
|
|
|
|
|
|
|
|
|
<a rel="next prefetch" href="ch20-03-graceful-shutdown-and-cleanup.html" class="mobile-nav-chapters next" title="Next chapter" aria-label="Next chapter" aria-keyshortcuts="Right">
|
|
|
|
|
|
<i class="fa fa-angle-right"></i>
|
|
|
|
|
|
</a>
|
|
|
|
|
|
|
|
|
|
|
|
<div style="clear: both"></div>
|
|
|
|
|
|
</nav>
|
|
|
|
|
|
</div>
|
|
|
|
|
|
</div>
|
|
|
|
|
|
|
|
|
|
|
|
<nav class="nav-wide-wrapper" aria-label="Page navigation">
|
|
|
|
|
|
<a rel="prev" href="ch20-01-single-threaded.html" class="nav-chapters previous" title="Previous chapter" aria-label="Previous chapter" aria-keyshortcuts="Left">
|
|
|
|
|
|
<i class="fa fa-angle-left"></i>
|
|
|
|
|
|
</a>
|
|
|
|
|
|
|
|
|
|
|
|
<a rel="next prefetch" href="ch20-03-graceful-shutdown-and-cleanup.html" class="nav-chapters next" title="Next chapter" aria-label="Next chapter" aria-keyshortcuts="Right">
|
|
|
|
|
|
<i class="fa fa-angle-right"></i>
|
|
|
|
|
|
</a>
|
|
|
|
|
|
</nav>
|
|
|
|
|
|
|
|
|
|
|
|
</div>
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
<script>
|
|
|
|
|
|
window.playground_copyable = true;
|
|
|
|
|
|
</script>
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
<script src="elasticlunr.min.js"></script>
|
|
|
|
|
|
<script src="mark.min.js"></script>
|
|
|
|
|
|
<script src="searcher.js"></script>
|
|
|
|
|
|
|
|
|
|
|
|
<script src="clipboard.min.js"></script>
|
|
|
|
|
|
<script src="highlight.js"></script>
|
|
|
|
|
|
<script src="book.js"></script>
|
|
|
|
|
|
|
|
|
|
|
|
<!-- Custom JS scripts -->
|
|
|
|
|
|
<script src="ferris.js"></script>
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
</div>
|
|
|
|
|
|
</body>
|
|
|
|
|
|
</html>
|
