本网站(662p.com)打包出售,且带程序代码数据,662p.com域名,程序内核采用TP框架开发,需要联系扣扣:2360248666 /wx:lianweikj
精品域名一口价出售:1y1m.com(350元) ,6b7b.com(400元) , 5k5j.com(380元) , yayj.com(1800元), jiongzhun.com(1000元) , niuzen.com(2800元) , zennei.com(5000元)
需要联系扣扣:2360248666 /wx:lianweikj
聊聊 Go 流水线编程模式
talkchan · 208浏览 · 发布于2021-11-08 +关注

流水线工作模型在工业领域内十分常见,它将工作流程分为多个环节,每个环节根据工作强度安排合适的人员数量。良好的流水线设计尽量让各环节的流通率平衡,最大化提高产能效率。

流水线工作模型在工业领域内十分常见,它将工作流程分为多个环节,每个环节根据工作强度安排合适的人员数量。良好的流水线设计尽量让各环节的流通率平衡,最大化提高产能效率。

Go 是一门实用性语言,流水线工作模型与 Go 融合地非常融洽,只不过我们一般使用另一个名词来表示流水线:pipeline。

pipeline

pipeline 由多个环节组成,具体在 Go 中,环节之间通过 channel 通信,同一个环节任务可以由多个 goroutine 来同时处理。

pipeline

pipeline 的核心是数据,通过 channel 来保证数据流动,每个环节的数据处理由 goroutine 完成。

除了开始环节和结束环节,每个环节都有任意数量的输入 channel 和输出 channel。开始环节被称为发送者或生产者,结束环节被称为接收者或消费者。

下面我们来看一个简单的 pipeline 例子,分为三个环节。

