go案例-《GO开发知识笔记》

admin
admin
admin
0
文章
0
评论
2025-11-04 01:21:29 编程 来源：ZONE.CI 全球网 0 阅读模式
  • 1、go Http Post 发送文件流
  • 二、sync.Map
  • 三、路由器
  • 四、解析yaml文件
  • 五、mq
  • 六、map
  • 七、多路复用
  • 八、for循环
  • 九、Stringer接口
  • 十、抽象类实现
  • 十一、接口
  • 十二、多线程
  • 十三、websocket
  • 十四、协程池
  • 十五、go调用C语言
  • include
  • include
  • 十七、go调用tol物联网
  • https://github.com/crossoverJie/gorm-optimistic">十八、乐观锁 参考：https://github.com/crossoverJie/gorm-optimistic
  • 十九、并发条件下的有序输出
  • go.mod

    网址：https://gobea.cn/

    1、go Http Post 发送文件流

    1. package main
    2. import (
    3.     "net/http"
    4.     "net/url"
    5.     "fmt"
    6.     "io/ioutil"
    7.     _ "io"
    8.     "bytes"
    9. )
    11. func main()  {
    12.     postFile()
    13. }
    14. func post()  {
    15.         //这是一个Post 参数会被返回的地址
    16.         strinUrl:="http://localhost:8080/aaa"`这里写代码片`
    17.         resopne,err:=   http.PostForm(strinUrl,url.Values{"num":{"456"},"num1":{"123"}})
    19.         if err !=nil {
    20.         fmt.Println("err=",err)
    21.         }
    22.         defer func() {
    23.             resopne.Body.Close()
    24.             fmt.Println("finish")
    25.         }()
    27.         body,err:=ioutil.ReadAll(resopne.Body)
    28.         if err!=nil {
    29.             fmt.Println(" post err=",err)
    30.         }
    31.         fmt.Println(string(body))
    32. }
    34. func postFile(){
    35.     //这是一个Post 参数会被返回的地址
    36.     strinUrl:="http://localhost:8080/aaa"
    37.     byte,err:=ioutil.ReadFile("post.txt")
    38.     resopne,err  :=http.Post(strinUrl,"multipart/form-data",bytes.NewReader(byte))
    39.     if err !=nil {
    40.         fmt.Println("err=",err)
    41.     }
    42.     defer func() {
    43.         resopne.Body.Close()
    44.         fmt.Println("finish")
    45.     }()
    46.     body,err:=ioutil.ReadAll(resopne.Body)
    47.     if err!=nil {
    48.         fmt.Println(" post err=",err)
    49.     }
    50.     fmt.Println(string(body))
    51. }

    二、sync.Map

    1. package main
    3. import (
    4.     "fmt"
    5.     "sync"
    6. )
    8. type Class struct {
    9.     Students sync.Map
    10. }
    12. func handler(key, value interface{}) bool {
    13.     fmt.Printf("Name :%s %s\n", key, value)
    14.     return true
    15. }
    17. func main() {
    19.     class := &Class{}
    21.     //存储值
    22.     class.Students.Store("Zhao", "class 1")
    23.     class.Students.Store("Qian", "class 2")
    24.     class.Students.Store("Sun", "class 3")
    26.     //遍历，传入一个函数，遍历的时候函数返回false则停止遍历
    27.     class.Students.Range(handler)
    29.     //查询
    30.     if _, ok := class.Students.Load("Li"); !ok {
    31.         fmt.Println("-->Li not found")
    32.     }
    34.     //查询或者追加
    35.     _, loaded := class.Students.LoadOrStore("Li", "class 4")
    36.     if loaded {
    37.         fmt.Println("-->Load Li success")
    38.     } else {
    39.         fmt.Println("-->Store Li success")
    40.     }
    42.     //删除
    43.     class.Students.Delete("Sun")
    44.     fmt.Println("-->Delete Sun success")
    46.     //遍历
    47.     class.Students.Range(handler)
    49. }

    三、路由器

    1. package main
    3. import (
    4.     "fmt"
    5.     "net/http"
    6. )
    8. type MyMux struct {
    9. }
    11. func (p *MyMux) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    12.     if r.URL.Path == "/" {
    13.         sayhelloName(w, r)
    14.         return
    15.     }
    16.     http.NotFound(w, r)
    17.     return
    18. }
    20. func sayhelloName(w http.ResponseWriter, r *http.Request) {
    21.     fmt.Fprintf(w, "Hello myroute!")
    22. }
    24. func main() {
    25.     mux := &MyMux{}
    26.     http.ListenAndServe(":9090", mux)
    27. }

    四、解析yaml文件

    1. package main
    3. import (
    4.     "fmt"
    5.     "log"
    7.     "gopkg.in/yaml.v2"
    8. )
    10. var data = `
    11. blog: xiaorui.cc
    12. best_authors: ["fengyun","lee","park"]
    13. desc:
    14.   counter: 521
    15.   plist: [3, 4]
    16. `
    18. type T struct {
    19.     Blog    string
    20.     Authors []string `yaml:"best_authors,flow"`
    21.     Desc    struct {
    22.         Counter int   `yaml:"Counter"`
    23.         Plist   []int `yaml:",flow"`
    24.     }
    25. }
    27. func main() {
    28.     t := T{}
    29.     //把yaml形式的字符串解析成struct类型
    30.     err := yaml.Unmarshal([]byte(data), &t)
    31.     //修改struct里面的记录
    32.     t.Blog = "this is Blog"
    33.     t.Authors = append(t.Authors, "myself")
    34.     t.Desc.Counter = 99
    35.     fmt.Printf("--- t:\n%v\n\n", t)
    36.     //转换成yaml字符串类型
    37.     d, err := yaml.Marshal(&t)
    38.     if err != nil {
    39.         log.Fatalf("error: %v", err)
    40.     }
    41.     fmt.Printf("--- t dump:\n%s\n\n", string(d))
    42. }

    五、mq

    1. package main
    3. import (
    4.     "flag"
    5.     "log"
    6.     "strconv"
    7.     "time"
    9.     "github.com/nats-io/nats.go"
    10.     "github.com/pborman/uuid"
    11. )
    13. const (
    14.     //url   = "nats://192.168.3.125:4222"
    15.     url = nats.DefaultURL
    16. )
    18. var (
    19.     nc  *nats.Conn
    20.     err error
    21. )
    23. func init() {
    24.     // if nc, err = nats.Connect(url); checkErr(err) {
    25.     //     fmt.
    26.     // }
    27. }
    29. func main() {
    31. }
    33. //发送消息
    34. func sendMessage() {
    35.     var (
    36.         subj = flag.String("subj", "", "subject name")
    37.     )
    38.     flag.Parse()
    39.     log.Println(*subj)
    40.     startClient(*subj)
    41.     time.Sleep(time.Second)
    42. }
    44. func startClient(subj string) {
    45.     for i := 0; i < 1; i++ {
    46.         id := uuid.New()
    47.         log.Println(id)
    48.         nc.Publish(subj, []byte(id+" Sun "+strconv.Itoa(i)))
    49.         nc.Publish(subj, []byte(id+" Rain "+strconv.Itoa(i)))
    50.         nc.Publish(subj, []byte(id+" Fog "+strconv.Itoa(i)))
    51.         nc.Publish(subj, []byte(id+" Cloudy "+strconv.Itoa(i)))
    52.     }
    53. }
    55. //接收消息
    56. func receiveMessage() {
    57.     var (
    58.         servername = flag.String("servername", "y", "name for server")
    59.         queueGroup = flag.String("group", "", "group name for Subscribe")
    60.         subj       = flag.String("subj", "", "subject name")
    61.     )
    62.     flag.Parse()
    64.     log.Println(*servername, *queueGroup, *subj)
    65.     startService(*subj, *servername+" worker1", *queueGroup)
    66.     startService(*subj, *servername+" worker2", *queueGroup)
    67.     startService(*subj, *servername+" worker3", *queueGroup)
    68.     select {}
    69. }
    71. func startService(subj, name, queue string) {
    72.     go async(nc, subj, name, queue)
    73. }
    75. func async(nc *nats.Conn, subj, name, queue string) {
    76.     nc.QueueSubscribe(subj, queue, func(msg *nats.Msg) {
    77.         log.Println(name, "Received a message From Async : ", string(msg.Data))
    78.     })
    79. }
    81. func checkErr(err error) bool {
    82.     if err != nil {
    83.         log.Println(err)
    84.         return false
    85.     }
    86.     return true
    87. }
    1. package main
    3. import (
    4.     "fmt"
    5.     "log"
    7.     "github.com/streadway/amqp"
    8. )
    10. func failOnError(err error, msg string) {
    11.     if err != nil {
    12.         log.Fatalf("%s: %s", msg, err)
    13.         panic(fmt.Sprintf("%s: %s", msg, err))
    14.     }
    15. }
    17. func main() {
    18.     conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    19.     failOnError(err, "Failed to connect to RabbitMQ")
    20.     defer conn.Close()
    22.     ch, err := conn.Channel()
    23.     failOnError(err, "Failed to open a channel")
    24.     defer ch.Close()
    26.     failOnError(err, "Failed to declare a queue")
    28.     err = ch.Qos(
    29.         1,     // prefetch count
    30.         0,     // prefetch size
    31.         false, //global
    32.     )
    33.     failOnError(err, "Failed to set Qos")
    35.     //发送消息
    36.     funcSend(ch)
    37.     //收消息
    38.     funcRecv(ch)
    40. }
    42. //生产者
    43. func funcSend(ch *amqp.Channel) {
    44.     q, err := ch.QueueDeclare(
    45.         "queue", //name
    46.         false,   //durables
    47.         false,   //delete when unused
    48.         false,   //exclusive
    49.         false,   //no wait
    50.         nil,     //args
    51.     )
    52.     failOnError(err, "Failed to declare a queue")
    53.     body := "Hello World!"
    54.     error := ch.Publish(
    55.         "",     // exchange
    56.         q.Name, // routing key
    57.         false,  // mandatory
    58.         false,  // immediate
    59.         amqp.Publishing{
    60.             ContentType: "text/plain",
    61.             Body:        []byte(body),
    62.         })
    63.     log.Printf("[x] Sent %s", body)
    64.     failOnError(error, "Failed to publish a message")
    65. }
    67. //消费者
    68. func funcRecv(ch *amqp.Channel) {
    69.     q, err := ch.QueueDeclare(
    70.         "queue", // name
    71.         false,   // durable
    72.         false,   // delete when unused
    73.         false,   // exclusive
    74.         false,   // no-wait
    75.         nil,     // arguments
    76.     )
    77.     failOnError(err, "Failed to declare a queue")
    78.     msgs, err := ch.Consume(
    79.         q.Name, // queue
    80.         "",     // consumer
    81.         true,   // auto-ack
    82.         false,  // exclusive
    83.         false,  // no-local
    84.         false,  // no-wait
    85.         nil,    // args
    86.     )
    87.     failOnError(err, "Failed to register a consumer")
    88.     forever := make(chan bool)
    89.     go func() {
    90.         for d := range msgs {
    91.             log.Printf("Received a message: %s", d.Body)
    92.         }
    93.     }()
    94.     log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
    95.     <-forever
    96. }

    六、map

    1. package main
    3. import "fmt"
    4. import "sync"
    5. import "time"
    7. func main(){
    8.   var mTest sync.Map
    9.   go SyncTest(mTest)
    10.   go SyncTest(mTest)
    11.   time.Sleep(time.Second * 20)
    12. }
    14. func SyncTest(mTest sync.Map){
    15.     for j := 0; j < 1000; j++{
    16.         fmt.Println(mTest.Load("A"))//Load 读取
    17.         mTest.Store("A", "a") //Store 存储
    18.         fmt.Println(mTest.Load("A"))
    19.         mTest.Delete("A") //Delete 删除
    20.         fmt.Println(mTest.Load("A"))
    21.         mTest.LoadOrStore("B","b") //LoadOrStore 读取失败则存储
    22.         fmt.Println(mTest.Load("B"))
    23.     }    
    24. }
    26. func funcMap() {
    27.     // 先声明map
    28.     var m1 map[string]string
    29.     // 再使用make函数创建一个非nil的map，nil map不能赋值
    30.     m1 = make(map[string]string)
    31.     // 最后给已声明的map赋值
    32.     m1["a"] = "aa"
    33.     m1["b"] = "bb"
    35.     // 直接创建
    36.     m2 := make(map[string]string)
    37.     // 然后赋值
    38.     m2["a"] = "aa"
    39.     m2["b"] = "bb"
    41.     // 初始化 + 赋值一体化
    42.     m3 := map[string]string{
    43.         "a": "aa",
    44.         "b": "bb",
    45.     }
    47.     fmt.Println(m3)
    49.     // ==========================================
    50.     // 查找键值是否存在
    51.     if v, ok := m1["a"]; ok {
    52.         fmt.Println(v)
    53.     } else {
    54.         fmt.Println("Key Not Found")
    55.     }
    57.     // 遍历map
    58.     for k, v := range m1 {
    59.         fmt.Println(k, v)
    60.     }
    62. }

    七、多路复用

    1. package main
    3. import (
    4.     "fmt"
    5.     "net/http"
    6. )
    8. // 设置多个处理器函数
    9. func handler1(w http.ResponseWriter, r *http.Request) {
    10.     fmt.Fprintf(w, "1 欢迎访问 www.ydook.com !")
    11. }
    13. func handler2(w http.ResponseWriter, r *http.Request) {
    14.     fmt.Fprintf(w, "2 欢迎访问 www.ydook.com !")
    15. }
    17. func handler3(w http.ResponseWriter, r *http.Request) {
    18.     fmt.Fprintf(w, "3 欢迎访问 www.ydook.com !")
    19. }
    21. func main() {
    22.     // 设置多路复用处理函数
    23.     mux := http.NewServeMux()
    25.     mux.HandleFunc("/h1", handler1)
    26.     mux.HandleFunc("/h2", handler2)
    27.     mux.HandleFunc("/h3", handler3)
    29.     // 设置服务器
    30.     server := &http.Server{
    31.         Addr:    "127.0.0.1:8000",
    32.         Handler: mux,
    33.     }
    35.     // 设置服务器监听请求端口
    36.     server.ListenAndServe()
    37. }

    八、for循环

    1. package main
    2. import (
    3.     "fmt"
    4.     "time"
    5. )
    6. func main()  {
    7.     str := []string{"I","am","Sergey"}
    8.     for _,v := range str{
    9.         go func() {
    10.             fmt.Println(v)
    11.         }()
    12.     }
    13.     time.Sleep(3 * time.Second)
    14. }
    1. package main
    2. import (
    3.     "fmt"
    4.     "time"
    5. )
    6. func main()  {
    7.     str := []string{"I","am","Sergey"}
    8.     for _,v := range str{
    9.         go func(v string) {
    10.             fmt.Println(v)
    11.         }(v)
    12.     }
    13.     time.Sleep(3 * time.Second)
    14. }
    1. package main
    3. import "fmt"
    5. func main() {
    6.     slice := []int{0, 1, 2, 3}
    7.     myMap := make(map[int]*int)
    9.     for index , value := range slice {
    10.         myMap[index] = &value
    11.     }
    12.     prtMap(myMap)
    13. }
    15. func prtMap(myMap map[int]*int) {
    16.     for key, value := range myMap {
    17.         fmt.Printf("map[%v]=%v\n", key, *value)
    18.     }
    19. }
    1. package main
    3. import "fmt"
    5. func main() {
    6.     slice := []int{0, 1, 2, 3}
    7.     myMap := make(map[int]*int)
    9.     for index , value := range slice {
    10.         v := value
    11.         myMap[index] = &v
    12.     }
    13.     prtMap(myMap)
    14. }
    16. func prtMap(myMap map[int]*int) {
    17.     for key, value := range myMap {
    18.         fmt.Printf("map[%v]=%v\n", key, *value)
    19.     }
    20. }

    九、Stringer接口

    1. package main
    3. import "fmt"
    5. type Person struct {
    6.     Name string
    7.     Age  int
    8. }
    10. func (p Person) String() string {
    11.     return fmt.Sprintf("%v (%v years)", p.Name, p.Age)
    12. }
    14. func main() {
    15.     a := Person{"Arthur Dent", 42}
    16.     z := Person{"Zaphod Beeblebrox", 9001}
    17.     fmt.Println(a, z)
    18. }

    十、抽象类实现

    1. type IPeople interface {
    2.     GetName() string
    3.     SetName(string)
    4.     GetAge() int
    5.     SetAge(int)
    6.     Run()
    7. }
    9. type AbstractPeople struct {
    10.     IPeople 
    11.     name string
    12.     age int
    13. }
    15. func (a AbstractPeople) GetName() string {
    16.     return a.name
    17. }
    19. func (a *AbstractPeople) SetName(newName string) {
    20.     a.name = newName
    21. }
    23. func (a AbstractPeople) GetAge() int {
    24.     return a.age
    25. }
    27. func (a *AbstractPeople) SetAge(newAge int) {
    28.     a.age = newAge
    29. }
    31. // user.
    32. func UsePeople(p IPeople) {
    33.     fmt.Println(p.GetName())
    34.     fmt.Println(p.GetAge())
    35.     p.SetName("Alice")
    36.     p.SetAge(p.GetAge() + 1)
    37.     p.Run()
    38. }

    十一、接口

    1. package main
    2. import (
    3.     "fmt"
    4. )
    5. // 定义一个数据写入器
    6. type DataWriter interface {
    7.     WriteData(data interface{}) error
    8. }
    9. // 定义文件结构，用于实现DataWriter
    10. type file struct {
    11. }
    12. // 实现DataWriter接口的WriteData方法
    13. func (d *file) WriteData(data interface{}) error {
    14.     // 模拟写入数据
    15.     fmt.Println("WriteData:", data)
    16.     return nil
    17. }
    18. func main() {
    19.     // 实例化file
    20.     f := new(file)
    21.     // 声明一个DataWriter的接口
    22.     var writer DataWriter
    23.     // 将接口赋值f，也就是*file类型
    24.     writer = f
    25.     // 使用DataWriter接口进行数据写入
    26.     writer.WriteData("data")
    27. }

    十二、多线程

    1. package main
    2. import (
    3.     "runtime"
    4.     "sync"
    5. )
    6. var (
    7.     cpunum = runtime.NumCPU()-1
    8. )
    9. func main(){
    10.     ch := make(chan string)
    11.     runtime.GOMAXPROCS(cpunum)
    12.     wg := sync.WaitGroup{}
    14.     for i := 0; i < cpunum; i++{
    15.         go WgReadLogs(ch, &wg)
    16.     }
    18.     wg.Add(2)
    19.     ch <- "./health/stat1.rec"
    20.     ch <- "./report/stat2.rec"
    21.     wg.Wait()
    22.     close(ch)
    23. }
    24. func WgReadLogs(ch chan string,wg *sync.WaitGroup){
    25.     for true{
    26.     tmp,ok := <-ch
    27.     if !ok{
    28.         break
    29.     }
    30.     ReadLogs(tmp)
    31.     wg.Done()
    32.     }
    33. }
    34. func ReadLogs(logname string){
    35.     fmt.Println(logname)
    36. }

    十三、websocket

    1. package main
    3. import (
    4.     "bytes"
    5.     "fmt"
    6.     "log"
    7.     "net/http"
    8.     "unsafe"
    10.     "golang.org/x/net/websocket"
    11. )
    13. func BytePtrToString(p *byte) string {
    14.     bufs := bytes.NewBufferString("")
    15.     for ps := p; *ps != byte(0); ps = (*byte)(unsafe.Pointer(1 + (uintptr)(unsafe.Pointer(ps)))) {
    16.         bufs.WriteByte(*ps)
    17.     }
    18.     return bufs.String()
    19. }
    21. func main() {
    22.     fmt.Println("websocket地址: ws://127.0.0.1:8080/runtime")
    23.     http.Handle("/runtime", websocket.Handler(echo))
    24.     if err := http.ListenAndServe(":8080", nil); err != nil {
    25.         log.Fatal(err)
    26.     }
    27. }
    29. //测试地址： http://coolaf.com/tool/chattest  首字母大写是公有的  Echo，首字母小写是私有的 echo
    30. func echo(w *websocket.Conn) {
    31.     var error error
    32.     for {
    33.         //只支持string类型
    34.         var reply string
    35.         if error = websocket.Message.Receive(w, &reply); error != nil {
    36.             log.Println("websocket出现异常", error)
    37.             break
    38.         }
    39.         fmt.Println("收到客户端消息:" + reply)
    40.         msg := reply + ", 我是服务端"
    41.         fmt.Println("发送客户端消息:" + msg)
    42.         if error = websocket.Message.Send(w, msg); error != nil {
    43.             log.Println("websocket出现异常", error)
    44.             break
    45.         }
    46.     }
    47. }

    十四、协程池

    1. package main
    3. import (
    4.     "fmt"
    5.     "time"
    6. )
    8. // 任务的属性应该是一个业务函数
    9. type Task struct {
    10.     f func() error // 函数名f, 无参，返回值为error
    11. }
    13. // 创建Task任务
    14. func NewTask(arg_f func() error) *Task {
    15.     task := Task{
    16.         f: arg_f,
    17.     }
    18.     return &task
    19. }
    21. // Task绑定业务方法
    22. func (task *Task) Execute() {
    23.     task.f() // 调用任务中已经绑定好的业务方法
    24. }
    26. // ------------------------------------------------
    27. type Pool struct {
    28.     EntryChannel chan *Task // 对外的Task入口
    29.     JobsChannel  chan *Task // 内部的Task队列
    30.     workerNum    int        // 协程池中最大的woker数量
    31. }
    33. // 创建Pool
    34. func NewPool(cap int) *Pool {
    35.     pool := Pool{
    36.         EntryChannel: make(chan *Task),
    37.         JobsChannel:  make(chan *Task),
    38.         workerNum:    cap,
    39.     }
    40.     return &pool
    41. }
    43. // Pool绑定干活的方法
    44. func (pool *Pool) worker(workID int) {
    45.     // worker工作 ： 永久从JobsChannel取任务 然后执行任务
    46.     for task := range pool.JobsChannel {
    47.         task.Execute()
    48.         fmt.Println("work ID ", workID, " has executed")
    49.     }
    50. }
    52. // Pool绑定协程池工作方法
    53. func (pool *Pool) run() {
    54.     // 定义worker数量
    55.     for i := 0; i < pool.workerNum; i++ {
    56.         go pool.worker(i)
    57.     }
    59.     // 从EntryChannel去任务，发送给JobsChannel
    60.     for task := range pool.EntryChannel {
    61.         pool.JobsChannel <- task // 添加task优先级排序逻辑
    62.     }
    63. }
    65. // ------------------------------------------------
    66. func main() {
    67.     // 创建一些任务
    68.     task := NewTask(func() error { // 匿名函数
    69.         fmt.Println(time.Now())
    70.         return nil
    71.     })
    73.     // 创建协程池
    74.     pool := NewPool(4)
    76.     // 创建多任务，抛给协程池
    77.     go func() { // 开启新的协程，防止阻塞
    78.         for {
    79.             pool.EntryChannel <- task
    80.         }
    81.     }()
    83.     // 启动协程池
    84.     pool.run()
    85. }
    1. package main
    3. import (
    4.     "fmt"
    5.     "sync"
    6.     "time"
    7. )
    9. // 每个协程都会运行该函数。
    10. // 注意，WaitGroup 必须通过指针传递给函数。
    11. func worker(id int, wg *sync.WaitGroup) {
    12.     fmt.Printf("Worker %d starting\n", id)
    14.     // 睡眠一秒钟，以此来模拟耗时的任务。
    15.     time.Sleep(time.Second)
    16.     fmt.Printf("Worker %d done\n", id)
    18.     // 通知 WaitGroup ，当前协程的工作已经完成。
    19.     wg.Done()
    20. }
    22. func main() {
    24.     // 这个 WaitGroup 被用于等待该函数开启的所有协程。
    25.     var wg sync.WaitGroup
    27.     // 开启几个协程，并为其递增 WaitGroup 的计数器。
    28.     for i := 1; i <= 5; i++ {
    29.         wg.Add(1)
    30.         go worker(i, &wg)
    31.     }
    33.     // 阻塞，直到 WaitGroup 计数器恢复为 0，即所有协程的工作都已经完成。
    34.     wg.Wait()
    35. }
    1. package main
    3. import (
    4.     "fmt"
    5.     "time"
    6. )
    8. // 这里是 worker，我们将并发执行多个 worker。
    9. // worker 将从 `jobs` 通道接收任务，并且通过 `results` 发送对应的结果。
    10. // 我们将让每个任务间隔 1s 来模仿一个耗时的任务。
    11. func worker(id int, jobs <-chan int, results chan<- int) {
    12.     for j := range jobs {
    13.         fmt.Println("worker", id, "processing job", j)
    14.         time.Sleep(time.Second)
    15.         results <- j * 2
    16.     }
    17. }
    19. func main() {
    21.     // 为了使用 worker 线程池并且收集他们的结果，我们需要 2 个通道。
    22.     jobs := make(chan int, 100)
    23.     results := make(chan int, 100)
    25.     // 这里启动了 3 个 worker，初始是阻塞的，因为还没有传递任务。
    26.     for w := 1; w <= 3; w++ {
    27.         go worker(w, jobs, results)
    28.     }
    30.     // 这里我们发送 9 个 `jobs`，然后 `close` 这些通道
    31.     // 来表示这些就是所有的任务了。
    32.     for j := 1; j <= 9; j++ {
    33.         jobs <- j
    34.     }
    35.     close(jobs)
    37.     // 最后，我们收集所有这些任务的返回值。
    38.     for a := 1; a <= 9; a++ {
    39.         <-results
    40.     }
    41. }
    1. package main
    3. import (
    4.     "context"
    5.     "fmt"
    6.     "strconv"
    7.     "sync/atomic"
    9.     pool "github.com/jolestar/go-commons-pool"
    10. )
    12. func Example_simple() {
    13.     type myPoolObject struct {
    14.         s string
    15.     }
    16.     v := uint64(0)
    17.     factory := pool.NewPooledObjectFactorySimple(
    18.         func(context.Context) (interface{}, error) {
    19.             return &myPoolObject{
    20.                     s: strconv.FormatUint(atomic.AddUint64(&v, 1), 10),
    21.                 },
    22.                 nil
    23.         })
    24.     ctx := context.Background()
    25.     p := pool.NewObjectPoolWithDefaultConfig(ctx, factory)
    26.     obj, err := p.BorrowObject(ctx)
    27.     if err != nil {
    28.         panic(err)
    29.     }
    30.     o := obj.(*myPoolObject)
    31.     fmt.Println(o.s)
    32.     err = p.ReturnObject(ctx, obj)
    33.     if err != nil {
    34.         panic(err)
    35.     }
    36.     // Output: 1
    37. }
    1. package main
    3. import (
    4.     "io/ioutil"
    5.     "net/http"
    6.     "runtime"
    8.     "github.com/Jeffail/tunny"
    9. )
    11. func main() {
    12.     numCPUs := runtime.NumCPU()
    14.     pool := tunny.NewFunc(numCPUs, func(payload interface{}) interface{} {
    15.         var result []byte
    17.         // TODO: Something CPU heavy with payload
    19.         return result
    20.     })
    21.     defer pool.Close()
    23.     http.HandleFunc("/work", func(w http.ResponseWriter, r *http.Request) {
    24.         input, err := ioutil.ReadAll(r.Body)
    25.         if err != nil {
    26.             http.Error(w, "Internal error", http.StatusInternalServerError)
    27.         }
    28.         defer r.Body.Close()
    30.         // Funnel this work into our pool. This call is synchronous and will
    31.         // block until the job is completed.
    32.         result := pool.Process(input)
    34.         w.Write(result.([]byte))
    35.     })
    37.     http.ListenAndServe(":8080", nil)
    38. }
    1. package main
    3. import (
    4.     "errors"
    5.     "fmt"
    6.     "log"
    7.     "sync"
    8.     "sync/atomic"
    9.     "time"
    10. )
    12. var (
    13.     // ErrInvalidPoolCap return if pool size <= 0
    14.     ErrInvalidPoolCap = errors.New("invalid pool cap")
    15.     // ErrPoolAlreadyClosed put task but pool already closed
    16.     ErrPoolAlreadyClosed = errors.New("pool already closed")
    17. )
    19. const (
    20.     // RUNNING pool is running
    21.     RUNNING = 1
    22.     // STOPED pool is stoped
    23.     STOPED = 0
    24. )
    26. // Task task to-do
    27. type Task struct {
    28.     Handler func(v ...interface{})
    29.     Params  []interface{}
    30. }
    32. // Pool task pool
    33. type Pool struct {
    34.     capacity       uint64
    35.     runningWorkers uint64
    36.     state          int64
    37.     taskC          chan *Task
    38.     PanicHandler   func(interface{})
    39.     sync.Mutex
    40. }
    42. // NewPool init pool
    43. func NewPool(capacity uint64) (*Pool, error) {
    44.     if capacity <= 0 {
    45.         return nil, ErrInvalidPoolCap
    46.     }
    47.     return &Pool{
    48.         capacity: capacity,
    49.         state:    RUNNING,
    50.         taskC:    make(chan *Task, capacity),
    51.     }, nil
    52. }
    54. // GetCap get capacity
    55. func (p *Pool) GetCap() uint64 {
    56.     return p.capacity
    57. }
    59. // GetRunningWorkers get running workers
    60. func (p *Pool) GetRunningWorkers() uint64 {
    61.     return atomic.LoadUint64(&p.runningWorkers)
    62. }
    64. func (p *Pool) incRunning() {
    65.     atomic.AddUint64(&p.runningWorkers, 1)
    66. }
    68. func (p *Pool) decRunning() {
    69.     atomic.AddUint64(&p.runningWorkers, ^uint64(0))
    70. }
    72. // Put put a task to pool
    73. func (p *Pool) Put(task *Task) error {
    75.     if p.getState() == STOPED {
    76.         return ErrPoolAlreadyClosed
    77.     }
    79.     // safe run worker
    80.     p.Lock()
    81.     if p.GetRunningWorkers() < p.GetCap() {
    82.         p.run()
    83.     }
    84.     p.Unlock()
    86.     // send task safe
    87.     p.Lock()
    88.     if p.state == RUNNING {
    89.         p.taskC <- task
    90.     }
    91.     p.Unlock()
    93.     return nil
    94. }
    96. func (p *Pool) run() {
    97.     p.incRunning()
    99.     go func() {
    100.         defer func() {
    101.             p.decRunning()
    102.             if r := recover(); r != nil {
    103.                 if p.PanicHandler != nil {
    104.                     p.PanicHandler(r)
    105.                 } else {
    106.                     log.Printf("Worker panic: %s\n", r)
    107.                 }
    108.             }
    109.         }()
    111.         for {
    112.             select {
    113.             case task, ok := <-p.taskC:
    114.                 if !ok {
    115.                     return
    116.                 }
    117.                 task.Handler(task.Params...)
    118.             }
    119.         }
    120.     }()
    121. }
    123. func (p *Pool) getState() int64 {
    124.     p.Lock()
    125.     defer p.Unlock()
    127.     return p.state
    128. }
    130. func (p *Pool) setState(state int64) {
    131.     p.Lock()
    132.     defer p.Unlock()
    134.     p.state = state
    135. }
    137. // close safe
    138. func (p *Pool) close() {
    139.     p.Lock()
    140.     defer p.Unlock()
    142.     close(p.taskC)
    143. }
    145. // Close close pool graceful
    146. func (p *Pool) Close() {
    148.     if p.getState() == STOPED {
    149.         return
    150.     }
    152.     p.setState(STOPED) // stop put task
    154.     for len(p.taskC) > 0 { // wait all task be consumed
    155.         time.Sleep(1e6) // reduce CPU load
    156.     }
    158.     p.close()
    159. }
    161. //https://github.com/wazsmwazsm/mortar
    162. func main() {
    163.     // 创建容量为 10 的任务池
    164.     pool, err := NewPool(10)
    165.     if err != nil {
    166.         panic(err)
    167.     }
    169.     wg := new(sync.WaitGroup)
    171.     for i := 0; i < 1000; i++ {
    172.         wg.Add(1)
    173.         // 创建任务
    174.         task := &Task{
    175.             Handler: func(v ...interface{}) {
    176.                 wg.Done()
    177.                 fmt.Println(v)
    178.             },
    179.         }
    180.         // 添加任务函数的参数
    181.         task.Params = []interface{}{i, i * 2, "hello"}
    182.         // 将任务放入任务池
    183.         pool.Put(task)
    184.     }
    185.     wg.Add(1)
    186.     // 再创建一个任务
    187.     pool.Put(&Task{
    188.         Handler: func(v ...interface{}) {
    189.             wg.Done()
    190.             fmt.Println(v)
    191.         },
    192.         Params: []interface{}{"hi!"}, // 也可以在创建任务时设置参数
    193.     })
    195.     wg.Wait()
    197.     // 安全关闭任务池（保证已加入池中的任务被消费完）
    198.     pool.Close()
    199.     // 如果任务池已经关闭, Put() 方法会返回 ErrPoolAlreadyClosed 错误
    200.     err = pool.Put(&Task{
    201.         Handler: func(v ...interface{}) {},
    202.     })
    203.     if err != nil {
    204.         fmt.Println(err) // print: pool already closed
    205.     }
    206. }

    十五、go调用C语言

    注意： rpc（远程函数调用）提供client端通过网络调用远程server端的函数的服务。 rpc-server端需要提供较高的吞吐能力，支持较大的并发连接。

    • epoll监听多个连接fd，实现IO复用

      1.   1）epoll没有最大并发连接的限制，上限是最大可以打开文件的数目，一般远大于2048<br />        2）epoll效率高，只管活跃的连接，而跟连接总数无关<br />        3）使用共享内存，省去了内存拷贝

    • 生产者&消费者模式对请求任务进行管理

      1.   一个IO-thread线程维护连接池中的连接，连接有3种状态：NO_USED、READY、BUSY。<br />        1）NO_USED状态，fd位置空闲，连接位尚未使用<br />        2）READY状态，已经与client端连接连接的fd，等待请求事件<br />        3）BUSY状态，已经收到请求事件，将fd放入任务队列，等待work-thread处理

      ```go package main

    /*

    include

    include

    void say(){ printf(“hello world\n”); } */ import “C”

    import ( “fmt” “net/http” “runtime” “syscall” “time”

    1. "github.com/gin-gonic/gin"

    )

    func main() { runtime.LockOSThread() router := gin.Default() router.GET(“/ping”, func(c gin.Context) { fmt.Println(“worker”, syscall.Getegid()) c.JSON(200, gin.H{ “message”: “pong”, }) }) s := &http.Server{ Addr: “:18282”, Handler: router, ReadTimeout: 600 time.Second, WriteTimeout: 600 * time.Second, MaxHeaderBytes: 1 << 20, } fmt.Println(“listen:”, syscall.Getegid()) s.ListenAndServe() fmt.Println(“main”, C.say()) }

    func testMain() { runtime.GOMAXPROCS(runtime.NumCPU()) ch1 := make(chan bool) ch2 := make(chan bool) fmt.Println(“main”, C.say()) go func() { runtime.LockOSThread() fmt.Println(“locked”, C.say()) go func() { fmt.Println(“locked child”, C.say()) ch1 <- true }() ch2 <- true }() <-ch1 <-ch2 }

    1. <a name="AlQ0C"></a>
    2. # 十六、go RPC
    3. ```go
    4. package main
    6. import (
    7.     "context"
    8.     "log"
    9.     "net"
    10.     "os"
    11.     "time"
    13.     "google.golang.org/grpc"
    14.     pb "google.golang.org/grpc/examples/helloworld/helloworld"
    15.     "google.golang.org/grpc/reflection"
    16. )
    18. func main() {
    19.     println(f(1))
    20. }
    22. const (
    23.     port        = ":50051"
    24.     address     = "localhost:50051"
    25.     defaultName = "world"
    26. )
    28. type server struct{} //服务对象
    30. // SayHello 实现服务的接口 在proto中定义的所有服务都是接口
    31. func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
    32.     return &pb.HelloReply{Message: "Hello " + in.Name}, nil
    33. }
    35. func serverMain() {
    36.     lis, err := net.Listen("tcp", port)
    37.     if err != nil {
    38.         log.Fatalf("failed to listen: %v", err)
    39.     }
    40.     s := grpc.NewServer() //起一个服务
    41.     pb.RegisterGreeterServer(s, &server{})
    42.     // 注册反射服务 这个服务是CLI使用的 跟服务本身没有关系
    43.     reflection.Register(s)
    44.     if err := s.Serve(lis); err != nil {
    45.         log.Fatalf("failed to serve: %v", err)
    46.     }
    47. }
    49. func clientMain() {
    50.     //建立链接
    51.     conn, err := grpc.Dial(address, grpc.WithInsecure())
    52.     if err != nil {
    53.         log.Fatalf("did not connect: %v", err)
    54.     }
    55.     defer conn.Close()
    56.     c := pb.NewGreeterClient(conn)
    58.     // Contact the server and print out its response.
    59.     name := defaultName
    60.     if len(os.Args) > 1 {
    61.         name = os.Args[1]
    62.     }
    63.     // 1秒的上下文
    64.     ctx, cancel := context.WithTimeout(context.Background(), time.Second)
    65.     defer cancel()
    66.     r, err := c.SayHello(ctx, &pb.HelloRequest{Name: name})
    67.     if err != nil {
    68.         log.Fatalf("could not greet: %v", err)
    69.     }
    70.     log.Printf("Greeting: %s", r.Message)
    71. }
    73. func f(x int) (_, __ int) {
    74.     _, __ = x, x
    75.     return
    76. }

    十七、go调用tol物联网

    1. package main
    3. import (
    4.     "fmt"
    5.     "sync"
    6.     "time"
    8.     mqtt "github.com/eclipse/paho.mqtt.golang"
    9. )
    11. const broker = "tcp://你的服务器.mqtt.iot.gz.baidubce.com:1883"
    12. const username = "你的服务器/client"
    13. const password = "UJq**密码**msq"
    14. const ClientID = "随意"
    16. //message的回调
    17. var onMessage mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
    18.     fmt.Printf("[%s] -> %s\n", msg.Topic(), msg.Payload())
    19. }
    21. var wg sync.WaitGroup
    22. var client mqtt.Client
    24. func main() {
    25.     //连接MQTT服务器
    26.     mqttConnect()
    27.     defer client.Disconnect(250) //注册销毁
    28.     wg.Add(1)
    29.     go mqttSubScribe("test1")
    30.     wg.Add(1)
    31.     go testPublish()
    32.     wg.Wait()
    33. }
    34. func mqttConnect() {
    35.     //配置
    36.     clinetOptions := mqtt.NewClientOptions().AddBroker(broker).SetUsername(username).SetPassword(password)
    37.     clinetOptions.SetClientID(ClientID)
    38.     clinetOptions.SetConnectTimeout(time.Duration(60) * time.Second)
    39.     //连接
    40.     client = mqtt.NewClient(clinetOptions)
    41.     //客户端连接判断
    42.     if token := client.Connect(); token.WaitTimeout(time.Duration(60)*time.Second) && token.Wait() && token.Error() != nil {
    43.         panic(token.Error())
    44.     }
    45. }
    47. func mqttSubScribe(topic string) {
    48.     defer wg.Done()
    49.     for {
    50.         token := client.Subscribe(topic, 1, onMessage)
    51.         token.Wait()
    52.     }
    53. }
    55. //测试 3秒发送一次，然后自己接收
    56. func testPublish() {
    57.     defer wg.Done()
    58.     for {
    59.         client.Publish("test1", 1, false, "TEST")
    60.         time.Sleep(time.Duration(3) * time.Second)
    61.     }
    62. }
    1. package main
    3. import (
    4.     "crypto/tls"
    5.     "crypto/x509"
    6.     "fmt"
    7.     "io/ioutil"
    8.     "log"
    9.     "time"
    11.     mqtt "github.com/eclipse/paho.mqtt.golang"
    12. )
    14. var messagePubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
    15.     fmt.Printf("Received message: %s from topic: %s\n", msg.Payload(), msg.Topic())
    16. }
    18. var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {
    19.     fmt.Println("Connected")
    20. }
    22. var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {
    23.     fmt.Printf("Connect lost: %v", err)
    24. }
    26. //https://cloud.emqx.cn/console/deployments/0?oper=new
    27. // ClientOptions：用于设置 broker，端口，客户端 id ，用户名密码等选项
    28. // messagePubHandler：全局 MQTT pub 消息处理
    29. // connectHandler：连接的回调
    30. // connectLostHandler：连接丢失的回调
    31. func main() {
    32.     var broker = "broker.emqx.io"
    33.     var port = 1883
    34.     opts := mqtt.NewClientOptions()
    35.     opts.AddBroker(fmt.Sprintf("tcp://%s:%d", broker, port))
    36.     opts.SetClientID("go_mqtt_client")
    37.     opts.SetUsername("emqx")
    38.     opts.SetPassword("public")
    39.     opts.SetDefaultPublishHandler(messagePubHandler)
    40.     opts.OnConnect = connectHandler
    41.     opts.OnConnectionLost = connectLostHandler
    42.     client := mqtt.NewClient(opts)
    43.     if token := client.Connect(); token.Wait() && token.Error() != nil {
    44.         panic(token.Error())
    45.     }
    46.     sub(client)
    47.     publish(client)
    48.     client.Disconnect(250)
    49. }
    51. //发布消息
    52. func publish(client mqtt.Client) {
    53.     num := 10
    54.     for i := 0; i < num; i++ {
    55.         text := fmt.Sprintf("Message %d", i)
    56.         token := client.Publish("topic/test", 0, false, text)
    57.         token.Wait()
    58.         time.Sleep(time.Second)
    59.     }
    60. }
    62. //订阅
    63. func sub(client mqtt.Client) {
    64.     topic := "topic/test"
    65.     token := client.Subscribe(topic, 1, nil)
    66.     token.Wait()
    67.     fmt.Printf("Subscribed to topic: %s", topic)
    68. }
    70. //如果想使用 TLS 连接，可以如下设置：
    71. func NewTlsConfig() *tls.Config {
    72.     certpool := x509.NewCertPool()
    73.     ca, err := ioutil.ReadFile("ca.pem")
    74.     if err != nil {
    75.         log.Fatalln(err.Error())
    76.     }
    77.     certpool.AppendCertsFromPEM(ca)
    78.     // Import client certificate/key pair
    79.     clientKeyPair, err := tls.LoadX509KeyPair("client-crt.pem", "client-key.pem")
    80.     if err != nil {
    81.         panic(err)
    82.     }
    83.     return &tls.Config{
    84.         RootCAs:            certpool,
    85.         ClientAuth:         tls.NoClientCert,
    86.         ClientCAs:          nil,
    87.         InsecureSkipVerify: true,
    88.         Certificates:       []tls.Certificate{clientKeyPair},
    89.     }
    90. }
    92. //如果不设置客户端证书，可以如下设置：
    93. func NewTlsConfigs() *tls.Config {
    94.     certpool := x509.NewCertPool()
    95.     ca, err := ioutil.ReadFile("ca.pem")
    96.     if err != nil {
    97.         log.Fatalln(err.Error())
    98.     }
    99.     certpool.AppendCertsFromPEM(ca)
    100.     return &tls.Config{
    101.         RootCAs:            certpool,
    102.         ClientAuth:         tls.NoClientCert,
    103.         ClientCAs:          nil,
    104.         InsecureSkipVerify: true,
    105.     }
    106. }

    十八、乐观锁 参考：https://github.com/crossoverJie/gorm-optimistic

    Quick start

    1. func BenchmarkUpdateWithOptimistic(b *testing.B) {
    2. dsn := "root:abc123@/test?charset=utf8&parseTime=True&loc=Local"
    3. db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{})
    4. if err != nil {
    5.      fmt.Println(err)
    6.      return
    7.  }
    8. b.RunParallel(func(pb *testing.PB) {
    9.      var out Optimistic
    10.      db.First(&out, Optimistic{Id: 1})
    11.      out.Amount = out.Amount + 10
    12.      err = UpdateWithOptimistic(db, &out, func(model Lock) Lock {
    13.    bizModel := model.(*Optimistic)
    14.    bizModel.Amount = bizModel.Amount + 10
    15.    return bizModel
    16.      }, 5, 0)
    17.      if err != nil {
    18.    fmt.Println(err)
    19.      }
    20.  })
    21. }

    Model

    1. type Optimistic struct {
    2.     Id      int64   `gorm:"column:id;primary_key;AUTO_INCREMENT" json:"id"`
    3.     UserId  string  `gorm:"column:user_id;default:0;NOT NULL" json:"user_id"` // 用户ID
    4.     Amount  float32 `gorm:"column:amount;NOT NULL" json:"amount"`             // 金额
    5.     Version int64   `gorm:"column:version;default:0;NOT NULL" json:"version"` // 版本
    6. }
    8. func (o *Optimistic) TableName() string {
    9.     return "t_optimistic"
    10. }
    12. func (o *Optimistic) GetVersion() int64 {
    13.     return o.Version
    14. }
    16. func (o *Optimistic) SetVersion(version int64) {
    17.     o.Version = version
    18. }

    十九、并发条件下的有序输出

    1. package main
    3. import (
    4.     "fmt"
    5.     "sync"
    6.     "time"
    7. )
    9. func main() {
    10.     q := make(chan int)
    11.     wg := new(sync.WaitGroup)
    12.     wg.Add(1)
    13.     go func(wg *sync.WaitGroup, q chan<- int) {
    14.         defer wg.Done()
    15.         defer close(q)
    16.         for i := 1; i <= 100; i++ {
    17.             q <- i
    18.             time.Sleep(time.Microsecond)  // 利用休眠实现对生产者的控制，但这种方案不可靠
    19.         }
    21.     }(wg, q)
    22.     for i := 0; i < 3; i++ {
    23.         wg.Add(1)
    24.         go func(wg *sync.WaitGroup, q <-chan int) {
    25.             defer wg.Done()
    26.             for {
    27.                 if v, ok := <-q; ok {
    28.                     fmt.Println(v)
    29.                 } else {
    30.                     return
    31.                 }
    32.             }
    33.         }(wg, q)
    34.     }
    35.     wg.Wait()
    36. }
    37. // 利用WaitGroup实现安全退出，使生产者休眠来进行控制，但是如果消费者的操作是耗时是不可靠谱的，就会变得不可靠
    1. package main
    3. import (
    4.     "fmt"
    5.     "time"
    6. )
    8. func print(ch <-chan int){
    9.     for{
    10.         fmt.Println(<-ch)
    11.     }
    13. }
    15. func main() {
    16.     ch0 :=make(chan int)
    17.     ch1 :=make(chan int)
    18.     ch2 :=make(chan int)
    19.     go print(ch0)
    20.     go print(ch1)
    21.     go print(ch2)
    22.     for i:=1; i<101; i++{
    23.         switch{
    24.         case i%3 == 0:
    25.             ch0 <- i
    26.         case i%3 == 1:
    27.             ch1 <- i
    28.         case i%3 == 2:
    29.             ch2 <-i
    30.         }
    31.         time.Sleep(time.Microsecond)
    32.     }
    33. }
    34. //  利用for循环实现安全退出，同样是使生产者休眠来进行控制，但是如果消费者的操作是耗时是不可靠谱的，就会变得不可靠
    1. package main
    3. import (
    4.     "fmt"
    5.     "sync"
    6.     "time"
    7. )
    9. func main() {
    10.     q := make(chan int,100)
    11.     // 填充管道
    12.     for i := 1; i <= 100; i++ {
    13.         q<-i
    14.     }
    15.     cond := sync.NewCond(&sync.Mutex{})
    16.     for i := 0; i < 3; i++ {
    17.         go func(c *sync.Cond, q <-chan int) {
    18.             for {
    19.                 c.L.Lock()
    20.                 c.Wait()
    21.                 if v, ok := <-q; ok {
    22.                     fmt.Println(v)
    23.                     c.L.Unlock()
    24.                 } else {
    25.                     c.L.Unlock()
    26.                     return
    27.                 }
    28.             }
    29.         }(cond, q)
    30.     }
    31.     time.Sleep(time.Second) // 使所有的goroutine进入等待状态
    32.     for i := 1; i <= 100; i++ {
    33.         cond.Signal() // 唤醒一个等待状态的goroutine
    34.         time.Sleep(time.Microsecond)
    35.     }
    36.     close(q)
    37.     time.Sleep(time.Second)
    38. }
    39. // 利用cond是所有goroutine进入阻塞状态直到被唤醒来实现有序输出，
    40. cond和其他锁不一样，临界资源被锁之后goroutine不会进行轮询，而是等待调度，减少资源开销
    1. package main
    3. import (
    4.     "fmt"
    5.     "sync"
    6.     "time"
    7. )
    9. func main() {
    10.     q := make(chan int)
    11.     wg := new(sync.Mutex)
    12.     go func( q chan<- int) {
    13.         defer close(q)
    14.         for i := 1; i <= 100; i++ {
    15.             q <- i
    16.         }
    18.     }(q)
    19.     for i := 0; i < 3; i++ {
    20.         go func(wg *sync.Mutex, q <-chan int) {
    21.             for {
    22.                 wg.Lock()
    23.                 if v, ok := <-q; ok {
    24.                     fmt.Println(v)
    25.                     wg.Unlock()
    26.                 } else {
    27.                     wg.Unlock()
    28.                     return
    29.                 }
    30.             }
    31.         }(wg, q)
    32.     }
    33.     time.Sleep(time.Second)  // 安全退出
    34. }
    35. // 使用互斥锁阻塞其他协程来实现抢占，使用休眠实现安全退出，
    36. 这种方案是可靠的，但最好能够使用WaitGroup来实现安全退出
    1. package main
    3. import (
    4.     "fmt"
    5.     "time"
    6. )
    8. func main() {
    9.     a := 0
    10.     fn := func() int {
    11.         a++
    12.         return a
    13.     }
    15.     ch := make(chan int, 1)
    16.     defer close(ch)
    18.     for i := 0; i < 3; i++ {
    19.         go func() {
    20.             for {
    21.                 ch <- 1
    22.                 n := fn()
    23.                 if n > 100 {
    24.                     break
    25.                 }
    26.                 fmt.Println(n)
    27.                 <-ch
    28.             }
    29.         }()
    30.     }
    31.     time.Sleep(time.Second)  // 安全退出
    32. }
    33. // 使用管道来塞其他协程来实现抢占，使用休眠实现安全退出
    34. ，这种方案是可靠的，同理，最好能够使用WaitGroup来实现安全退出

    go.mod

    1. module datacenter
    3. go 1.15
    5. require (
    6.     github.com/Jeffail/tunny v0.0.0-20181108205650-4921fff29480
    7.     github.com/brianvoe/gofakeit/v6 v6.0.0
    8.     github.com/chanxuehong/wechat v0.0.0-20201110083048-0180211b69fd
    9.     github.com/coreos/go-semver v0.3.0 // indirect
    10.     github.com/dgrijalva/jwt-go v3.2.0+incompatible
    11.     github.com/fortytw2/leaktest v1.3.0 // indirect
    12.     github.com/gin-gonic/gin v1.6.3
    13.     github.com/go-playground/validator/v10 v10.4.1 // indirect
    14.     github.com/go-redis/redis v6.15.9+incompatible
    15.     github.com/go-sql-driver/mysql v1.5.0
    16.     github.com/gobwas/httphead v0.1.0 // indirect
    17.     github.com/gobwas/pool v0.2.1 // indirect
    18.     github.com/gobwas/ws v1.0.3
    19.     github.com/golang/protobuf v1.4.2
    20.     github.com/gotomicro/ego v0.3.7
    21.     github.com/jmoiron/sqlx v1.3.1
    22.     github.com/jolestar/go-commons-pool v2.0.0+incompatible
    23.     github.com/mattn/go-sqlite3 v2.0.1+incompatible // indirect
    24.     github.com/nats-io/nats-server/v2 v2.1.7 // indirect
    25.     github.com/nats-io/nats.go v1.10.0
    26.     github.com/pborman/uuid v1.2.1
    27.     github.com/qiniu/api.v7/v7 v7.7.0
    28.     github.com/spf13/pflag v1.0.5 // indirect
    29.     github.com/streadway/amqp v0.0.0-20200108173154-1c71cc93ed71
    30.     github.com/tal-tech/go-zero v1.0.28
    31.     github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5 // indirect
    32.     golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9
    33.     golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb
    34.     golang.org/x/time v0.0.0-20191024005414-555d28b269f0
    35.     google.golang.org/grpc v1.29.1
    36.     google.golang.org/protobuf v1.25.0
    37.     github.com/eclipse/paho.mqtt.golang v1.2.0
    38. )

    版权声明

    本站仅做备份收录，仅供研究与教学参考之用。
    读者将信息用于其他用途的，全部法律及连带责任由读者自行承担，本站不承担任何责任。

    ZONE.CI 全球网 | 安全领域涉猎者-乌云独行地带
    ZONE.CI 全球网
    安全领域涉猎者-乌云独行地带
    以太坊cppgolang区别 编程

    以太坊cppgolang区别

    以太坊是一种去中心化的开源平台，它采用智能合约技术，旨在构建和运行不受干扰的分布式应用程序。作为目前最受欢迎的区块链平台之一，以太坊提供了多种编程语言的支持，其
    12-282 评论
    progolang 编程

    progolang

    Go语言（Golang）是由Google开发的一门静态类型编程语言。作为一名专业的Golang开发者，我深知这门语言的优势和特点。在本文中，我将介绍Golang
    12-281 评论
    golangn个发送者 编程

    golangn个发送者

    Golang是一种开源的编程语言，由Google团队开发，旨在提高程序的并发性和简化软件开发过程。在Go语言中，有时需要向多个接收者发送信息。本文将介绍如何在G
    12-280 评论
    golang技能图谱 编程

    golang技能图谱

    从互联网行业的快速发展到人工智能技术的日益成熟，各种编程语言也应运而生。而在这众多的编程语言中，Golang（即Go）作为一门强大且高效的开发语言备受关注。Go
    12-281 评论
    ZONE.CI 全球网 | 安全领域涉猎者-乌云独行地带
    安全领域涉猎者-乌云独行地带
    ZONE.CI 全球网
    评论：0   参与：  9