singleflight 使用方法以及源码阅读

1、简介

安装方式：

go get -u golang.org/x/sync/singleflight

singleflight 是Go官方扩展同步包的一个库。通过给每次函数调用分配一个key，相同key的函数并发调用时，在函数执行期间，相同函数的调用，只会被执行一次，返回相同的结果。其本质是对函数调用的结果进行复用。

2、使用方法

2.1 使用Do获取函数执行结果

Do方法是同步返回函数执行结果

package main

import (

	"fmt"

	"golang.org/x/sync/singleflight"

	"runtime"

	"sync"

	"time"

)

func main()  {

	var sg singleflight.Group

	var wg sync.WaitGroup

	for i := 0; i < 10; i++ {

		wg.Add(1)

		go func(j int) {

			defer wg.Done()

			v, err, shared := sg.Do("testDo", testDo)

			fmt.Printf("i: %v, v:%v, err:%v, shared:%v\n", j, v, err, shared)

		}(i)

	}

	wg.Wait()

}

func testDo() (interface{}, error) {

	// 模拟函数执行需要的时间

	time.Sleep(time.Millisecond)

	return "testDo", nil

}

2.2 使用DoChan获取函数执行结果

DoChan返回一个 channel，函数执行的结果通过 channel 来进行传递。

package main

import (

	"fmt"

	"golang.org/x/sync/singleflight"

	"runtime"

	"sync"

	"time"

)

func main()  {

	var sg singleflight.Group

	var wg sync.WaitGroup

	for i := 0; i < 10; i++ {

		wg.Add(1)

		go func(j int) {

			defer wg.Done()

			ch := sg.DoChan("testDoChan", testDoChan)

			select {

			case ret := <- ch:

				fmt.Printf("i: %v, v:%v, err:%v, shared:%v\n", j, ret.Val, ret.Err, ret.Shared)

			}

		}(i)

	}

	wg.Wait()

}

func testDoChan() (interface{}, error) {

	// 模拟函数执行需要的时间

	time.Sleep(time.Millisecond)

	return "testDoChan", nil

}

3、源码解读

3.1 Group

//Group 整个库的核心结构体

type Group struct {

	mu sync.Mutex       // 并发时，保护 m

	m  map[string]*call // 使用 懒加载 方式进行初始化

}

3.2 call

//call m中的value

type call struct {

	wg sync.WaitGroup

	//相同key，fn执行的返回结果

	val interface{}

	err error

	//fn执行期间，相同 key 添加的次数，第一次添加不算

	dups  int

	chans []chan<- Result // DoChan 返回fn执行的结果

}

3.3 Group.Do

//Do 执行函数的地方，key: 给函数自定义的标识

//fn: 需要执行的函数，fn开始运行后，未运行结果前，这个期间对相同key的调用，都会返回第一次执行fn返回的结果

//v:fn执行返回的结果，err:fn执行返回的err

//shared:fn执行结果是否会共享，fn运行期间，是否有相同的key被调用，有则返回true,反之返回false

func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool) {

	g.mu.Lock()

	if g.m == nil {//懒加载

		g.m = make(map[string]*call)

	}

	if c, ok := g.m[key]; ok {//fn执行期间，又有相同的key添加进来执行

		c.dups++ //fn执行期间，有相同的key添加进来

		g.mu.Unlock()

		c.wg.Wait() //等待fn执行结果（fn函数里面，会调用c.wg.Done）

		//-------

		//判断fn执行过程中，是否有 panic 或者 runtime.Goexit()

		//感觉主要是为了 DoChan 函数，DoChan 返回的是channel，防止fn函数执行期间出现问题，导致无法往 chan 里面写入结果。

		//从而导致 外面需要获取 fn 执行结果的协程一直在等待

		if e, ok := c.err.(*panicError); ok {

			panic(e)

		} else if c.err == errGoexit {

			runtime.Goexit()

		}

		return c.val, c.err, true

	}

	//---- 以下是key 第一次添加到 m 中时，执行的代码---

	c := new(call)

	c.wg.Add(1)

	g.m[key] = c

	g.mu.Unlock()

	// 执行 fn 的地方

	g.doCall(c, key, fn) // 没有新开一个协程，和DoChan不同。

	return c.val, c.err, c.dups > 0

}

3.4 Group.DoChan

//DoChan 和Do 十分类似，只不过返回的结果通过 chan 来传递

