第五章 Go语言并发

喜欢ヅ旅行 2023-06-02 13:56 11阅读 0赞

## 一、goroute ##

#### 　　理论 ####

进程：进程是程序在操作系统中的一次执行过程，系统进行资源分配和调度的一个独立单位

线程：线程是进程的一个执行实体，是CPU调度和分配的基本单位，是比进程跟小的能独立运行的基本单位

进程与线程的关系：一个进程一颗创建和撤销多个线程，同一个进程中的多个线程之间可以并发执行

协程：独立的栈空间，共享堆空间，调度由用户自己控制，本质上优点类似用户级线程，这些用户线程的调用也是自己实现的

协程和线程的关系：一个线程可以跑多个协程，协程是轻量级线程

并发：同一时间段内执行多个操作

并行：同一时刻执行多个操作

Gorouteine多线程：线程是有操作系统进行管理，处于内核态，线程之间进行切换需要发生用户态到内核态的切换

当系统中运行大量线程，系统会变得非常慢，用户态的线程支持大量线程的创建，也叫协程或者goroutine

#### 　　Goroutine使用介绍 ####

//创建goroutine
    func hello() {
        fmt.Println("hello goroutine")
    }
    func main() {
        go hello()
        fmt.Prinln("main thread terminate")
        time.Sleep(time.Second)    //goroutine会随着主线程的结束而结束，需要等待1s
    }
    
    //启动多个goroutine
    func hello(i int) {
        fmnt.Println("hello goroutine",i)
    }
    func test() {
        for i := 0<10;i++{
            go hello(i)
        }
    }
    func main() {
        test()
        time.Sleep(time.Second)
    }
    
    //多核控制
    var i int
    func calc() {for {i++)}
    func main() {
        cpu := runtime.NumCPU()    //显示当前CPU有多少核
        fmt.Prinln("cpu",cpu)
        runtime.GOMAXPROCS(1)    //只允许调用一个cpu内核
        for i := 0;i<10;i++ {
            go calc()
        }
        time.Sleep(time.Hour)
    }

#### 　　Goroutine原理解析 ####

概念：一个操作系统线程对应用户态多个goroutine，可同时使用多个操作系统线程，操作系统线程对goroutine是多对多关系

调度：当操作系统线程需要执行时，开始逐个执行goroutine

![1648958-20190901170543136-599530989.png][]蓝色goroutine执行结束后再执行灰色，当前有两个线程在执行

系统调用的处理

![1648958-20190901170646976-710050573.png][]

## 二、goroutine之间的通信 ##

#### 　　全局变量和锁同步 ####

import (
        "fmt"
        "time"
        "sync"
    )
    var  m make(map[int]uint64)
    type tesk struct {
        n int
    }
    func calc(t *task) {
        var sum uint64
        sum = 1
        for i := 1;i<t.n;i++ {
            sum *= uint64(i)
        }
        lock.Lock()
        m[t.n] = sum
        lockUnlock()
    }
    func main() {
        for i := 0;i<100;i++ {
            t := &task{n:i}
            go calc(t)
        }
        time.Sleep(10 * time.Second)
        lock.Lock()
        for k,v := range m {
            fmt.Printf("%d!=%v\n",k,v)
        }
        lock.Unlock()
    }

#### 　　channel(队列管道) ####

介绍：本质上是一个队列/容器，先进先出，类似unix中的管道pipe

定义：需要指定勇气中元素的类型　　var 变量名 chan 数据类型

var test chan int
    var test chan string
    var test chan map[string]string
    var test chan stcurt
    var test chan *stcurt    //定义一个地址，只能传入地址

入队：a <- 100

出队：data := <-a

