Go Concurrency

 1: // 警告：下面代码存在竞争条件！
 2: package main
 3: 
 4: import "fmt"
 5: 
 6: func main() {
 7: 	var data int
 8: 
 9: 	go func() {
10: 		data++
11: 	}()
12: 
13: 	if data == 0 {
14: 		fmt.Printf("the value is %v.\n", data)
15: 	}
16: }

package main

import (
	"net/http"
	"fmt"
	"sync"
)

func main() {
	var wg sync.WaitGroup
	var urls = []string{
		"http://www.golang.org/",
		"http://www.google.com/",
		"http://www.somestupidname.com/",
		"http://nosuch/",
	}
	for _, url := range urls {
		// Increment the WaitGroup counter.
		wg.Add(1)
		// Launch a goroutine to fetch the URL.
		go func(url string) {
			// Decrement the counter when the goroutine completes.
			defer wg.Done()
			// Fetch the URL.
			response, err := http.Get(url)
			if err != nil {
				fmt.Printf("Error=%s\n", err)
			} else {
				fmt.Printf("StatusCode=%d, url=%s\n", response.StatusCode, url)
			}
		}(url)
	}
	// Wait for all HTTP fetches to complete.
	wg.Wait()
}

package main

import (
	"fmt"
	"sync"
)

func main() {
	hello := func(wg *sync.WaitGroup, id int) {
		defer wg.Done()
		fmt.Printf("Hello from %v!\n", id)
	}

	const numGreeters = 5
	var wg sync.WaitGroup
	wg.Add(numGreeters)
	for i := 0; i < numGreeters; i++ {
		go hello(&wg, i+1)
	}

	wg.Wait()
}

 1: package main
 2: 
 3: import (
 4: 	"fmt"
 5: 	"math/rand"
 6: 	"sync"
 7: 	"time"
 8: )
 9: 
10: func main() {
11: 
12: 	var mutex sync.Mutex   // 写为 var mutex = new(sync.Mutex) 也行
13: 	var items []int
14: 
15: 	go func() {
16: 		// 消费者代码（删除队列items中的元素）
17: 		for {
18: 			mutex.Lock()
19: 			for len(items) > 0 { // 消耗队列items中所有元素
20: 				lastIndex := len(items) - 1
21: 				lastItem := items[lastIndex]
22: 				items = items[:lastIndex] // 删除最后一个元素
23: 				fmt.Printf("consume %d\n", lastItem)
24: 			}
25: 			mutex.Unlock()
26: 
27: 			// 必需Sleep，否则当队列items为空时CPU占用会很高
28: 			time.Sleep(10 * time.Millisecond)  // 很难决定Sleep多久合适！
29: 		}
30: 	}()
31: 
32: 	// 生产者代码（往队列items中增加元素）
33: 	for {
34: 		newItem := rand.Intn(10000)
35: 
36: 		mutex.Lock()
37: 		items = append(items, newItem)
38: 		fmt.Printf("produce %d\n", newItem)
39: 		mutex.Unlock()
40: 
41: 		// Other works
42: 		time.Sleep(time.Duration(rand.Intn(10)) * time.Second)
43: 	}
44: }

 1: package main
 2: 
 3: import (
 4: 	"fmt"
 5: 	"sync"
 6: 	"runtime"
 7: )
 8: 
 9: func main() {
10: 	myPool := &sync.Pool{
11: 		New: func() interface{} {
12: 			fmt.Println("Creating new instance.")
13: 			return struct{}{}
14: 		},
15: 	}
16: 
17: 	fmt.Println("line 17")
18: 	myPool.Get()              // myPool中没对象，调用Get()会触发New()
19: 	fmt.Println("line 19")
20: 	myPool.Get()              // myPool中没对象，调用Get()会触发New()
21: 	fmt.Println("line 21")
22: 	myPool.Get()              // myPool中没对象，调用Get()会触发New()
23: 	fmt.Println("line 23")
24: 	myPool.Put(myPool.New())  // 加入一个对象到myPool中
25: 	fmt.Println("line 25")
26: 	myPool.Put(myPool.New())  // 加入一个对象到myPool中
27: 	fmt.Println("line 27")
28: 	myPool.Get()              // myPool中有两个对象，调用Get()会取出其中一个对象，Pool中还剩下一个对象
29: 	fmt.Println("line 29")
30: 	runtime.GC()              // 执行GC时会清除myPool中所有对象
31: 	myPool.Get()              // 此时，myPool中没对象，调用Get()会触发New()
32: 	fmt.Println("line 32")
33: }

package main

