建站
咨詢
售后
建站咨詢
新聞中心
引言
Go并發(fā)原語(yǔ)使得構(gòu)建流式數(shù)據(jù)管道，高效利用I/O和多核變得簡(jiǎn)單。這篇文章介紹了幾個(gè)管道例子，重點(diǎn)指出在操作失敗時(shí)的細(xì)微差別，并介紹了優(yōu)雅處理失敗的技術(shù)。
什么是管道？
Go沒(méi)有正式的管道定義。管道只是眾多并發(fā)程序的一類。一般的，一個(gè)管道就是一些列的由channel連接起來(lái)的階段。每個(gè)階段都有執(zhí)行相同邏輯的goroutine。在每個(gè)階段中，goroutine
- 從channel讀取上游數(shù)據(jù)
- 在數(shù)據(jù)上執(zhí)行一些操作，通常會(huì)產(chǎn)生新的數(shù)據(jù)
- 通過(guò)channel將數(shù)據(jù)發(fā)往下游
每個(gè)階段都可以有任意個(gè)輸入channel和輸出channel，除了第一個(gè)和最有一個(gè)channel（只有輸入channel或只有輸出channel）。第一個(gè)步驟通常叫數(shù)據(jù)源或者生產(chǎn)者，最后一個(gè)叫做存儲(chǔ)池或者消費(fèi)者。
我們先從一個(gè)簡(jiǎn)單的管道例子來(lái)解釋這些概念和技術(shù)，稍后我們會(huì)介紹一個(gè)更為復(fù)雜的例子。
數(shù)字的平方
假設(shè)管道有三個(gè)階段。
第一步，gen函數(shù),是一個(gè)將數(shù)字列表轉(zhuǎn)換到一個(gè)channel中的函數(shù)。Gen函數(shù)啟動(dòng)了一個(gè)goroutine，將數(shù)字發(fā)送到channel，并在所有數(shù)字都發(fā)送完后關(guān)閉channel。
- func gen(nums ...int) <-chan int {
- out := make(chan int)
- go func() {
- for _, n := range nums {
- out <- n
- }
- close(out)
- }()
- return out
- }
第二個(gè)階段，sq，從上面的channel接收數(shù)字，并返回一個(gè)包含所有收到數(shù)字的平方的channel。在上游channel關(guān)閉后，這個(gè)階段已經(jīng)往下游發(fā)送完所有的結(jié)果，然后關(guān)閉輸出channel：
- func sq(in <-chan int) <-chan int {
- out := make(chan int)
- go func() {
- for n := range in {
- out <- n * n
- }
- close(out)
- }()
- return out
- }
main函數(shù)建立這個(gè)管道，并執(zhí)行第一個(gè)階段，從第二個(gè)階段接收結(jié)果并逐個(gè)打印，直到channel被關(guān)閉。
- func main() {
- // Set up the pipeline.
- c := gen(2, 3)
- out := sq(c)
- // Consume the output.
- fmt.Println(<-out) // 4
- fmt.Println(<-out) // 9
- }
因?yàn)閟q對(duì)輸入channel和輸出channel擁有相同的類型，我們可以任意次的組合他們。我們也可以像其他階段一樣，將main函數(shù)重寫成一個(gè)循環(huán)遍歷。
- func main() {
- // Set up the pipeline and consume the output.
- for n := range sq(sq(gen(2, 3))) {
- fmt.Println(n) // 16 then 81
- }
- }
扇出扇入（Fan-out, fan-in）
多個(gè)函數(shù)可以從同一個(gè)channel讀取數(shù)據(jù)，直到這個(gè)channel關(guān)閉，這叫扇出。這是一種多個(gè)工作實(shí)例分布式地協(xié)作以并行利用CPU和I/O的方式。
一個(gè)函數(shù)可以從多個(gè)輸入讀取并處理數(shù)據(jù)，直到所有的輸入channel都被關(guān)閉。這個(gè)函數(shù)會(huì)將所有輸入channel導(dǎo)入一個(gè)單一的channel。這個(gè)單一的channel在所有輸入channel都關(guān)閉后才會(huì)關(guān)閉。這叫做扇入。
我們可以設(shè)置我們的管道執(zhí)行兩個(gè)sq實(shí)例，每一個(gè)實(shí)例都從相同的輸入channel讀取數(shù)據(jù)。我們引入了一個(gè)新的函數(shù)，merge，來(lái)扇入結(jié)果:
- func main() {
- in := gen(2, 3)
- // Distribute the sq work across two goroutines that both read from in.
- c1 := sq(in)
- c2 := sq(in)
- // Consume the merged output from c1 and c2.
- for n := range merge(c1, c2) {
- fmt.Println(n) // 4 then 9, or 9 then 4
- }
- }
merge函數(shù)為每一個(gè)輸入channel啟動(dòng)一個(gè)goroutine，goroutine將數(shù)據(jù)拷貝到同一個(gè)輸出channel。這樣就將多個(gè)channel轉(zhuǎn)換成一個(gè)channel。一旦所有的output goroutine啟動(dòng)起來(lái)，merge就啟動(dòng)另一個(gè)goroutine，在所有輸入拷貝完畢后關(guān)閉輸出channel。
向一個(gè)關(guān)閉了的channel發(fā)送數(shù)據(jù)會(huì)觸發(fā)異常，所以在調(diào)用close之前確認(rèn)所有的發(fā)送動(dòng)作都執(zhí)行完畢很重要。sync.WaitGroup類型為這種同步提供了一種簡(jiǎn)便的方法:
- func merge(cs ...<-chan int) <-chan int {
- var wg sync.WaitGroup
- out := make(chan int)
- // Start an output goroutine for each input channel in cs. output
- // copies values from c to out until c is closed, then calls wg.Done.
- output := func(c <-chan int) {
- for n := range c {
- out <- n
- }
- wg.Done()
- }
- wg.Add(len(cs))
- for _, c := range cs {
- go output(c)
- }
- // Start a goroutine to close out once all the output goroutines are
- // done. This must start after the wg.Add call.
- go func() {
- wg.Wait()
- close(out)
- }()
- return out
- }
停止的藝術(shù)
我們所有的管道函數(shù)都遵循一種模式：
- 發(fā)送者在發(fā)送完畢時(shí)關(guān)閉其輸出channel。
- 接收者持續(xù)從輸入管道接收數(shù)據(jù)直到輸入管道關(guān)閉。
這種模式使得每一個(gè)接收函數(shù)都能寫成一個(gè)range循環(huán)，保證所有的goroutine在數(shù)據(jù)成功發(fā)送到下游后就關(guān)閉。
但是在真實(shí)的案例中，并不是所有的輸入數(shù)據(jù)都需要被接收處理。有些時(shí)候是故意這么設(shè)計(jì)的：接收者可能只需要數(shù)據(jù)的子集就夠了；或者更一般的，因?yàn)檩斎霐?shù)據(jù)有錯(cuò)誤而導(dǎo)致接收函數(shù)提早退出。上面任何一種情況下，接收者都不應(yīng)該繼續(xù)等待后續(xù)的數(shù)據(jù)到來(lái)，并且我們希望上游函數(shù)停止生成后續(xù)步驟已經(jīng)不需要的數(shù)據(jù)。
在我們的管道例子中，如果一個(gè)階段無(wú)法消費(fèi)所有的輸入數(shù)據(jù)，那些發(fā)送這些數(shù)據(jù)的goroutine就會(huì)一直阻塞下去：
- // Consume the first value from output.
- out := merge(c1, c2)
- fmt.Println(<-out) // 4 or 9
- return
- // Since we didn't receive the second value from out,
- // one of the output goroutines is hung attempting to send it.
- }
這是一種資源泄漏：goroutine會(huì)占用內(nèi)存和運(yùn)行時(shí)資源。goroutine棧持有的堆引用會(huì)阻止GC回收資源。而且goroutine不能被垃圾回收，必須主動(dòng)退出。
我們必須重新設(shè)計(jì)管道中的上游函數(shù)，在下游函數(shù)無(wú)法接收所有輸入數(shù)據(jù)時(shí)退出。一種方法就是讓輸出channel擁有一定的緩存。緩存可以存儲(chǔ)一定數(shù)量的數(shù)據(jù)。如果緩存空間足夠，發(fā)送操作就會(huì)馬上返回:
- c := make(chan int, 2) // buffer size 2
- c <- 1 // succeeds immediately
- c <- 2 // succeeds immediately
- c <- 3 // blocks until another goroutine does <-c and receives 1
如果在channel創(chuàng)建時(shí)就知道需要發(fā)送數(shù)據(jù)的數(shù)量，帶緩存的channel會(huì)簡(jiǎn)化代碼。例如，我們可以重寫gen函數(shù)，拷貝一系列的整數(shù)到一個(gè)帶緩存的channel而不是創(chuàng)建一個(gè)新的goroutine：
- func gen(nums ...int) <-chan int {
- out := make(chan int, len(nums))
- for _, n := range nums {
- out <- n
- }
- close(out)
- return out
- }
反過(guò)來(lái)我們看管道中被阻塞的goroutine，我們可以考慮為merge函數(shù)返回的輸出channel增加一個(gè)緩存：
- func merge(cs ...<-chan int) <-chan int {
- var wg sync.WaitGroup
- out := make(chan int, 1) // enough space for the unread inputs
- // ... the rest is unchanged ...
雖然這樣可以避免了程序中g(shù)oroutine的阻塞，但這是很爛的代碼。選擇緩存大小為1取決于知道m(xù)erge函數(shù)接收數(shù)字的數(shù)量和下游函數(shù)消費(fèi)數(shù)字的數(shù)量。這是很不穩(wěn)定的：如果我們向gen多發(fā)送了一個(gè)數(shù)據(jù)，或者下游函數(shù)少消費(fèi)了數(shù)據(jù)，我們就又一次阻塞了goroutine。
然而，我們需要提供一種方式，下游函數(shù)可以通知上游發(fā)送者下游要停止接收數(shù)據(jù)。
#p#
顯式取消
當(dāng)main函數(shù)決定在沒(méi)有從out接收所有的數(shù)據(jù)而要退出時(shí)，它需要通知上游的goroutine取消即將發(fā)送的數(shù)據(jù)?？梢酝ㄟ^(guò)向一個(gè)叫做done的channel發(fā)送數(shù)據(jù)來(lái)實(shí)現(xiàn)。因?yàn)橛袃蓚€(gè)潛在阻塞的goroutine，main函數(shù)會(huì)發(fā)送兩個(gè)數(shù)據(jù)：
- func main() {
- in := gen(2, 3)
- // Distribute the sq work across two goroutines that both read from in.
- c1 := sq(in)
- c2 := sq(in)
- // Consume the first value from output.
- done := make(chan struct{}, 2)
- out := merge(done, c1, c2)
- fmt.Println(<-out) // 4 or 9
- // Tell the remaining senders we're leaving.
- done <- struct{}{}
- done <- struct{}{}
- }
對(duì)發(fā)送goroutine而言，需要將發(fā)送操作替換為一個(gè)select語(yǔ)句，要么out發(fā)生發(fā)送操作，要么從done接收數(shù)據(jù)。done的數(shù)據(jù)類型是空的struct，因?yàn)槠渲禑o(wú)關(guān)緊要：僅僅表示out需要取消發(fā)送操作。output 繼續(xù)在輸入channel循環(huán)執(zhí)行，因此上游函數(shù)是不會(huì)阻塞的。（接下來(lái)我們會(huì)討論如何讓循環(huán)提早退出）
- func merge(done <-chan struct{}, cs ...<-chan int) <-chan int {
- var wg sync.WaitGroup
- out := make(chan int)
- // Start an output goroutine for each input channel in cs. output
- // copies values from c to out until c is closed or it receives a value
- // from done, then output calls wg.Done.
- output := func(c <-chan int) {
- for n := range c {
- select {
- case out <- n:
- case <-done:
- }
- }
- wg.Done()
- }
- // ... the rest is unchanged ...
這種方法有一個(gè)問(wèn)題：每一個(gè)下游函數(shù)需要知道潛在可能阻塞的上游發(fā)送者的數(shù)量，以發(fā)送響應(yīng)的信號(hào)讓其提早退出。跟蹤這些數(shù)量是無(wú)趣的而且很容易出錯(cuò)。
我們需要一種能夠讓未知或無(wú)界數(shù)量的goroutine都能夠停止向下游發(fā)送數(shù)據(jù)的方法。在Go中，我們可以通過(guò)關(guān)閉一個(gè)channel實(shí)現(xiàn)。因?yàn)閺囊粋€(gè)關(guān)閉了的channel執(zhí)行接收操作總能馬上成功，并返回相應(yīng)數(shù)據(jù)類型的零值。
這意味著main函數(shù)僅通過(guò)關(guān)閉done就能實(shí)現(xiàn)將所有的發(fā)送者解除阻塞。關(guān)閉操作是一個(gè)高效的對(duì)發(fā)送者的廣播信號(hào)。我們擴(kuò)展管道中所有的函數(shù)接受done作為一個(gè)參數(shù)，并通過(guò)defer來(lái)實(shí)現(xiàn)相應(yīng)channel的關(guān)閉操作。因此，無(wú)論main函數(shù)在哪一行退出都會(huì)通知上游退出。
- func main() {
- // Set up a done channel that's shared by the whole pipeline,
- // and close that channel when this pipeline exits, as a signal
- // for all the goroutines we started to exit.
- done := make(chan struct{})
- defer close(done)
- in := gen(done, 2, 3)
- // Distribute the sq work across two goroutines that both read from in.
- c1 := sq(done, in)
- c2 := sq(done, in)
- // Consume the first value from output.
- out := merge(done, c1, c2)
- fmt.Println(<-out) // 4 or 9
- // done will be closed by the deferred call.
- }
現(xiàn)在每一個(gè)管道函數(shù)在done被關(guān)閉后就可以馬上返回了。merge函數(shù)中的output可以在接收管道的數(shù)據(jù)消費(fèi)完之前返回，因?yàn)閛utput函數(shù)知道上游發(fā)送者sq會(huì)在done關(guān)閉后停止產(chǎn)生數(shù)據(jù)。同時(shí)，output通過(guò)defer語(yǔ)句保證wq.Done會(huì)在所有退出路徑上調(diào)用。
- func merge(done <-chan struct{}, cs ...<-chan int) <-chan int {
- var wg sync.WaitGroup
- out := make(chan int)
- // Start an output goroutine for each input channel in cs. output
- // copies values from c to out until c or done is closed, then calls
- // wg.Done.
- output := func(c <-chan int) {
- defer wg.Done()
- for n := range c {
- select {
- case out <- n:
- case <-done:
- return
- }
- }
- }
- // ... the rest is unchanged ...
類似的，sq也可以在done關(guān)閉后馬上返回。sq通過(guò)defer語(yǔ)句使得任何退出路徑都能關(guān)閉其輸出channel out。
- func sq(done <-chan struct{}, in <-chan int) <-chan int {
- out := make(chan int)
- go func() {
- defer close(out)
- for n := range in {
- select {
- case out <- n * n:
- case <-done:
- return
- }
- }
- }()
- return out
- }
管道構(gòu)建的指導(dǎo)思想如下：
- 每一個(gè)階段在所有發(fā)送操作完成后關(guān)閉輸出channel。
- 每一個(gè)階段持續(xù)從輸入channel接收數(shù)據(jù)直到輸入channel被關(guān)閉或者生產(chǎn)者被解除阻塞（譯者：生產(chǎn)者退出）。
管道解除生產(chǎn)者阻塞有兩種方法：要么保證有足夠的緩存空間存儲(chǔ)將要被生產(chǎn)的數(shù)據(jù)，要么顯式的通知生產(chǎn)者消費(fèi)者要取消接收數(shù)據(jù)。
樹形摘要
讓我們來(lái)看一個(gè)更為實(shí)際的管道。
MD5是一個(gè)信息摘要算法，對(duì)于文件校驗(yàn)非常有用。命令行工具md5sum很有用，可以打印一系列文件的摘要值。
- % md5sum *.go
- d47c2bbc28298ca9befdfbc5d3aa4e65 bounded.go
- ee869afd31f83cbb2d10ee81b2b831dc parallel.go
- b88175e65fdcbc01ac08aaf1fd9b5e96 serial.go
我們的例子程序和md5sum類似，但是接受一個(gè)單一的文件夾作為參數(shù)，打印該文件夾下每一個(gè)普通文件的摘要值，并按路徑名稱排序。
- % go run serial.go .
- d47c2bbc28298ca9befdfbc5d3aa4e65 bounded.go
- ee869afd31f83cbb2d10ee81b2b831dc parallel.go
- b88175e65fdcbc01ac08aaf1fd9b5e96 serial.go
我們程序的main函數(shù)調(diào)用一個(gè)工具函數(shù)MD5ALL，該函數(shù)返回一個(gè)從路徑名稱到摘要值的哈希表，然后排序并輸出結(jié)果：
- func main() {
- // Calculate the MD5 sum of all files under the specified directory,
- // then print the results sorted by path name.
- m, err := MD5All(os.Args[1])
- if err != nil {
- fmt.Println(err)
- return
- }
- var paths []string
- for path := range m {
- paths = append(paths, path)
- }
- sort.Strings(paths)
- for _, path := range paths {
- fmt.Printf("%x %s\n", m[path], path)
- }
- }
MD5ALL是我們討論的核心。在 serial.go中，沒(méi)有采用任何并發(fā)，僅僅遍歷文件夾，讀取文件并求出摘要值。
- // MD5All reads all the files in the file tree rooted at root and returns a map
- // from file path to the MD5 sum of the file's contents. If the directory walk
- // fails or any read operation fails, MD5All returns an error.
- func MD5All(root string) (map[string][md5.Size]byte, error) {
- m := make(map[string][md5.Size]byte)
- err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
- if err != nil {
- return err
- }
- if info.IsDir() {
- return nil
- }
- data, err := ioutil.ReadFile(path)
- if err != nil {
- return err
- }
- m[path] = md5.Sum(data)
- return nil
- })
- if err != nil {
- return nil, err
- }
- return m, nil
- }
#p#
并行摘要求值
在parallel.go中，我們將MD5ALL分成兩階段的管道。第一個(gè)階段，sumFiles，遍歷文件夾，每個(gè)文件一個(gè)goroutine進(jìn)行求摘要值，然后將結(jié)果發(fā)送一個(gè)數(shù)據(jù)類型為result的channel中：
- type result struct {
- path string
- sum [md5.Size]byte
- err error
- }
sumFiles 返回兩個(gè)channel：一個(gè)用于生成結(jié)果，一個(gè)用于filepath.Walk返回錯(cuò)誤。Walk函數(shù)為每一個(gè)普通文件啟動(dòng)一個(gè)goroutine，然后檢查done，如果done被關(guān)閉，walk馬上就會(huì)退出。
- func sumFiles(done <-chan struct{}, root string) (<-chan result, <-chan error) {
- // For each regular file, start a goroutine that sums the file and sends
- // the result on c. Send the result of the walk on errc.
- c := make(chan result)
- errc := make(chan error, 1)
- go func() {
- var wg sync.WaitGroup
- err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
- if err != nil {
- return err
- }
- if info.IsDir() {
- return nil
- }
- wg.Add(1)
- go func() {
- data, err := ioutil.ReadFile(path)
- select {
- case c <- result{path, md5.Sum(data), err}:
- case <-done:
- }
- wg.Done()
- }()
- // Abort the walk if done is closed.
- select {
- case <-done:
- return errors.New("walk canceled")
- default:
- return nil
- }
- })
- // Walk has returned, so all calls to wg.Add are done. Start a
- // goroutine to close c once all the sends are done.
- go func() {
- wg.Wait()
- close(c)
- }()
- // No select needed here, since errc is buffered.
- errc <- err
- }()
- return c, errc
- }
MD5All 從c中接收摘要值。MD5All 在遇到錯(cuò)誤時(shí)提前退出，通過(guò)defer關(guān)閉done。
- func MD5All(root string) (map[string][md5.Size]byte, error) {
- // MD5All closes the done channel when it returns; it may do so before
- // receiving all the values from c and errc.
- done := make(chan struct{})
- defer close(done)
- c, errc := sumFiles(done, root)
- m := make(map[string][md5.Size]byte)
- for r := range c {
- if r.err != nil {
- return nil, r.err
- }
- m[r.path] = r.sum
- }
- if err := <-errc; err != nil {
- return nil, err
- }
- return m, nil
- }
有界并行
parallel.go中實(shí)現(xiàn)的MD5ALL，對(duì)每一個(gè)文件啟動(dòng)了一個(gè)goroutine。在一個(gè)包含大量大文件的文件夾中，這會(huì)導(dǎo)致超過(guò)機(jī)器可用內(nèi)存的內(nèi)存分配。（譯者注：即發(fā)生OOM）
我們可以通過(guò)限制讀取文件的并發(fā)度來(lái)避免這種情況發(fā)生。在bounded.go中，我們通過(guò)創(chuàng)建一定數(shù)量的goroutine讀取文件。現(xiàn)在我們的管道現(xiàn)在有三個(gè)階段：遍歷文件夾，讀取文件并計(jì)算摘要值，收集摘要值。
第一個(gè)階段，walkFiles，輸出文件夾中普通文件的文件路徑：
- func walkFiles(done <-chan struct{}, root string) (<-chan string, <-chan error) {
- paths := make(chan string)
- errc := make(chan error, 1)
- go func() {
- // Close the paths channel after Walk returns.
- defer close(paths)
- // No select needed for this send, since errc is buffered.
- errc <- filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
- if err != nil {
- return err
- }
- if info.IsDir() {
- return nil
- }
- select {
- case paths <- path:
- case <-done:
- return errors.New("walk canceled")
- }
- return nil
- })
- }()
- return paths, errc
- }
中間的階段啟動(dòng)一定數(shù)量的digester goroutine，從paths接收文件名稱，并向c發(fā)送result結(jié)構(gòu):
- func digester(done <-chan struct{}, paths <-chan string, c chan<- result) {
- for path := range paths {
- data, err := ioutil.ReadFile(path)
- select {
- case c <- result{path, md5.Sum(data), err}:
- case <-done:
- return
- }
- }
- }
和前一個(gè)例子不同，digester并不關(guān)閉其輸出channel，因?yàn)檩敵鯿hannel是共享的，多個(gè)goroutine會(huì)向同一個(gè)channel發(fā)送數(shù)據(jù)。MD5All 會(huì)在所有的digesters 結(jié)束后關(guān)閉響應(yīng)的channel。
- // Start a fixed number of goroutines to read and digest files.
- c := make(chan result)
- var wg sync.WaitGroup
- const numDigesters = 20
- wg.Add(numDigesters)
- for i := 0; i < numDigesters; i++ {
- go func() {
- digester(done, paths, c)
- wg.Done()
- }()
- }
- go func() {
- wg.Wait()
- close(c)
- }()
我們也可以讓每一個(gè)digester創(chuàng)建并返回自己的輸出channel，但如果這樣的話，我們需要額外的goroutine來(lái)扇入這些結(jié)果。
最后一個(gè)階段從c中接收所有的result數(shù)據(jù)，并從errc中檢查錯(cuò)誤。這種檢查不能在之前的階段做，因?yàn)樵谶@之前，walkFiles 可能被阻塞不能往下游發(fā)送數(shù)據(jù)：
- m := make(map[string][md5.Size]byte)
- for r := range c {
- if r.err != nil {
- return nil, r.err
- }
- m[r.path] = r.sum
- }
- // Check whether the Walk failed.
- if err := <-errc; err != nil {
- return nil, err
- }
- return m, nil
- }
結(jié)論
這篇文章介紹了如果用Go構(gòu)建流式數(shù)據(jù)管道的技術(shù)。在這樣的管道中處理錯(cuò)誤有點(diǎn)取巧，因?yàn)楣艿乐忻恳粋€(gè)階段可能被阻塞不能往下游發(fā)送數(shù)據(jù)，下游階段可能已經(jīng)不關(guān)心輸入數(shù)據(jù)。我們展示了關(guān)閉channel如何向所有管道啟動(dòng)的goroutine廣播一個(gè)done信號(hào)，并且定義了正確構(gòu)建管道的指導(dǎo)思想。
深入閱讀：
? Go并發(fā)模式（視頻）展示了Go并發(fā)原語(yǔ)的基本概念和幾個(gè)實(shí)現(xiàn)的方法
? 高級(jí)Go并發(fā)模式（視頻）包含幾個(gè)更為復(fù)雜的Go并發(fā)原語(yǔ)的使用，尤其是select
? Douglas McIlroy的Squinting at Power Series論文展示了類似Go的并發(fā)模式如何為復(fù)雜的計(jì)算提供優(yōu)雅的支持。
文章標(biāo)題：Go并發(fā)模式：管道和顯式取消
文章URL：http://m.5511xx.com/article/cojihge.html