//test1    
    func main() {    
        var c chan int
        fmt.Printf("c =%v\n",c)
        c = make(chan int,10)    //如果不分配空间，则入队一直阻塞，运行报错
        fmt.Printf("c = %v\n",c)
        c <- 100        //<-c不给变量存储则为丢弃这个元素
        data := <-c
        fmt.Println("data",data)
    }
    
    //test2
    type student struct {
        name string
    }
    func main() {
        var stuChan chan student
        stuChan = make(chan *Student,10)
        stu := student{name:"stu01"}
        stuChan <- stu
    }
    
    //test3
    func produce(c chan int) {
        c <- 1000
        fmt.Println("produce finished")
    }
    func consume (c chan int) {
        data := <-c
        fmt.Println("data",data)
    }
    func main() {
        var c chan int
        fmt.Printf("c = %v\n",c)
        c = make(chan int)
        go produce(c)
        go consume(c)
        time.Sleep(time.Second * 3)
    }    //不带缓冲区的队列，有出队的时候才能入队，否则入队失败

#### 　　channel练习 ####

//chan和goroutine同步
    func hello(c chan bool){
        fmt.Println("hello goroutine")
        c <- true
    }
    func main(){
        var exitChan chan bool
        exitChan = make(chan bool)
        go hello(exitChan)
        fmt.Println("main thread terminate")
        <-exitChan
    }
    
    //单向chan    
    func sendData(sendch chan<- int) {    chan<- 表示之能入队
        sendch <- 10
    }
    func readData(sendch <-chan int) {    <- chan 表示只能出队
        data := <-sendch
        fmt.Println("data:", data)
    }
    func main() {
        chnl := make(chan int)
        go sendData(chnl)
        readData(chnl)
    }
    
    //chan关闭
    func producer(chnl chan int) {
        for i := 0; i < 10; i++ {
            chnl <- i
        }
        close(chnl)    //关闭chan
    }
    func main() {
        ch := make(chan int)
        go producer(ch)
        for {
            v, ok := <-ch
            if ok == false {    //判断chan是否关闭，如果无法判断管道是否关闭会一直取出默认值0保持死循环    
                fmt.Println("chan is closed")
                break
            }
            fmt.Println("Received", v, ok)
        }
    }
    
    
    //for range
    func produer(chnl chan int) {
        for i := 0; i < 10; i++ {
            chnl <- i
        }
        close(chnl)
    }
    func main() {
        ch := make(chan int)
        go produer(ch)
        for v := range ch {    //chan关闭无数据后自动运行结束
            fmt.Println("receive:", v)
        }
    }
    
    //带缓冲区的chanel
    func write( ch chan int) {
        for i:= 0;i<5;i++ {
            ch <- i
            fmt.Println("successfully wrote",i,"to ch")
        }
        close(ch)
    }
    func main() {
        ch := make(chan int,2)
        go write(ch)
        time.Sleep(2 * time.Second)
        for v := range ch {
            fmt.Println("read value",v,"from ch")
            time.Sleep(2 *time.Second)
        }
    }
    
    //channel的长度和容量
    func main() {
        ch := make(chan string,3)
        ch <- "naveen"
        ch <- "paul"
        fmt.Println("capacity is",cap(ch))
        fmt.Println("length is",len(ch))
        fmt.Println("read value",<-ch)
        fmt.Println("new length is",len(ch))
    }
    
    //waitgroup等待一组goroutine结束,使用不带缓冲区的channel实现
    func process(i int,ch chan bool) {
        fmt.Println("started Goroutine",i)
        time.Sleep(2 * time.Second)
        fmt.Printf("Goroutine %d ended\n",i)
        ch <- true
    }
    func main() {
        no := 3
        exitChan := make(chan bool,no)
        for i := 0;i<no;i++ {
            go process(i,exitChan)
        }
        for i:= 0;i<no;i++ {
            <exitChan
        }
        fmt.Println("All goroutines finshed executing")
    }
    //使用sync-WaitGroup实现
    func process(i int,wg *sync.WaitGroup) {
        fmt.Println("started Goroutine",i)
        time.Sleep(2 * time.Second)
        fmt.Printf("Goroutine %d ended\n",i)
        wg.Done()    //wg-1
    }
    func main() {
        no := 3
        var wq sync.WaitGroup    //定义一个wgh，默认为0
        for i := 0; i<no; i++ {
        wg.Add(1)    wg+1
        go process(i,&wg)
    }
        wg.Wait()    //当wg为0时才会向下执行
        fmt.Println("All goroutines finished executing“）
    }”

####  ####

## 三、select和线程之间的安全 ##

#### 　　select语义介绍 ####

定义：多channel操作，同时监听一个或多个channel，直到其中一个channel ready，如果其中合格多个chgannel同时ready，随机选择一个进行操作，语法和weitch case类似，只获取一个数值就结束

func server1(ch chan string) {
        time.Sleep(time.Second * 3)
        ch <- "response from server1"
    }
    func server2(ch chan string) {
        time.Sleep(time.Second)
        ch <-"response from server2")
    }
    func main() {
        output1 := make(chan string)
        output2 := make(chan stirng)
        go server1(output1)
        go server2(output2)
        //s1 := <-output1    //等待3s
        //fmt.Println("s1:",s1)
        //s2 := <-output2    //等待1s后等待s1结束再执行
        //fmt.Println("s2",s2)
        select {    //最终只取一个channel的结果
        case s1 := <-output1:
            fmt.Println("s1:",s1)
        case s2 := <-output2:
            fmt.Prinln("s2:",s2)
        }
    }

