Go并发

前言:

近期开始想认真的学一学一种编程语言，之前学习C和JAVA,Cpp都如同过水浮萍，希望这次能坚持下去。仅以此记录我的go语言学习过程。

go的优越性

性能:go的通道可以使得并发变得很方便。go能自己维护线程池，利用Goroutine可以看作一个轻量级的线程。在go中你不需要去写复杂的进程线程协程，你只需要掌握goroutine的使用，就能简单(也许)的提高代码效率

或许你已经知道了线程进程之间的关系，但是goroutine这个单词其实是取自Coroutines,协程，程序员自己实现的轻量级线程

goroutine

how_to_use

我们使用go关键字调用goroutine, 在下面我给出一个示例并可以体现出goroutine的一些作用。普通函数

package main  
  
import (  
    "fmt"  
)  
  
func hi() {  
    fmt.Println("hi here we go")  
}  
func main() {  
    hi()  
    fmt.Println("Its Man!!!")  
}

加上go


package main  
  
import (  
    "fmt"  
)  
  
func hi() {  
    fmt.Println("hi here we go")  
}  
func main() {  
    go hi()  
    fmt.Println("Its Man!!!")  
}

我们可以发现，第二个只运行了Its Man!!!。只运行了主协程的代码，而hi()似乎被跳过了。其实这里就是因为我们给hi()相当于单开了一个线程，这个hi()和主协程是异步运行的。hi()还没执行完而主协程就执行完了这个程序就结束了。我们如果想要hi()执行可以在后面等他一会。我们可以用sleep,不过其实对于go并发的话，WaitGroup更合适。

1	time.Sleep(time.Second)

我们在末尾加上这个可以延长主协程的时间使得hi()在关闭前完成运行。以此为基础我们能实现多次运行

package main  
  
import (  
    "fmt"  
    "sync")  
  
var wg sync.WaitGroup  
  
func hi() {  
    fmt.Println("hi here we go")  
    wg.Done() //用于减少计数器  
  
}  
func main() {  
    wg.Add(10) //告诉需要连个Goroutine需要等待完成  
    for i := 0; i < 10; i++ {  
       go hi()  
    }  
    wg.Wait() //等待完成  
    fmt.Println("Its Man!!!")  
}

这里是10个goroutine同时运行，所以运行效率比正常遍历快接近10倍。感兴趣可以测一测。

tips

与传统的栈调用不同的是，gortoutine的栈调用是一种动态的。一般LUNix默认系统线程的栈是2MB,这可能导致我们开不了多少个线程不过在GO我们无需担心这一点，gortoutine的栈一般为2kb. 这里我么示例一个计算可以发现

package main  
  
import (  
    "fmt"  
    "sync")  
  
const n = 1 << 16  
  
var wg sync.WaitGroup  
  
func hi(i int) {  
    fmt.Printf("hi here we %d \n", i*i)  
    wg.Done() //用于减少计数器  
  
}  
func main() {  
    wg.Add(n) //告诉需要连个Goroutine需要等待完成  
    for i := 0; i < n; i++ {  
       go hi(i)  
    }  
    wg.Wait() //等待完成  
    fmt.Println("Its Man!!!")  
}

可以看到我这里开了2的16次方个goroutine不过运行起来还是挺快的，一秒左右

runtime包

什么是runtime

runtime 描述了程序运行时候执行的软件/指令，在每种语言有着不同的实现。(当然这只是一种抽象的概念)

go中的runtime包

go中的runtime负责栈管理和并发控制等机制。在我看来它能让我们更灵活的使用goroutine

Gosched()

runtime.Gosched()，让当前执行的goroutine让出CPU，让其更好的获得执行的机会。示例

func d() {  
    for i := 1; i < 10; i++ {  
       fmt.Println("D:", i*i)  
    }  
}  
func main() {  
    go d()  
    runtime.Gosched()  
    println("main")  
  
}

