go 模拟并发搜索案例

案例

当我们打开搜索栏查询东西时，搜索引擎具体做了哪些工作呢？

依据查询的数据对网页，图片，视频，地图，新闻等等进行搜索，然后聚合到一个页面来展示（当然了，还少不了广告！）

我们使用简单的代码来模拟这一过程

Search1.0

package main

import (
	"fmt"
    "math/rand"
    "time"
)

//定义三个搜索函数
var (
	Web   = simpleSearch("web")
	Image = simpleSearch("image")
	Video = simpleSearch("video")
)

//Result 简易搜索结果
type Result struct {
	kind  string //查询类型
	query string //查询的数据
}

//Search 定义搜索类型的函数
type Search func(query string) Result

//simpleSearch 依据不同类型返回不同的搜索结果
func simpleSearch(kind string) Search {
	return func(query string) Result {
		//模拟搜索需要的时间
		time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
		//返回搜索结果
		return Result{query: query, kind: kind}
	}
}

//Google 聚合了三个搜索引擎，并返回结果
func Google(query string) (results []Result) {
	results = append(results, Web(query))
	results = append(results, Image(query))
	results = append(results, Video(query))
	return
}

func main() {
	rand.Seed(time.Now().UnixNano())
	start := time.Now()
	results := Google("golang")
	elapsed := time.Since(start)
	fmt.Println(results)
	fmt.Println("花费:", elapsed)
}

执行结果：

[{web golang} {image golang} {video golang}]
花费: 181.984952ms

[{web golang} {image golang} {video golang}]
花费: 145.453691ms

[{web golang} {image golang} {video golang}]
花费: 153.791985ms

以上函数花费的时间取决于每一个搜索，如果Web()消耗了大量时间，那总消耗时间必然增加。因为在这一版本中所有函数都是串行的。

使用并发来修改下Google()函数，就叫Google2.0吧

Search2.0

func Google2(query string) (results []Result)  {
	c := make(chan Result)
	//开启三个goroutine并发的去查询
	go func() { c <- Web(query)}()
	go func() { c <- Image(query)}()
	go func() { c <- Video(query)}()
	for i := 0; i < 3; i++ {
		result := <-c
		results = append(results, result)
	}
	return
}

使用Google2()来模拟搜索结果：

[{image golang} {video golang} {web golang}]
花费: 86.626765ms

[{web golang} {image golang} {video golang}]
花费: 17.148741ms

[{video golang} {image golang} {web golang}]
花费: 67.732687ms

[{web golang} {image golang} {video golang}]
花费: 59.748459ms

Search2.1

这次的执行时间明显比之前的版本缩短了好多！看来效率提高了。我们在优化下Google2(),加入超时控制，跳过哪些搜索慢的服务。

func Google2_1(query string) (results []Result) {
	c := make(chan Result)
	go func() { c <- Web(query) }()
	go func() { c <- Image(query) }()
	go func() { c <- Video(query) }()
	timeout := time.After(50 * time.Millisecond)
	for i := 0; i < 3; i++ {
		select {
		case result := <-c:
			results = append(results, result)
		case <-timeout:
			fmt.Println("超时了")
			return
		}
	}
	return
}

多执行几次后的结果

超时了
[{image golang} {video golang}]
花费: 50.617606ms

超时了
[{web golang}]
花费: 50.848679ms

超时了
[{image golang} {video golang}]
花费: 50.847017ms

[{video golang} {web golang} {image golang}]
花费: 33.452545ms

超时了
[{web golang}]
花费: 52.410269ms

Search3.0

在Google2_1()版本中，我们保证了每次查询不超过50ms，超过我们就丢弃那些慢的服务。但是又有一个新问题：如何避免丢弃速度比较慢的服务器结果呢？

我们可以复制服务器，比如同时有3个Web，3个Video，3个Image服务器，将请求发送到多个服务器中，并使用第一个给我们响应的服务器数据。

func First(query string, replicas ...Search) Result {
	c := make(chan Result)
	searchReplica := func(i int) {
		c <- replicas[i](query)
	}
	for i := range replicas {
		go searchReplica(i)
	}
	return <-c
}