default分支：放case分支的channel都没有ready的话，执行default，用来判断channel是否满了或者为空

//test1    
    func server1(ch chan string) {
        time.Sleep(time.Second * 3)
        ch <- "resdponse from server1"
    }
    func server2(ch chan string) {
        time.Sleep(time.Second)
        ch <- "response from server2"
    }
    func main() {
        output1 := make(chan string)
        output2 := make(chan string)
        go server1(output1)
        go server2(output2)
        select {    //最终只取一个channel的结果
        case s1 := <- output1:
            fmt.Println("s1;",s1)
        case s2 := <-output2:
            fmt.Println("s2:",s2)
        default:
            fmt.Println("run default")    //因为s1和s2需要等待，default瞬间执行
        }
    }
    
    //test2
    func write(ch chan string) {
        for {
            select {
            case ch <- "hello":
                fmt.,Println("write succ")
            default:
                fmt.Println("channel is full")
            }
            time.Sleep(time.Millisecond * 500)
        }
    }
    func main() {
        output1 := make(chan string,10)
        go write(output1)
        for s := range.output1 {
            fmt.Println("recv:"s)
            time.Sleep(time.Second)
        }
    }

#### 　　线程安全介绍 ####

定义：多个goroutine同时操作一个资源，这个资源又叫做临界区

var x int
    var wg sync WaitGrroup
    func add() {
        for i := 0;i<500;i++ {
            x = x + 1
        }
        wg.Done()
    }
    func main() {
        wg.Add(2)
        go add()
        go add()
        wg.Wait()
        fmt.Println("x:"x)    //结果不一定等于1000
    }

#### 　　互斥锁介绍 ####

定义：同时有且只有一个线程进入临界区，其他的线程则在等待锁，当互斥锁释放之后，等待锁的线程才可以获取锁进入临界区，多个线程同时等待同一个锁，唤醒的策略是随机的

var x int
    var wg synv.WaitGroup
    var mutex sync.Mutex    //生明锁
    func add() {
        for i :=0；i<500;i++ {
            mutex.Lock()    //加锁
            x = x + 1
            mutex.Unlock()    //释放
        }
        wg.Done()
    }
    func main() {
        wg.Add(2)
        go add()
        go add()
        wg.Wait()
        fmt.Println("x:",x)
    }

#### 　　读写锁介绍 ####

定义：读多写少的场景，分为读锁和写锁，当一个goroutine获取写锁之后，其他goroutine获取写锁或读锁都需要等待