func (g *Group) DoChan(key string, fn func() (interface{}, error)) <-chan Result {

	ch := make(chan Result, 1)

	g.mu.Lock()

	if g.m == nil {//懒加载

		g.m = make(map[string]*call)

	}

	if c, ok := g.m[key]; ok {//fn执行期间，又有相同的key添加进来执行

		c.dups++

		c.chans = append(c.chans, ch)

		g.mu.Unlock()

		return ch

	}

	//---- 以下是key 第一次添加到 m 中时，执行的代码---

	c := &call{chans: []chan<- Result{ch}}

	c.wg.Add(1)

	g.m[key] = c

	g.mu.Unlock()

	go g.doCall(c, key, fn) // 新开启了一个协程，和Do不同

	return ch

}

3.5 Group.doCall

• 双defer+normalReturn+recovered 判断fn执行是panic还是runtime.Goexit

//doCall 真正运行fn的地方，需要重点理解

func (g *Group) doCall(c *call, key string, fn func() (interface{}, error)) {

	normalReturn := false //是否正常返回，默认false

	recovered := false //是否recover，默认false

	// use double-defer to distinguish panic from runtime.Goexit,

	//使用双 defer 来区分 panic和runtime.Goexit

	//是需要结合 normalReturn 和 recovere 的值来进行判断，从而区分是panic还是runtime.Goexit

	defer func() {

		// the given function invoked runtime.Goexit

		if !normalReturn && !recovered {

			//既没有正常返回，又没有被 recover，所以是fn执行期间，调用了 runtime.Goexit()

			c.err = errGoexit

		}

		g.mu.Lock()

		defer g.mu.Unlock()

		c.wg.Done()

		// 走到这里，fn函数已经执行过了

		if g.m[key] == c {

			delete(g.m, key) //fn函数执行完毕，好让后续的key可以继续进来执行fn函数

		}

		if e, ok := c.err.(*panicError); ok { // recover住的错误

			// In order to prevent the waiting channels from being blocked forever,

			// needs to ensure that this panic cannot be recovered.

			if len(c.chans) > 0 { //通过使用DoChan来执行 fn，发生的错误

				go panic(e) // recover只能够 recover住同一个协程里的panic，不是同一个协程的无法捕获。

				select {} // 保证协程不退出，错误会直接暴露出去

			} else {  //通过使用 Do来执行fn，发生的错误

				panic(e)

			}

		} else if c.err == errGoexit {

			// Already in the process of goexit, no need to call again

			//第一个调用的fn函数的协程已经退出，相同key的函数因为 chan 接收不到数据，会发生死锁（）

			//fatal error: all goroutines are asleep - deadlock!

		} else {

			// Normal return

			for _, ch := range c.chans {

				ch <- Result{c.val, c.err, c.dups > 0}

			}

		}

	}()

	func() {

		defer func() {

			if !normalReturn {//fn执行期间，发生了panic

				if r := recover(); r != nil {

					c.err = newPanicError(r)  // 标识为panic错误，Do函数中判断时，好做区分 e, ok := c.err.(*panicError)

				}

			}

		}()

		c.val, c.err = fn()

		normalReturn = true  //fn执行期间，没有panic

	}()

	if !normalReturn {

		recovered = true  //fn执行期间，发生了panic,并且被 recover住了，注意：调用runtime.Goexit()时，是无法recover的

	}

}

3.6 Group.Forget

//Forget 使用Do执行fn时，可以手动删除 g.m 中的key

func (g *Group) Forget(key string){

	g.mu.Lock()

	delete(g.m, key)

	g.mu.Unlock()

}

4、执行流程

菜鸟一枚，文中难免有错误的地方，如有，恳请大佬指出。

5、参考资料

绝对详尽的singleflight讲解

singleflight 使用记录以及源码阅读的更多相关文章

1. EventBus源码解析 源码阅读记录

   EventBus源码阅读记录 repo地址: greenrobot/EventBus EventBus的构造 双重加锁的单例. static volatile EventBus defaultInst ...

