goConcurrency/selectStmt.go

package goConcurrency

import (
	"fmt"
	"math/rand"
	"os"
	"os/signal"
	"sync"
	"time"
)

func SelectStmt() {
	// 声明需要的变量
	var a [4]int // 注意后边索引的使用
	var c1, c2, c3, c4 = make(chan int), make(chan int), make(chan int), make(chan int)
	//var i1, i2 = 0, 42
	var i1 = 0

	// 操作channel的goroutine
	go func() {
		c1 <- 10
	}()
	go func() {
		<-c2
	}()
	go func() {
		close(c3)
	}()
	go func() {
		c4 <- 40
	}()

	// select 多路监听的goroutine
	go func() {
		select {
		// 监听是否可以从 c1 中接收值
		case i1 = <-c1:
			println("received ", i1, " from channel c1")
		// 监听对c2的写操作
		//case c2 <- i2:
		//	println("sent ", i2, " to channel c2")
		case c2 <- i2Value():
			println("sent ", i2Value(), " to channel c2")
		// 监听c3的关闭状态
		case i3, ok := <-c3:
			if ok {
				println("received ", i3, " from channel c3")
			} else {
				println("channel c3 was closed")
			}
		// 测试左值表达式的执行时机
		case a[f()] = <-c4:
			println("received ", a[f()], " from channel c4")
		// 默认case
		default:
			println("on channel operation.")
		}
	}()

	// 简单阻塞，等待goroutine执行完毕
	time.Sleep(100 * time.Millisecond)
}

func i2Value() int {
	println("send value is evaluate")
	return 42
}

func f() int {
	print("f() is running.")
	return 2
}

func SelectBlock() {
	// 空select阻塞
	//println("before select")
	//select {} // block()
	//println("after select")

	// nil select阻塞
	var ch chan int // nil channel, 不能receive和send
	go func() {
		ch <- 1024
		println("send data")
	}()
	println("before select")
	select {
	case <-ch:
	case ch <- 42:
	}
	println("after select")
}

func SelectNilChannel() {
	// 一，初始化channel
	ch := make(chan int) // non nil channel

	// 二，操作channel的goroutine
	go func() {
		// 随机写入int
		rand.Seed(time.Now().Unix())
		for {
			ch <- rand.Intn(10)
			time.Sleep(400 * time.Millisecond)
		}
	}()

	// 三，select处理定时内的channel
	// 到达时间，停止处理
	go func() {
		// 设置定时器
		t := time.After(3 * time.Second)
		sum := 0
		for {
			select {
			case v := <-ch:
				println("received value ", v)
				sum += v
			case <-t:
				// 将 channel 设置为 nil，不再处理ch
				ch = nil
				println("ch was set inl, sum is ", sum)
			}
		}
	}()

	time.Sleep(5 * time.Second)
}

//func main() {
//	println("before select")
//	select {} // block()
//	println("after select")
//}

func SelectFor() {
	// 定义channel
	ch := make(chan int)
	// 操作Channel
	// send to channel
	go func() {
		for {
			// 模拟演示数据来自于随机数
			//实操时，数据可以来自各种I/O，例如网络、缓存、数据库等
			ch <- rand.Intn(100)
			time.Sleep(200 * time.Millisecond)
		}
	}()

	// 监控Channel
	go func() {
		// 持续监控
		for {
			select {
			case v := <-ch:
				println("received from channel, value is ", v)
			}
		}
	}()

	time.Sleep(3 * time.Second)

}

func SelectNonBlock() {
	// 一，初始化数据
	counter := 10 // 参与人数
	max := 20     // [0, 19] // 最大范围
	rand.Seed(time.Now().UnixMilli())
	answer := rand.Intn(max) // 随机答案
	println("The answer is ", answer)
	println("------------------------------")

	// 正确答案channel
	bingoCh := make(chan int, counter)

	// 二，模拟猜
	// wg
	wg := sync.WaitGroup{}
	wg.Add(counter)
	for i := 0; i < counter; i++ {
		// 每个goroutine代表一个猜数字的人
		go func() {
			defer wg.Done()
			result := rand.Intn(max)
			println("someone guess ", result)
			// 答案争取，写入channel
			if result == answer {
				bingoCh <- result
			}
		}()
	}
	wg.Wait()

	println("------------------------------")
	// 三，检测是否有人猜中
	select {
	case result := <-bingoCh:
		println("some one hint the answer ", result)
	default:
		// 非阻塞的保证，存在default case
		println("no one hint the answer")
	}
}