当一个goroutine获取读锁之后，其他的goroutine获取写锁都会等待，但其他goroutine获取读锁时，

var rwlock sync.RWMutex
    var wg sync.WaitGroup
    var x int
    func write() {
        rwlock.Lock()
        fmt.Println("write lock")
        x = x + 1
        time.Sleep(time.Second * 2)
        fmt.Println("write unlock")
        rwlock.Unlock()
        wg.Done()
    }
    func read(i int) {
        rwlock.RLock()
        fmt.Printf("goroutine:%d x=%d\n",i,x)
        time.Sleep(time.Second)
        rwlock.RUnlock()
        wg.Done()
    }
    func main() {
        wg.Add(1)
        go write()
        for i:=0;i<10;i++ {
            wg.Add(1)
            go read(i)
        }
        wg.Wait()
    }
    
    /*
    goroutine:9 x=0    //当channel中没有数据时，返回默认值
    goroutine:0 x=0
    write lock
    write unlock
    goroutine:5 x=1
    goroutine:3 x=1
    goroutine:8 x=1
    goroutine:2 x=1
    goroutine:6 x=1
    goroutine:1 x=1
    goroutine:7 x=1
    goroutine:4 x=1
    */

读写锁和互斥锁的对比

//读写锁
    var rwlok sync.RWMutex
    var wg sync.WaitGroup
    var x int
    func write() {
        for i :=0;i<100;i++ {
            rwlock.Lock()
            x = x+1
            time.Sleep(time.Millisecond * 10)
            rwlock.Unlock()
        }
        wg.Done()
    }
    func read(i int) {
        for i:=0;i<100;i++ {
            rwlock.RLock()
            time.Sleep(time.Millsecond)
            rwlock.RUnlock()
        }
        wg.Done()
    }
    func main() {
        start := time.Now().UnixNano()
        wg.Add(1)
        go write()
        for i:=0;i<10;i++ {
            wg.Add(1)
            go read(i)
        }
        wg.Wait()
        end := time.Now().UnixNano()
        cost := (end-start)/1000/1000
        fmt.Println("cost:",cost,"ms")    //cost:1173ms
    }
    
    //互斥锁
    var wg synv.WaitGroup
    var mutex synv.Mutex
    var x int
    func write() {
        for i:=0;i<100;i++ {
            mutex.Lock()
            x = x+1
            time.Sleep(time.Millisecond * 10)
            mutex.Unlock()
        }
        wg.Done()
    }
    func read(i int) {
        for i:=0;i<100;i++ {
            mutex.Lock()
            time.Sleep(time.Millisecond)
            mutex.Unlock()
        }
        wg.Done()
    }
    func main() {
        start := time.Now().UnixNano()
        wg.Add(1)
        go write()
        for i:= 0;i<10;i++ {
            wg.Add(1)
            go read(i)
        }
        wg.Wait()
        end := time.Now().UnixName()
        cost := (end-start)/1000/1000
        fmt.,Println("cost:",cost,"ms")    //cost:2410ms
    }

#### 　　原子操作介绍 ####

定义：加锁代价比较耗时需要上下文切换，针对基本数据类型，可以使用原子操作保证线程安全，原子操作在用户态就可以完成，因此性能比互斥锁要高

内容：