import (
	"log"
	"strconv"
	"sync"
)

func main() {
	var myMap = make(map[int]string)

	// 并发写入 map
	var wg sync.WaitGroup
	for i := 0; i <= 99; i++ {
		wg.Add(1)
		go func(key int) {
			defer wg.Done()
			value := "value" + strconv.Itoa(key)
			myMap[key] = value    // 尽管每个 goroutine 访问 map 的 key 不同，但这在 Go 中存在竞争条件！
		}(i)
	}
	wg.Wait()

	// 依次输出每个 key
	for i := 0; i <= 99; i++ {
		result, ok := myMap[i]
		if ok {
			log.Printf("%v=%v", i, result)
		} else {
			log.Printf("value not found for key: %v", i)
		}
	}
}

package main

import (
	"log"
	"strconv"
	"sync"
)

func main() {
	var myMap sync.Map

	// 并发写入 map
	var wg sync.WaitGroup
	for i := 0; i <= 99; i++ {
		wg.Add(1)
		go func(key int) {
			defer wg.Done()
			value := "value" + strconv.Itoa(key)
			myMap.Store(key, value)   // 并发安全，无竞争条件
		}(i)
	}
	wg.Wait()

	// 依次输出每个 key
	for i := 0; i <= 99; i++ {
		result, ok := myMap.Load(i)
		if ok {
			log.Printf("%v=%s", i, result.(string))
		} else {
			log.Printf("value not found for key: %v", i)
		}
	}

	// 使用 Range 也可以遍历 Map 中的每个 key-value 对
	myMap.Range(func(k, v interface{}) bool {
		log.Printf("%v=%s", k, v.(string))
		return true
	})
}

 1: package main
 2: 
 3: import (
 4: 	"fmt"
 5: 	"net/http"
 6: )
 7: 
 8: func main() {
 9: 	checkStatus := func(urls ...string) <-chan *http.Response {
10: 		responses := make(chan *http.Response)
11: 		go func() {
12: 			defer close(responses)
13: 			for _, url := range urls {
14: 				resp, err := http.Get(url)
15: 				if err != nil {             // 在这里处理错误不合适，后面介绍更好的方式
16: 					fmt.Println(err)
17: 					continue
18: 				}
19: 				responses <- resp
20: 			}
21: 		}()
22: 		return responses
23: 	}
24: 
25: 	urls := []string{"https://www.bing.com", "https://badhost"}
26: 	for response := range checkStatus(urls...) {
27: 		fmt.Printf("Response: %v\n", response.Status)
28: 	}
29: }

package main

import (
	"fmt"
	"net/http"
)

type Result struct {
	Error    error
	Response *http.Response
}

func main() {
	checkStatus := func(urls ...string) <-chan Result {
		results := make(chan Result)
		go func() {
			defer close(results)
			for _, url := range urls {
				resp, err := http.Get(url)
				results <- Result{Error: err, Response: resp}    // 把错误从当前goroutine中propagate出去
			}
		}()
		return results
	}

	errCount := 0
	urls := []string{"https://www.bing.com", "https://badhost"}
	for result := range checkStatus(urls...) {
		if result.Error != nil {                                 // 在这里处理错误更合适。这个位置掌握了更多的信息，比如可以方便地统计错误总数等
			fmt.Printf("error: %v\n", result.Error)
			errCount++
			continue
		}
		fmt.Printf("Response: %v\n", result.Response.Status)
	}
}

package main

import (
	"fmt"
	"sync"
)

func gen(nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		for _, n := range nums {
			out <- n
		}
		close(out)
	}()
	return out
}

func sq(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		for n := range in {
			out <- n * n
		}
		close(out)
	}()
	return out
}

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
}

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
	}
}

// A Context carries a deadline, cancelation signal, and request-scoped values
// across API boundaries. Its methods are safe for simultaneous use by multiple
// goroutines.
type Context interface {
    // Done returns a channel that is closed when this Context is canceled
    // or times out.
    Done() <-chan struct{}

    // Err indicates why this context was canceled, after the Done channel
    // is closed.
    Err() error

    // Deadline returns the time when this Context will be canceled, if any.
    Deadline() (deadline time.Time, ok bool)

    // Value returns the value associated with key or nil if none.
    Value(key interface{}) interface{}
}

 1: var a string
 2: 
 3: func f() {
 4: 	print(a)
 5: }
 6: 
 7: func hello() {
 8: 	a = "hello, world"
 9: 	go f()
10: }

1: var a string
2: 
3: func hello() {
4: 	go func() { a = "hello" }()
5: 	print(a)
6: }