func SelectRace() {
	// 一，初始化数据
	// 模拟查询结果，需要与具体的querier建立联系
	type Rows struct {
		// 数据字段

		// 索引标识
		Index int
	}
	// 模拟的querier数量
	const QuerierNum = 8
	// 用于通信的channel，数据，停止信号
	ch := make(chan Rows, 1)
	stopChs := [QuerierNum]chan struct{}{}
	for i := range stopChs {
		stopChs[i] = make(chan struct{})
	}
	// wg,rand
	wg := sync.WaitGroup{}
	rand.Seed(time.Now().UnixMilli())

	// 二，模拟querier查询，每个查询持续不同的时间
	wg.Add(QuerierNum)
	for i := 0; i < QuerierNum; i++ {
		// 每一个 querier
		go func(i int) {
			defer wg.Done()
			// 模拟执行时间
			randD := rand.Intn(1000)
			println("querier ", i, " start fetch data, need duration is ", randD, " ms.")
			// 查询结果的channel
			chRst := make(chan Rows, 1)

			// 执行查询工作
			go func() {
				// 模拟时长
				time.Sleep(time.Duration(randD) * time.Millisecond)
				chRst <- Rows{
					Index: i,
				}
			}()

			// 监听查询结果和停止信号channel
			select {
			// 查询结果
			case rows := <-chRst:
				println("querier ", i, " get result.")
				// 保证没有其他结果写入，才写入结果
				if len(ch) == 0 {
					ch <- rows
				}
			// stop信号
			case <-stopChs[i]:
				println("querier ", i, " is stopping.")
				return
			}

		}(i)
	}

	// 三，等待第一个查询结果的反馈
	wg.Add(1)
	go func() {
		defer wg.Done()
		// 等待ch中传递的结果
		select {
		// 等待第一个查询结果
		case rows := <-ch:
			println("get first result from ", rows.Index, ". stop other querier.")
			// 循环结构，全部通知querier结束
			for i := range stopChs {
				// 当前返回结果的goroutine不需要了，因为已经结束
				if i == rows.Index {
					continue
				}
				stopChs[i] <- struct{}{}
			}

		// 计划一个超时时间
		case <-time.After(5 * time.Second):
			println("all querier timeout.")
			// 循环结构，全部通知querier结束
			for i := range stopChs {
				stopChs[i] <- struct{}{}
			}
		}
	}()

	wg.Wait()
}