//加减操作
    func AddInt32(addr *int32,delta int32)(new int32)
    func AddInt64(addr *int64,delta int64)(new int64)
    func AddUint32(addr *uint32,delta uint32)(new uint32)
    func AddUint64(addr *int64,delta uint64)(new uint64)
    func AddUintptr(addr *uintptr,delta uintptr)(new uintptr)
    
    //比较并交换
    func CompareAndSwapInt32(addr *int32,old,new int32)(swapped bool)
    func CompareAndSwapInt64(addr *int64,old,new int64)(swapped bool)
    func CompareAndSwapPointer(addr *unsafe,Pointer,old,new unsafe.Pointer)(swapped bool)
    func CompareAndSwapUint32(addr *uint32,old,new uint32)(swapped bool)
    func CompareAndSwapUint64(addr *uint64,old,new uint64)(swapped bool)
    func CompareAndSwapUintptr(addr *uintptr,old,new uintptr)(swapped bool)
    
    //读取操作
    func LoadInt32(addr *int32)(val int32)
    func LoadInt64(addr *int64)(val int64)
    func LoadIntPointer(addr *unsafe.Pointer)(val unsafe.Pointer)
    func LoadUint32(addr *uint32)(val uint32)
    func LoadUint64(addr *uint64)(val uint64)
    func LoadUintptr(addr *uintptr)(val uintptr)
    
    //写入操作
    func StoreInt32(addr *int32,val int32)
    func StoreInt64(addr *int64,val int64)    
    func StorePointer(addr *unsafe.Pointer,val unsafe.Pointer)
    func StoreUint32(addr *uint32,val uint32)
    func StoreUint64(addr *uint64,val uint64)
    func StoreUintptr(addr *uintptr,val uintptr)
    
    //交换操作
    func SwapInt32(addr *int32,new int32)(old int32)
    func SwapInt64(addr *int64,new int64)(old int64)
    func SwapIntPointer(addr *unsafe.Pointer,new unsafe.Pointer)(old unsafe.Pointer)
    func SwapUint32(addr *uint32,new uint32)(old uint32)
    func SwapUint64(addr *uint64,new uint64)(old uint64)
    func SwapUintptr(addr *uintptr,new uintptr)(old uintptr)

原子性操作和互斥锁的对比

//test'
    var x int32
    var wg sync.WaitGroup
    func add() {
        for i:= 0;i<500;i++ {
            atomic.AddInt32(&x,1)
        }
        wg.Done()
    }
    func main() {
        wg.Add(2)
        go add()
        go add()
        wg.Wait()
        fmt.Println("x:",x)
    }
    
    //区别
    var x int32
    var wg stnc.WaitGroup
    var mutex sync.Mutex
    func addMutex() {
        for i:=0;i<500;i++ {
            mutex.Lock()
            x = x+1
            mutex.Unlock()
        }
    }
    func add() {
        for i := 0;i<500;i++ {
            atomic.AddInt32(&x,1)
        }
        wg.Done()
    }
    func main() {
        start := time.Now().UnixNano()
        for i:= 0;i<500;i++ {
            wg.Add(1)
            go add()
            go addMutex()
        }
        wg.Wait()
        end := time.Now().UnixNano()
        cost := (end - start)/1000/1000
        fmt.Println("x:"x,"cost:",cost,"ms")
    }    //当线程开的很多时，原子性操作比互斥锁效率高

## 四、网络编程 ##

#### 　　TCP/IP协议介绍 ####

定义：提供已连接英特网的计算机进行通信的协议，TCP(传输控制协议)——应用程序时间通信，UDP(用户数据包协议)——应用程序之间的简单通信，IP(网络协议)——计算机之间的通信，DHCP(动态配置协议)——针对动态寻址

TCP协议：面向连接的协议，可靠传输，发送的数据保证对方能够收到，保证时序，发送的数据按发送的顺序到达，全双工模式，通过IP和端口进行计算机之间进行访问，域名和已知端口，http:80,http:443,ssl:22端口等

IP协议：用于计算机之间进行通信，是TCP/UDP协议的底层，IP是无连接的，负责把数据包路由到目的地

#### 　　Go快速实现TCP服务器 ####

服务端处理流程：监听端口，接收客户端的连接，创建goroutine，处理该连接