func Google3(query string) (results []Result) {
	c := make(chan Result)
	go func() { c <- First(query, simpleSearch("web1"), simpleSearch("web2"), simpleSearch("web3")) }()
	go func() { c <- First(query, simpleSearch("image"), simpleSearch("image2"), simpleSearch("image3")) }()
	go func() { c <- First(query, simpleSearch("video1"), simpleSearch("video2"), simpleSearch("video3")) }()
	timeout := time.After(60 * time.Millisecond)
	for i := 0; i < 3; i++ {
		select {
		case result := <-c:
			results = append(results, result)
		case <-timeout:
			fmt.Println("超时了")
			return
		}
	}
	return
}

多次运行结果如下

[{image2 golang} {video1 golang} {web3 golang}]
花费: 18.424318ms

[{image2 golang} {video3 golang} {web1 golang}]
花费: 16.540826ms

[{image2 golang} {web1 golang} {video2 golang}]
花费: 34.311563ms

[{video2 golang} {image golang} {web1 golang}]
花费: 50.067555ms

[{video1 golang} {web2 golang} {image golang}]
花费: 43.243104ms

[{video2 golang} {image3 golang} {web2 golang}]
花费: 36.292079ms

这次我们能够将数据查询压缩至更少的响应时间，同时我们使用了多台服务器，并取其中最快响应的数据。

总结

通过以上几个简单的demo，我们就使用go语言中的并发特性将

响应慢
顺序执行
单一服务（故障敏感,只要一个服务出问题，则回影响全局）

改造成

快速响应
并发执行
可复制
健壮的

的服务！

本文代码

package main

import (
	"fmt"
    "math/rand"
    "time"
)

//定义三个搜索函数
var (
	Web   = simpleSearch("web")
	Image = simpleSearch("image")
	Video = simpleSearch("video")
)

//Result 简易搜索结果
type Result struct {
	kind  string //查询类型
	query string //查询的数据
}

//Search 定义搜索类型的函数
type Search func(query string) Result

//simpleSearch 依据不同类型返回不同的搜索结果
func simpleSearch(kind string) Search {
	return func(query string) Result {
		//模拟搜索需要的时间
		time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
		//返回搜索结果
		return Result{query: query, kind: kind}
	}
}

//Google 聚合了三个搜索引擎，并返回结果
func Google(query string) (results []Result) {
	results = append(results, Web(query))
	results = append(results, Image(query))
	results = append(results, Video(query))
	return
}

func Google2(query string) (results []Result) {
	c := make(chan Result)
	go func() { c <- Web(query) }()
	go func() { c <- Image(query) }()
	go func() { c <- Video(query) }()
	for i := 0; i < 3; i++ {
		result := <-c
		results = append(results, result)
	}
	return
}

func Google2_1(query string) (results []Result) {
	c := make(chan Result)
	go func() { c <- Web(query) }()
	go func() { c <- Image(query) }()
	go func() { c <- Video(query) }()
	timeout := time.After(50 * time.Millisecond)
	for i := 0; i < 3; i++ {
		select {
		case result := <-c:
			results = append(results, result)
		case <-timeout:
			fmt.Println("超时了")
			return
		}
	}
	return
}

func First(query string, replicas ...Search) Result {
	c := make(chan Result)
	searchReplica := func(i int) {
		c <- replicas[i](query)
	}
	for i := range replicas {
		go searchReplica(i)
	}
	return <-c
}

func Google3(query string) (results []Result) {
	c := make(chan Result)
	//https://blog.68hub.com/posts/go-concurrency-search-demo
	go func() { c <- First(query, simpleSearch("web1"), simpleSearch("web2"), simpleSearch("web3")) }()
	go func() { c <- First(query, simpleSearch("image"), simpleSearch("image2"), simpleSearch("image3")) }()
	go func() { c <- First(query, simpleSearch("video1"), simpleSearch("video2"), simpleSearch("video3")) }()
	timeout := time.After(60 * time.Millisecond)
	for i := 0; i < 3; i++ {
		select {
		case result := <-c:
			results = append(results, result)
		case <-timeout:
			fmt.Println("超时了")
			return
		}
	}
	return
}

func main() {
	rand.Seed(time.Now().UnixNano())
	start := time.Now()
	//results := Google("golang")
	//results := Google2("golang")
	//results := Google2_1("golang")
	results := Google3("golang")
	elapsed := time.Since(start)
	fmt.Println(results)
	fmt.Println("花费:", elapsed)
}

参考

https://talks.golang.org/2012/concurrency.slide