func SelectAll() {
	// 一，初始化资源
	// 整体的资源类型
	type Content struct {
		Subject string
		Tags    []string
		Views   int
		// 标识操作了资源的哪个部分
		part string
	}
	// 三个用于表示不同部分的常量
	const (
		PartSubject = "subject"
		PartTags    = "tags"
		PartViews   = "views"
	)
	// 同步停止的信号channel的map
	stopChs := map[string]chan struct{}{
		PartSubject: make(chan struct{}),
		PartTags:    make(chan struct{}),
		PartViews:   make(chan struct{}),
	}
	// 接收和发送操作的通信channel
	ch := make(chan Content, len(stopChs))
	// timeover 表示超时的channel
	to := time.After(time.Millisecond * 800)
	// waitgroup
	wg := sync.WaitGroup{}
	// 初始化全局随机数种子
	rand.Seed(time.Now().UnixMilli())

	// 二，goroutine执行每个部分的获取
	// 为每个部分的处理使用goroutine完成
	for part := range stopChs {
		wg.Add(1)
		go func(part string) {
			defer wg.Done()
			// 一，初始化信息
			randD := rand.Intn(1000) //ms
			fmt.Println("start fetch ", part, " data, need duration is ", randD, " ms.")
			// 查询结果的channel
			chRst := make(chan Content, 1)

			// 二，模拟延时的获取资源
			go func() {
				// 模拟执行时间
				time.Sleep(time.Duration(randD) * time.Millisecond)
				// 模拟业务逻辑
				content := Content{
					part: part,
				}
				// 基于不同的part，完成不同属性的设置
				switch part {
				case PartSubject:
					content.Subject = "Subject of content"
				case PartTags:
					content.Tags = []string{"go", "Goroutine", "Channel", "select"}
				case PartViews:
					content.Views = 1024
				}
				// 发送到rstCh
				chRst <- content
			}()

			// 三，监控资源获取成功还是超时到达
			select {
			// 查询到结果
			case rst := <-chRst:
				fmt.Println("querier ", part, " get result.")
				ch <- rst
			// 超时到了
			case <-stopChs[part]:
				fmt.Println("querier ", part, " is stopping.")
				return
			}
		}(part)
	}

	// 三，接收每个部分，整合在一起
	wg.Add(1)
	go func() {
		defer wg.Done()
		// 一，初始化资源
		// 整体的资源
		content := Content{}
		// 标识哪个部分已经接收完毕
		received := map[string]struct{}{}

		// 二，等待接收或者超时到期
		// 超时时间到要通知未完成的goroutine结束
		// 未到超时时间，将结果整合到一起，并判定是否需要继续等待
	loopReceive:
		for {
			select {
			// 超时
			case <-to:
				println("querier timeout. Content is incomplete.")
				// 超时时间到要通知未完成的goroutine结束
				// 遍历stopCh,判定是否存在与received中即可
				for part := range stopChs {
					if _, exists := received[part]; !exists {
						// 不存在，说明没有处理完，应该技术
						stopChs[part] <- struct{}{}
					}
				}
				// 关闭
				close(ch)
				// 不再继续监听了，结束！
				break loopReceive

			// 有处理完毕的业务
			case rst := <-ch:
				println("received some part ", rst.part)
				// 根据不同的part，更新整体content的字段
				// 同时要记录，哪个part已经完成接收了
				switch rst.part {
				case PartSubject:
					content.Subject = rst.Subject
					received[PartSubject] = struct{}{}
				case PartTags:
					content.Tags = rst.Tags
					received[PartTags] = struct{}{}
				case PartViews:
					content.Views = rst.Views
					received[PartViews] = struct{}{}
				}
				// 判定是否已经接收完毕，需要继续等待
				finish := true // 完成标志
				// 确认是否都接收了
				for part := range stopChs {
					if _, exists := received[part]; !exists {
						// 不存在，说明存在没有处理完毕的，没有结束
						finish = false
						break
					}
				}
				// 判定 finish
				if finish {
					// 说明，全部已经处理完毕，结束了！
					fmt.Println("all querier finished. Content is complete")
					close(ch)
					break loopReceive
				}

			}
		}

		// 三，输出结果
		fmt.Println("content:", content)
	}()

	wg.Wait()

}

func SelectChannelCloseSignal() {
	wg := sync.WaitGroup{}
	// 定义无缓冲channel
	// 作为一个终止信号使用（啥功能的信号都可以，信号本身不分功能）
	ch := make(chan struct{})

	// goroutine，用来close, 表示发出信号
	wg.Add(1)
	go func() {
		defer wg.Done()
		time.Sleep(2 * time.Second)
		fmt.Println("发出信号, close(ch)")
		close(ch)
	}()

	// goroutine，接收ch，表示接收信号
	wg.Add(1)
	go func() {
		defer wg.Done()
		// 先正常处理，等待ch的信号到来
		for {
			select {
			case <-ch:
				fmt.Println("收到信号, <-ch")
				return
			default:

			}
			// 正常的业务逻辑
			fmt.Println("业务逻辑处理中....")
			time.Sleep(300 * time.Millisecond)
		}
	}()

	wg.Wait()
}

func SelectSignal() {
	// 一：模拟一段长时间运行的goroutine
	go func() {
		for {
			fmt.Println(time.Now().Format(".15.04.05.000"))
			time.Sleep(300 * time.Millisecond)
		}
	}()

	// 要求主goroutine等待上面的goroutine，方案：
	// 1. wg.Wait(),要可以保证goroutine可以正常结束return
	// 2. time.Sleep()，要保证时间周期的长度够用，但又不能太久
	// 3. select{}，持久的阻塞

	// 持久阻塞
	//select {}

	// 二，监控系统的中断信号,interrupt
	// 1 创建channel，用于传递信号
	chSignal := make(chan os.Signal, 1)
	// 2 设置该channel可以监控哪些信号
	signal.Notify(chSignal, os.Interrupt)
	//signal.Notify(chSignal, os.Interrupt, os.Kill)
	//signal.Notify(chSignal) // 全部类型的信号都可以使用该channel
	// 3 监控channel
	select {
	case <-chSignal:
		fmt.Println("received os signal: Interrupt")
	}
}