package main
    import (
        "fmt"
        "net"
    )
    func process(conn net.Conn) {
        defer conn,Close()
        for {
            var buf [128]byte
            n,err := conn.Read(buf[:])
            if err != nil {
                fmt.Printf("read from conn failed,err:%v",err)
                break
            }
            str := string(buf[:n])
            fmt.Printf("recv from lient,data:%v\n",str)
        }
    }
    func main() {
        listen,err := net.Listen("tcp","0.0.0.0:8080")
        if err != nil P
            fmt.Println("listen failed,err:",err(
            return
        }
        for {
            conn,err := listen.Accept()
            if err != nil {
                fmt.Printf("accept failed,err:%c\n",err)
                continue
            }
            go process(conn)
        }
    }

#### 　　Go快速实现TCP客户端 ####

客户端处理流程：建立服务端的连接，进行数据收发，关闭连接

//客户端
    import (
        "bufio"
        "fmt"
        "net"
        "os"
        "strings"
    )
    func main() {
        conn,err := net.Dial("tcp","0.0.0.0:8080")
        if err := nil {
            fmt.Println("dial failed,err:%v\n",err)
            return
        }
        reader := bufio.NewReader(os.Stdin)
        for {
            data,err := reader.ReadString("\n")
            if err != nil {
                fmt.Printf("read from console failed,err:%v\n",err)
                break
            }
            data = strings.TrimSpace(data)
            if data == "Q" {
                return
            }
            _,err != nil {
            if err != nil {
                fmt.Printf("write failed,err:%v\n",err)
                break
            }
        }
    }
    
    //访问网页
    func main() {
        conn,err := net.Dial("tcp","www.baidu.com:80")
        if err != nil {
            fmt.Printf{"dial failed,err:%v\n",err)
            return
        }
        data := "GET/HTTP/1.1\r\n"
        data += "HOST:www.baidu.com\r\n"
        data += "connection:close\r\n"
        data += "\r\n\r\n"
        _,err = io.WriteString(conn,data)
        if err != nil {
            fmt.printf("write string failed,err:%v\n",err)
            return
        }
        var buf [1024]byte
        for {
            n,err := connRead(buf[:])
            if err != nil || n == 0 {
                break
            }
            fmt.Println(string(buf[:]))
        }
    }    //得到html源码

#### 　　UDP协议介绍 ####

定义：用户数据报协议，无连接直接进行数据发送，不可靠没有时序，实时性比较好，通常用于视频世博相关领域

//server
    func main() {
        listen,err := net.ListenUDP("udp",&net.UDPAddr{
            IP: net.IPv4(0.0.0.),
            Port:8080,
        })
        if err != nil {
            fmt.Printf("listen failed,err:%v\n",err)
            return
        }
        for {
            var data [1024]byte
            count,addr,err := listen.ReadFromUDP(data[:])
            if err != nil {
                fmt.Printf("read udp failed,err:%v\n",err)
                continue
            }
            fmt.Printf("data:%s addr:%v count:%d\n",string(data[0:count]),addr,count)
            _,err = listen.WaiteToUDP([]byte("hello client"),addr)
            if err != nil {
                fmt.Printf("read udp failed,err:%v\n",err)
                continue
            }
        }
    }
    
    //client
    func main() {
        socket,err := net.DialUDP("udp4",nil,&net.UDPAddr{
            IP: net.IPv4(127.0.0.1),
            Port:8080,
        })
        if err != nil {
            fmt.Println("连接失败",err)
            return
        }
        defer socker.Close()
        senddata := []byte("hello server")
        _,err = socket.Write(senddata)
        if err != nil {
            fmt.Println("发送数据失败",err)
            return
        }
        data := make([]byte,4096)
        read,remoteAddr,err := socket.ReadFromUDP(data)
        if err != nil {
            fmt.Println("读取数据失败",err)
            return
        }
        fmt.Println(read,remoteAdd)
        fmt.Printf("%s\n",data)
    }

转载于:https://www.cnblogs.com/parallel-Y/p/11432988.html

[1648958-20190901170543136-599530989.png]: /images/20230601/61b42c211fd14c23b988e52b42ba3df3.png
[1648958-20190901170646976-710050573.png]: /images/20230601/bb2cb22e50bd485b9b197ed128bce75d.png