2. 【原】AFNetworking源码阅读（五）

   [原]AFNetworking源码阅读(五) 本文转载请注明出处 —— polobymulberry-博客园 1. 前言 上一篇中提及到了Multipart Request的构建方法- [AFHTTP ...

3. 【原】AFNetworking源码阅读（三）

   [原]AFNetworking源码阅读(三) 本文转载请注明出处 —— polobymulberry-博客园 1. 前言 上一篇的话,主要是讲了如何通过构建一个request来生成一个data tas ...

4. 【原】AFNetworking源码阅读（一）

   [原]AFNetworking源码阅读(一) 本文转载请注明出处 —— polobymulberry-博客园 1. 前言 AFNetworking版本:3.0.4 由于我平常并没有经常使用AFNetw ...

5. [PHP源码阅读]explode和implode函数

   explode和implode函数主要用作字符串和数组间转换的操作,比如获取一段参数后根据某个字符分割字符串,或者将一个数组的结果使用一个字符合并成一个字符串输出.在PHP中经常会用到这两个函数,因此 ...

6. 【原】SDWebImage源码阅读（五）

   [原]SDWebImage源码阅读(五) 本文转载请注明出处 —— polobymulberry-博客园 1. 前言 前面的代码并没有特意去讲SDWebImage的缓存机制,主要是想单独开一章节专门讲 ...

7. 源码阅读系列：EventBus

   title: 源码阅读系列:EventBus date: 2016-12-22 16:16:47 tags: 源码阅读 --- EventBus 是人们在日常开发中经常会用到的开源库,即使是不直接用的 ...

8. angular源码阅读，依赖注入的原理：injector,provider,module之间的关系。

   最开始使用angular的时候,总是觉得它的依赖注入方式非常神奇. 如果你跳槽的时候对新公司说,我曾经使用过angular,那他们肯定会问你angular的依赖注入原理是什么? 这篇博客其实是angu ...

9. CI框架源码阅读笔记5 基准测试 BenchMark.php

   上一篇博客(CI框架源码阅读笔记4 引导文件CodeIgniter.php)中,我们已经看到:CI中核心流程的核心功能都是由不同的组件来完成的.这些组件类似于一个一个单独的模块,不同的模块完成不同的功 ...

10. CI框架源码阅读笔记4 引导文件CodeIgniter.php

    到了这里,终于进入CI框架的核心了.既然是“引导”文件,那么就是对用户的请求.参数等做相应的导向,让用户请求和数据流按照正确的线路各就各位.例如,用户的请求url: http://you.host.c ...

随机推荐

1. 【2020NOI.AC省选模拟#2】A. 旋转

   题目链接 原题解: 把每个点的坐标视为复数,那么每次询问就是区间求平均数(先求和然后除以个数).一个点绕着原点旋转就是乘以$(\cos 60^\circ +i\sin 60^\circ)$. 一个点绕 ...

2. continue的使用

   continue是终止本次循环,进行下一个循环如果要跳到最开始的循环abc : for(){ for(){ for(){ continue abc; } }}

3. mysql之存储引擎-第二篇

   什么是存储引擎? 数据库存储引擎是数据库底层软件组件,数据库管理系统使用数据引擎进行创建,查询,更新和删除数据操作.不同的存储引擎提供了不同的存储机制,索引技巧及特定功能. 存储引擎类型 InnoDB ...

4. 4、excel快速排序从1开始

   在分世界杯的文件时,要求把每一行从1开始排列,自己的做法就是先输入1和2,然后拖黑1和2,接着鼠标一直往下拖,这样就了. 公式的方法: 输入公式=Row()-1,如何在这个单元格的右下角双击就可以了 ...

5. SourceMap解析

   前端发展至今已不再是刀耕火种的年代了,出现了typescript.babel.uglify.js等功能强大的工具.我们手动撰写的代码一般具有可读性,并且可以享受高级语法.类型检查带来的便利,但经过工具 ...

6. UnsupportedOperationException异常

   看看下面的例子,这样输出什么呢? public class test { public static void main(String[] args) { String arr = "ab, ...

7. 获取客户端ip，请求头伪造ip，解决办法

   可以在请求头加入 X-Forwarder-For 来伪造访问的ip地址 //Nginx支持X-Forwarded-For 配置 proxy_set_header X-Forwarded-For $pr ...

8. 文献阅读——Watch Sat and LTO for CaDiCaL and Kissat

   Norbert Mantheynmanthey@conp-solutions.comDresden, Germany Proceedings of SAT Competition 2022: Solv ...

9. dpkt 简单应用

   import dpktfrom dpkt.utils import mac_to_str,inet_to_strcap=f'D:/test_pacp/6.pcap'with open(cap,'rb' ...

10. LOJ数列分块入门九题（中）

    #6281. 数列分块入门 5 - 题目 - LibreOJ (loj.ac) 区间开方,区间求和题. 显然,针对区间维护开方操作很难做到,于是考虑其值的性质,显然,int范围内的值最多开方6次就会变 ...