如果这里不加runtime.Goscher()，go d()还没完成执行，主协程就已经完成执行结束了运行，我们加上了之后，主协程会把执行机会让给另一个goroutine d(）。 and如果这样写


func d(j int) {  
    for i := 1; i < 10; i++ {  
       fmt.Println("D:", j, i*i)  
    }  
}  
func main() {  
    for i := 0; i < 10; i++ {  
       go d(i)  
    }  
    runtime.Gosched()  
    println("main")  //并不一定表示主协程结束
  
}

你会发现我们连续创建的10个goroutine并不会全部运行，因为主协程提前完成了运行。

Goexit()

退出当前的协程

func d(j int) {  
    if j%2 == 0 {  
       runtime.Goexit()  
    }  
    for i := 1; i < 10; i++ {  
       fmt.Println("D:", j, i*i)  
  
    }  
    wg.Done()  
}  
  
func main() {  
    wg.Add(5)  
    for i := 0; i < 10; i++ {  
       go d(i)  
    }  
    wg.Wait()  
    println("SUCCESS")  
  
}

比如这里我们就退出了j是偶数的是很的goroutine的运行

GOMAXPROCS

runtime.GOMAXPROCS(n)，PROCS指的其实就是核心的数量。输入n我们可以控制最大的调用的核心数量，如果不写默认调用完。一般调用了n个核心，那么共有n个核心能够同时运行，这里的同时不是指的并发。而是物理并行，剩下的由调度器调度执行。这里涉及到了go语言的G-M-P模型。这里担心自己语言描述不够好这里摘抄一下Go语言中文文档的描述。

GPM是Go语言运行时（runtime）层面的实现，是go语言自己实现的一套调度系统。区别于操作系统调度OS线程。

1.G很好理解，就是个goroutine的，里面除了存放本goroutine信息外还有与所在P的绑定等信息。
2.P管理着一组goroutine队列，P里面会存储当前goroutine运行的上下文环境（函数指针，堆栈地址及地址边界），P会对自己管理的goroutine队列做一些调度（比如把占用CPU时间较长的goroutine暂停、运行后续的goroutine等等）当自己的队列消费完了就去全局队列里取，如果全局队列里也消费完了会去其他P的队列里抢任务。
3.M（machine）是Go运行时（runtime）对操作系统内核线程的虚拟， M与内核线程一般是一一映射的关系，一个groutine最终是要放到M上执行的；

P与M一般也是一一对应的。他们关系是： P管理着一组G挂载在M上运行。当一个G长久阻塞在一个M上时，runtime会新建一个M，阻塞G所在的P会把其他的G 挂载在新建的M上。当旧的G阻塞完成或者认为其已经死掉时回收旧的M。

P的个数是通过runtime.GOMAXPROCS设定（最大256），Go1.5版本之后默认为物理线程数。在并发量大的时候会增加一些P和M，但不会太多，切换太频繁的话得不偿失。

Channel

单纯的多个函数并发是没有意义的，函数之间要有数据交换才有意义。go语言就是通过channel共享内存

CSP模型

CSP 是 Communicating Sequential Process 的简称，中文可以叫做通信顺序进程，是一种并发编程模型，是一个很强大的并发数据模型，是上个世纪七十年代提出的，用于描述两个独立的并发实体通过共享的通讯 channel(管道)进行通信的并发模型。相对于Actor模型，CSP中channel是第一类对象，它不关注发送消息的实体，而关注与发送消息时使用的channel。

channel的使用

channel的类型创建

channel是一种引用类型，声明通道类型的格式

1	var 变量 chan 元素类型

channel的操作

1	make(chan 元素类型1，[缓冲区大小])

send

1	ch<-10 //接受数据到其中

recv

1	x:=<-ch //从其中接受数据

关闭

close(ch)

channel的特性

阻塞

简单说，用发送必须有接受，否则会卡住。发送和接受方必须同步。

无缓冲通道无缓冲的通道又叫做阻塞的通道。其实就是上面说的特性
有缓冲的通道初始化给定通道一个容量。没什么好说的当生产者的效率大于消费者的时候我们设立一个缓冲。

package main  
  
import (  
    "fmt"  
    "time")  
  
func main() {  
    logs := make(chan string, 10) // 缓冲区=5  
  
    // 日志生产者（速度快）  
    go func() {  
       for i := 0; i < 20; i++ {  
          logs <- fmt.Sprintf("log-%d", i)  
          fmt.Println("生产:", i)  
          time.Sleep(50 * time.Millisecond) // 模拟快生产  
       }  
       close(logs)  
    }()  
  
    // 日志消费者（写文件，速度慢）  
    go func() {  
       for log := range logs {  
          fmt.Println("消费:", log)  
          time.Sleep(200 * time.Millisecond) // 慢消费  
       }  
    }()  
  
    time.Sleep(5 * time.Second)  
}

对生产效率的提升不算很大，不过在大型场景的是很会避免一些阻塞情况

单向通道

通道也可以限定方向，只能send或recv。

package main  
  
import "fmt"  
  
func counter(out chan<- int) {  
    for i := 0; i < 123; i++ {  
       out <- i  
    }  
    close(out)  
}  
  
func squarer(out chan<- int, in <-chan int) {  
    for i := range in {  
       out <- i * i  
    }  
    close(out)  
}  
func printer(in <-chan int) {  
    for i := range in {  
       fmt.Println(i)  
    }  
}  
  
func main() {  
    ch1 := make(chan int)  
    ch2 := make(chan int)  
    go counter(ch1)  
    go squarer(ch2, ch1)  
    printer(ch2)  
}

一些应用

示例一个题目。

from Crypto.Util.number import *
from hashlib import md5
from secret import KEY， flag  


assert int(KEY).bit_length() == 36
assert not isPrime(KEY)

p = getPrime(1024)
q = getPrime(1024)
n = p * q
e = 0x10001

ck = pow(KEY, e, n)


assert flag == b'NSSCTF{' + md5(str(KEY).encode()).hexdigest().encode() + b'}'

print(f"{n = }")
print(f"{e = }")
print(f"{ck = }")

'''
n = 26563847822899403123579768059987758748518109506340688366937229057385768563897579939399589878779201509595131302887212371556759550226965583832707699167542469352676806103999861576255689028708092007726895892953065618536676788020023461249303717579266840903337614272894749021562443472322941868357046500507962652585875038973455411548683247853955371839865042918531636085668780924020410159272977805762814306445393524647460775620243065858710021030314398928537847762167177417552351157872682037902372485985979513934517709478252552309280270916202653365726591219198063597536812483568301622917160509027075508471349507817295226801011
e = 65537
ck = 8371316287078036479056771367631991220353236851470185127168826270131149168993253524332451231708758763231051593801540258044681874144589595532078353953294719353350061853623495168005196486200144643168051115479293775329183635187974365652867387949378467702492757863040766745765841802577850659614528558282832995416523310220159445712674390202765601817050315773584214422244200409445854102170875265289152628311393710624256106528871400593480435083264403949059237446948467480548680533474642869718029551240453665446328781616706968352290100705279838871524562305806920722372815812982124238074246044446213460443693473663239594932076

'''

原版，正常用python可能要花个5-6分钟但是用.go会快很多5-6秒

package main  
  
import (  
    "crypto/md5"  
    "fmt"    "math/big"    "sync")  
  
func main() {  
    // 参数  
    nStr := "26563847822899403123579768059987758748518109506340688366937229057385768563897579939399589878779201509595131302887212371556759550226965583832707699167542469352676806103999861576255689028708092007726895892953065618536676788020023461249303717579266840903337614272894749021562443472322941868357046500507962652585875038973455411548683247853955371839865042918531636085668780924020410159272977805762814306445393524647460775620243065858710021030314398928537847762167177417552351157872682037902372485985979513934517709478252552309280270916202653365726591219198063597536812483568301622917160509027075508471349507817295226801011"  
    e := big.NewInt(65537)  
  
    ckStr := "8371316287078036479056771367631991220353236851470185127168826270131149168993253524332451231708758763231051593801540258044681874144589595532078353953294719353350061853623495168005196486200144643168051115479293775329183635187974365652867387949378467702492757863040766745765841802577850659614528558282832995416523310220159445712674390202765601817050315773584214422244200409445854102170875265289152628311393710624256106528871400593480435083264403949059237446948467480548680533474642869718029551240453665446328781616706968352290100705279838871524562305806920722372815812982124238074246044446213460443693473663239594932076"  
  
    n := new(big.Int)  
    n.SetString(nStr, 10)  
    ck := new(big.Int)  
    ck.SetString(ckStr, 10)  
  
    // 计算 inv_ck = ck^{-1} mod n    inv_ck := new(big.Int).ModInverse(ck, n)  
  
    // 指定线程数  
    numWorkers := 16 << 12  
  
    var dic sync.Map  
    var wg sync.WaitGroup  
  
    limitS := 1 << 20  
    chunkSize := limitS / numWorkers  
  
    // 并发生成字典  
    for c := 0; c < numWorkers; c++ {  
       start := c * chunkSize  
       end := (c + 1) * chunkSize  
       wg.Add(1)  
       go func(start, end int) {  
          defer wg.Done()  
          invE := new(big.Int).Neg(e) // -e  
          for i := start + 1; i <= end; i++ {  
             iBig := new(big.Int).SetInt64(int64(i))  
             v := new(big.Int).Exp(iBig, invE, n)  
             dic.Store(v.String(), i)  
  
          }  
       }(start, end)  
    }  
    wg.Wait()  
    fmt.Println("预计算完成，字典就绪。")  
  
    // 并发搜索  
    found := false  
    var foundSecret *big.Int  
    limitJ := 1 << 18  
    chunkSizeJ := limitJ / numWorkers  
  
    var mu sync.Mutex  
    for c := 0; c < numWorkers; c++ {  
       start := c * chunkSizeJ  
       end := (c + 1) * chunkSizeJ  
       wg.Add(1)  
       go func(start, end int) {  
          defer wg.Done()  
          for j := start + 1; j <= end; j++ {  
             if found {  
                return  
             }  
             jBig := new(big.Int).SetInt64(int64(j))  
             powJ := new(big.Int).Exp(jBig, e, n)  
             s := new(big.Int).Mul(inv_ck, powJ)  
             s.Mod(s, n)  
  
             if val, ok := dic.Load(s.String()); ok {  
                i := val.(int) // 强转为 int                iBig := new(big.Int).SetInt64(int64(i))  
                secret := new(big.Int).Mul(iBig, jBig)  
                fmt.Printf("j=%d,i=%d", j, i)  
                mu.Lock()  
                if !found {  
                   found = true  
                   foundSecret = secret  
                }  
                mu.Unlock()  
                return  
             }  
          }  
       }(start, end)  
    }  
    wg.Wait()  
  
    if found {  
       data := []byte(foundSecret.String())  
       h := md5.Sum(data)  
       flag := fmt.Sprintf("NSSCTF{%x}", h)  
       fmt.Println("FLAG:", flag)  
    } else {  
       fmt.Println("搜索完成，但未找到匹配项。")  
    }  
}

参考

并发编程

本博客采用知识共享署名-非商业性使用-相同方式共享 4.0 国际许可协议(CC BY-NC-SA 4.0) 发布.