第一个环节,generate 函数:它充当生产者角色,将数据写入 channel,并把该 channel 返回。当所有数据写入完毕,关闭 channel。

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

    第二个环节,square 函数:它是数据处理的角色,从开始环节中的 channel 取出数据,计算平方,将结果写入新的 channel ,并把该新的 channel 返回。当所有数据计算完毕,关闭该新 channel。

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

      main 函数负责编排整个 pipeline ,并充当消费者角色:读取第二个环节的 channel 数据,打印出来。

      func main() { 
       // Set up the pipeline. 
       c := generate(2, 3) 
       out := square(c) 
      
       // Consume the output. 
       for n := range out { 
        fmt.Println(n) 
       } 
      }

        Fan-out,fan-in

        在上述例子中,环节之间通过非缓冲的 channel 传递数据,节点中的数据都是单个 goroutine 处理与消费。

        这种工作模式并不高效,会让整个流水线的效率取决于最慢的环节。因为每个环节中的任务量是不同的,这意味着我们需要的机器资源是存在差异的。任务量小的环节,尽量占有少量的机器资源,任务量重的环节,需要更多线程并行处理。

        以汽车组装为例,我们可以将组装轮胎的工作分发给 4 个人一起干,当轮胎组装完毕之后,再交由剩下的环节。

        多个 goroutine 可以从同一个 channel 读取数据,直到该通道关闭,这称为 fan-out(扇出)。

        这个称呼比较形象,它将数据进行分散,所以被称为扇出。扇出是一种分发任务的模式。

        fan-out

        单个 goroutine 可以从多个输入 channel 中读取数据,直到所有输入都关闭。具体做法是将输入 channel 多路复用到同一个 channel 上,当所有输入 channel 都关闭时,该 channel 也关闭,这称为 fan-in(扇入)。

        它将数据进行聚合,所以被称为扇入。扇入是一种整合任务结果的模式。

        fan-in

        在汽车组装的例子中,分发轮胎任务给每个人是 Fan-out,合并轮胎组装结果就是 Fan-in。

        channel 的多路复用

        扇出的编码模型比较简单,本文不多研究,我们提供一个扇入编程示例。

        创建一个生成器函数 generate,通过 interval 参数控制消息生成频率。生成器返回消息 channel mc与停止 channel sc,停止 channel 用于停止生成器任务。

        func generate(message string, interval time.Duration) (chan string, chan struct{}) { 
         mc := make(chan string) 
         sc := make(chan struct{}) 
        
         go func() { 
          defer func() { 
           close(sc) 
          }() 
        
          for { 
           select { 
           case <-sc: 
            return 
           default: 
            time.Sleep(interval) 
        
            mc <- message 
           } 
          } 
         }() 
        
         return mc, sc 
        }

          stopGenerating 函数通过通过向 sc 中传入空结构体,通知 generate退出,调用 close(mc) 关闭消息 channel

          func stopGenerating(mc chan string, sc chan struct{}) { 
           sc <- struct{}{} 
          
           close(mc) 
          }

            多路复用函数 multiplex 创建并返回整合消息 channel 和控制并发的 wg。

            func multiplex(mcs ...chan string) (chan string, *sync.WaitGroup) { 
             mmc := make(chan string) 
             wg := &sync.WaitGroup{} 
            
             for _, mc := range mcs { 
              wg.Add(1) 
            
              go func(mc chan string, wg *sync.WaitGroup) { 
               defer wg.Done() 
            
               for m := range mc { 
                mmc <- m 
               } 
              }(mc, wg) 
             } 
            
             return mmc, wg 
            }

              在 main 函数中,创建两个消息 channel 并复用它们生成 mmc ,打印来自 mmc 的每条消息。另外,我们还实现了接收系统断信号(终端上执行 CTRL+C 即可发送中断信号)的优雅的关闭机制。

              func main() { 
               // create two sample message and stop channels 
               mc1, sc1 := generate("message from generator 1", 200*time.Millisecond) 
               mc2, sc2 := generate("message from generator 2", 300*time.Millisecond) 
              
               // multiplex message channels 
               mmc, wg1 := multiplex(mc1, mc2) 
              
               // create errs channel for graceful shutdown 
               errs := make(chan error) 
              
               // wait for interrupt or terminate signal 
               go func() { 
                sc := make(chan os.Signal, 1) 
                signal.Notify(sc, syscall.SIGINT, syscall.SIGTERM) 
                errs <- fmt.Errorf("%s signal received", <-sc) 
               }() 
              
               // wait for multiplexed messages 
               wg2 := &sync.WaitGroup{} 
               wg2.Add(1) 
               go func() { 
                defer wg2.Done() 
              
                for m := range mmc { 
                 fmt.Println(m) 
                } 
               }() 
              
               // wait for errors 
               if err := <-errs; err != nil { 
                fmt.Println(err.Error()) 
               } 
              
               // stop generators 
               stopGenerating(mc1, sc1) 
               stopGenerating(mc2, sc2) 
               wg1.Wait() 
              
               // close multiplexed messages channel 
               close(mmc) 
               wg2.Wait() 
              }

                总结

                本文简单介绍了流水线编程模式,它和我们熟悉的生产者-消费者模式非常相似。

                具体到 Go 编程实践中,pipeline 将数据流分为多个环节,channel 用于数据流动,goroutine 用于处理数据。fan-out 用于分发任务,fan-in 用于数据整合,通过 FAN 模式可以让流水线更好地并发。

                当然,还有些细节需要注意,例如停止通知机制,可参照本文 channel 的多路复用章节示例中的 stopGenerating 函数;如何通过 sync.WaitGroup 做好并发控制,这些都是需要读者在实际编码中去体会掌握的。


                GO

                相关推荐

                PHP实现部分字符隐藏

                沙雕mars · 1312浏览 · 2019-04-28 09:47:56
                Java中ArrayList和LinkedList区别

                kenrry1992 · 896浏览 · 2019-05-08 21:14:54
                Tomcat 下载及安装配置

                manongba · 957浏览 · 2019-05-13 21:03:56
                JAVA变量介绍

                manongba · 953浏览 · 2019-05-13 21:05:52
                什么是SpringBoot

                iamitnan · 1077浏览 · 2019-05-14 22:20:36
                加载中

                0评论

                评论
                大家好,我是一名专注技术开发的技术屌丝,有什么问题可以互相交流,一起学习进步,谢谢。
                分类专栏
                小鸟云服务器
                扫码进入手机网页