1.mutexԴ??
2.Rust并发:标准库sync::Once源码分析
3.Qt互斥锁(QMutex)的使用、QMutexLocker的使用(含源码+注释)
4.C++ shared_mutex应用以及源码解析
5.Go并åç¼ç¨ï¼goroutineï¼channelåsync详解
mutexԴ??
我最近建立了一个在线自习室(App:番茄ToDO)用于相互监督学习,感兴趣的小伙伴可以加入。自习室加入码:D5A7A
Java并发包下的类大多基于AQS(AbstractQueuedSynchronizer)框架实现,而AQS线程安全的实现依赖于两个关键类:Unsafe和LockSupport。
其中,股票突破平台源码Unsafe主要提供CAS操作(关于CAS,在文章《读懂AtomicInteger源码(多线程专题)》中讲解过),LockSupport主要提供park/unpark操作。实际上,park/unpark操作的最终调用还是基于Unsafe类,因此Unsafe类才是核心。
Unsafe类的实现是由native关键字说明的,这意味着这个方法是原生函数,是用C/C++语言实现的,并被编译成了DLL,由Java去调用。
park函数的作用是将当前调用线程阻塞,而unpark函数则是唤醒指定线程。
park是泡妞秘籍网站源码等待一个许可,unpark是为某线程提供一个许可。如果线程A调用park,除非另一个线程调用unpark(A)给A一个许可,否则线程A将阻塞在park操作上。每次调用一次park,需要有一个unpark来解锁。
并且,unpark可以先于park调用,但不管unpark先调用多少次,都只提供一个许可,不可叠加。只需要一次park来消费掉unpark带来的许可,再次调用会阻塞。
在Linux系统下,park和unpark是通过Posix线程库pthread中的mutex(互斥量)和condition(条件变量)来实现的。
简单来说,mutex和condition保护了一个叫_counter的信号量。当park时,这个变量被设置为0,黑暗之光源码当unpark时,这个变量被设置为1。当_counter=0时线程阻塞,当_counter>0时直接设为0并返回。
每个Java线程都有一个Parker实例,Parker类的部分源码如下:
由源码可知,Parker类继承于PlatformParker,实际上是用Posix的mutex和condition来实现的。Parker类里的_counter字段,就是用来记录park和unpark是否需要阻塞的标识。
具体的执行逻辑已经用注释标记在代码中,简要来说,就是检查_counter是不是大于0,如果是,则把_counter设置为0,返回。如果等于零,继续执行,阻塞等待。arrays.sort 源码
unpark直接设置_counter为1,再unlock mutex返回。如果_counter之前的值是0,则还要调用pthread_cond_signal唤醒在park中等待的线程。源码如下:
(如果不会下载JVM源码可以后台回复“jdk”,获得下载压缩包)
Rust并发:标准库sync::Once源码分析
一次初始化同步原语Once,其核心功能在于确保闭包仅被执行一次。常见应用包括FFI库初始化、静态变量延迟初始化等。
标准库中的Once实现更为复杂,其关键在于如何高效地模拟Mutex阻塞与唤醒机制。这一机制依赖于线程暂停和唤醒原语thread::park/unpark,它们是实现多线程同步对象如Mutex、Condvar等的基础。
具体实现中,Once维护四个内部状态,状态与等待队列头指针共同存储于AtomicUsize中,利用4字节对齐优化空间。
构造Once实例时,性爱网源码初始化状态为Incomplete。调用Once::call_once或Once::call_once_force时,分别检查是否已完成初始化,未完成则执行闭包,闭包执行路径标记为冷路径以节省资源,同时避免泛型导致的代码膨胀。
闭包执行逻辑由Once::call_inner负责,线程尝试获取执行权限,未能获取则进入等待状态,获取成功后执行闭包,结束后唤醒等待线程。
等待队列通过无锁侵入式链表实现,节点在栈上分配,以优化内存使用。Once::wait函数实现等待线程逻辑,WaiterQueue的drop方法用于唤醒所有等待线程,需按特定顺序操作栈节点,以避免use after free等潜在问题。
思考题:如何在实际项目中利用Once实现资源安全共享?如何评估Once与Mutex等同步原语在不同场景下的性能差异?
Qt互斥锁(QMutex)的使用、QMutexLocker的使用(含源码+注释)
Qt中的互斥锁(QMutex)和QMutexLocker是为了解决多线程并发控制中的同步问题。QMutexLocker是一种封装了QMutex的便捷工具,用于自动管理锁的获取和释放,降低了在复杂程序中出错的风险。
QMutex在使用时需要手动进行锁定和解锁操作,但这种繁琐的过程容易导致忘记或错误操作。为简化这一过程,QMutexLocker被设计为局部变量,创建时传入一个QMutex指针并自动锁定,释放时自动解锁。这种设计使得代码更简洁,不易出错。
通过对比使用QMutex和QMutexLocker的代码,可以看出QMutexLocker省去了显式的锁定和解锁操作。例如,CMoveFuncClass使用QMutexLocker确保了在跨线程操作中的互斥性,而CThread则直接使用QMutex,但需要手动进行同步。在CMainWindow中,使用QMutexLocker的线程能确保互斥执行,另一个线程则在前一个线程完成后运行,证实了QMutexLocker的有效性。
总的来说,QMutexLocker为多线程编程提供了便利,减少了因忘记锁定或解锁带来的潜在问题。但在某些场景,如多线程循环输出,可能需要更精确的控制,此时QMutex可能更为合适。因此,选择使用QMutex或QMutexLocker应根据具体的需求和线程结构来决定。
C++ shared_mutex应用以及源码解析
在实际应用中,处理并发问题是开发实践中的一大挑战。当多个线程同时访问同一资源时,数据竞态问题便随之而来。为了解决此问题,互斥量(mutex)应运而生,它允许同一时刻只有一个线程访问临界资源,实现资源访问的排他性。
当线程间的读写操作频率不一致时,常规的互斥量无法满足高效访问的需求。此时,共享互斥锁(shared_mutex)成为了解决方案,它允许多个线程同时读取资源,而写操作则需要独占资源。这尤其适用于读操作频繁而写操作不频繁的场景,能显著提升程序效率。
下面,我们通过代码实例来探讨共享互斥锁的使用。定义读写锁时,首先引入`std::shared_mutex`。通过`std::shared_lock`操作,可以以共享方式立即获取锁,或在构造时以独占方式上锁。锁的释放则在析构函数中完成。
三个线程的示例代码展示了读写操作的并发执行。运行结果显示,读操作线程得到的临界资源值准确无误,证明了共享互斥锁在读操作并发时的正确性。然而,读操作线程的输出显示了一定程度的混乱,这并非共享互斥锁的问题,而是输出流操作的并发性导致的。
深入源码解析,我们可以发现`std::shared_lock`和`std::unique_lock`的实现细节。两者均使用RAII技术进行锁管理,但共享锁允许以共享或独占方式获取锁,而独占锁仅允许独占获取。源码中展示了锁的上锁和解锁过程,以及内部状态管理,包括持有锁状态的判断和更新。
共享互斥锁的底层实现基于`shared_mutex_base`类,通过一组成员变量和API封装了锁的管理逻辑。尝试加锁和解锁过程体现了锁的非阻塞特性。在进行锁的释放时,需要考虑共享持有状态,确保锁的正确释放。
总结而言,共享互斥锁提供了高效且灵活的并发控制机制,适用于读操作频繁、写操作不频繁的场景。通过深入源码解析,我们能够更全面地理解锁的实现细节和工作原理,从而在实际开发中更加有效地应用共享互斥锁,解决并发问题。
Go并åç¼ç¨ï¼goroutineï¼channelåsync详解
ä¼é ç并åç¼ç¨èå¼ï¼å®åç并åæ¯æï¼åºè²ç并åæ§è½æ¯Goè¯è¨åºå«äºå ¶ä»è¯è¨çä¸å¤§ç¹è²ãå¨å½ä»è¿ä¸ªå¤æ ¸æ¶ä»£ï¼å¹¶åç¼ç¨çæä¹ä¸è¨èå»ã使ç¨Goå¼å并åç¨åºï¼æä½èµ·æ¥é常ç®åï¼è¯è¨çº§å«æä¾å ³é®ågoç¨äºå¯å¨åç¨ï¼å¹¶ä¸å¨åä¸å°æºå¨ä¸å¯ä»¥å¯å¨æåä¸ä¸ä¸ªåç¨ã
ä¸é¢å°±æ¥è¯¦ç»ä»ç»ã
goroutineGoè¯è¨ç并åæ§è¡ä½ç§°ä¸ºgoroutineï¼ä½¿ç¨å ³é®è¯goæ¥å¯å¨ä¸ä¸ªgoroutineã
goå ³é®è¯åé¢å¿ é¡»è·ä¸ä¸ªå½æ°ï¼å¯ä»¥æ¯æåå½æ°ï¼ä¹å¯ä»¥æ¯æ åå½æ°ï¼å½æ°çè¿åå¼ä¼è¢«å¿½ç¥ã
goçæ§è¡æ¯éé»å¡çã
å æ¥çä¸ä¸ªä¾åï¼
packagemainimport("fmt""time")funcmain(){ gospinner(*time.Millisecond)constn=fibN:=fib(n)fmt.Printf("\rFibonacci(%d)=%d\n",n,fibN)//Fibonacci()=}funcspinner(delaytime.Duration){ for{ for_,r:=range`-\|/`{ fmt.Printf("\r%c",r)time.Sleep(delay)}}}funcfib(xint)int{ ifx<2{ returnx}returnfib(x-1)+fib(x-2)}ä»æ§è¡ç»ææ¥çï¼æå计ç®åºäºææ³¢é£å¥æ°åçå¼ï¼è¯´æç¨åºå¨spinnerå¤å¹¶æ²¡æé»å¡ï¼èä¸spinnerå½æ°è¿ä¸ç´å¨å±å¹ä¸æå°æ示å符ï¼è¯´æç¨åºæ£å¨æ§è¡ã
å½è®¡ç®å®ææ³¢é£å¥æ°åçå¼ï¼mainå½æ°æå°ç»æ并éåºï¼spinnerä¹è·çéåºã
åæ¥çä¸ä¸ªä¾åï¼å¾ªç¯æ§è¡æ¬¡ï¼æå°ä¸¤ä¸ªæ°çåï¼
packagemainimport"fmt"funcAdd(x,yint){ z:=x+yfmt.Println(z)}funcmain(){ fori:=0;i<;i++{ goAdd(i,i)}}æé®é¢äºï¼å±å¹ä¸ä»ä¹é½æ²¡æï¼ä¸ºä»ä¹å¢ï¼
è¿å°±è¦çGoç¨åºçæ§è¡æºå¶äºãå½ä¸ä¸ªç¨åºå¯å¨æ¶ï¼åªæä¸ä¸ªgoroutineæ¥è°ç¨mainå½æ°ï¼ç§°ä¸ºä¸»goroutineãæ°çgoroutineéè¿goå ³é®è¯å建ï¼ç¶å并åæ§è¡ãå½mainå½æ°è¿åæ¶ï¼ä¸ä¼çå¾ å ¶ä»goroutineæ§è¡å®ï¼èæ¯ç´æ¥æ´åç»ææægoroutineã
é£æ没æåæ³è§£å³å¢ï¼å½ç¶æ¯æçï¼è¯·å¾ä¸çã
channelä¸è¬åå¤è¿ç¨ç¨åºæ¶ï¼é½ä¼éå°ä¸ä¸ªé®é¢ï¼è¿ç¨é´éä¿¡ã常è§çéä¿¡æ¹å¼æä¿¡å·ï¼å ±äº«å åçãgoroutineä¹é´çéä¿¡æºå¶æ¯ééchannelã
使ç¨makeå建ééï¼
ch:=make(chanint)//chçç±»åæ¯chanintééæ¯æä¸ä¸ªä¸»è¦æä½ï¼sendï¼receiveåcloseã
ch<-x//åéx=<-ch//æ¥æ¶<-ch//æ¥æ¶ï¼ä¸¢å¼ç»æclose(ch)//å ³éæ ç¼å²channelmakeå½æ°æ¥å两个åæ°ï¼ç¬¬äºä¸ªåæ°æ¯å¯éåæ°ï¼è¡¨ç¤ºéé容éãä¸ä¼ æè ä¼ 0表示å建äºä¸ä¸ªæ ç¼å²ééã
æ ç¼å²ééä¸çåéæä½å°ä¼é»å¡ï¼ç´å°å¦ä¸ä¸ªgoroutineå¨å¯¹åºçééä¸æ§è¡æ¥æ¶æä½ãç¸åï¼å¦ææ¥æ¶å æ§è¡ï¼é£ä¹æ¥æ¶goroutineå°ä¼é»å¡ï¼ç´å°å¦ä¸ä¸ªgoroutineå¨å¯¹åºééä¸æ§è¡åéã
æ以ï¼æ ç¼å²ééæ¯ä¸ç§åæ¥ééã
ä¸é¢æ们使ç¨æ ç¼å²ééæä¸é¢ä¾åä¸åºç°çé®é¢è§£å³ä¸ä¸ã
packagemainimport"fmt"funcAdd(x,yint,chchanint){ z:=x+ych<-z}funcmain(){ ch:=make(chanint)fori:=0;i<;i++{ goAdd(i,i,ch)}fori:=0;i<;i++{ fmt.Println(<-ch)}}å¯ä»¥æ£å¸¸è¾åºç»æã
主goroutineä¼é»å¡ï¼ç´å°è¯»åå°ééä¸çå¼ï¼ç¨åºç»§ç»æ§è¡ï¼æåéåºã
ç¼å²channelå建ä¸ä¸ªå®¹éæ¯5çç¼å²ééï¼
ch:=make(chanint,5)ç¼å²ééçåéæä½å¨ééå°¾é¨æå ¥ä¸ä¸ªå ç´ ï¼æ¥æ¶æä½ä»ééç头é¨ç§»é¤ä¸ä¸ªå ç´ ãå¦æéé满äºï¼åéä¼é»å¡ï¼ç´å°å¦ä¸ä¸ªgoroutineæ§è¡æ¥æ¶ãç¸åï¼å¦æééæ¯ç©ºçï¼æ¥æ¶ä¼é»å¡ï¼ç´å°å¦ä¸ä¸ªgoroutineæ§è¡åéã
æ没ææè§ï¼å ¶å®ç¼å²ééåéåä¸æ ·ï¼ææä½é½è§£è¦äºã
ååchannelç±»åchan<-intæ¯ä¸ä¸ªåªè½åéçééï¼ç±»å<-chanintæ¯ä¸ä¸ªåªè½æ¥æ¶çééã
ä»»ä½ååééé½å¯ä»¥ç¨ä½ååééï¼ä½åè¿æ¥ä¸è¡ã
è¿æä¸ç¹éè¦æ³¨æï¼closeåªè½ç¨å¨åéééä¸ï¼å¦æç¨å¨æ¥æ¶ééä¼æ¥éã
çä¸ä¸ªååééçä¾åï¼
packagemainimport"fmt"funccounter(outchan<-int){ forx:=0;x<;x++{ out<-x}close(out)}funcsquarer(outchan<-int,in<-chanint){ forv:=rangein{ out<-v*v}close(out)}funcprinter(in<-chanint){ forv:=rangein{ fmt.Println(v)}}funcmain(){ n:=make(chanint)s:=make(chanint)gocounter(n)gosquarer(s,n)printer(s)}syncsyncå æä¾äºä¸¤ç§éç±»åï¼sync.Mutexåsync.RWMutexï¼åè æ¯äºæ¥éï¼åè æ¯è¯»åéã
å½ä¸ä¸ªgoroutineè·åäºMutexåï¼å ¶ä»goroutineä¸ç®¡è¯»åï¼åªè½çå¾ ï¼ç´å°é被éæ¾ã
packagemainimport("fmt""sync""time")funcmain(){ varmutexsync.Mutexwg:=sync.WaitGroup{ }//主goroutineå è·åéfmt.Println("Locking(G0)")mutex.Lock()fmt.Println("locked(G0)")wg.Add(3)fori:=1;i<4;i++{ gofunc(iint){ //ç±äºä¸»goroutineå è·åéï¼ç¨åºå¼å§5ç§ä¼é»å¡å¨è¿éfmt.Printf("Locking(G%d)\n",i)mutex.Lock()fmt.Printf("locked(G%d)\n",i)time.Sleep(time.Second*2)mutex.Unlock()fmt.Printf("unlocked(G%d)\n",i)wg.Done()}(i)}//主goroutine5ç§åéæ¾étime.Sleep(time.Second*5)fmt.Println("readyunlock(G0)")mutex.Unlock()fmt.Println("unlocked(G0)")wg.Wait()}RWMutexå±äºç»å ¸çååå¤è¯»æ¨¡åï¼å½è¯»é被å ç¨æ¶ï¼ä¼é»æ¢åï¼ä½ä¸é»æ¢è¯»ãèåéä¼é»æ¢åå读ã
packagemainimport("fmt""sync""time")funcmain(){ varrwMutexsync.RWMutexwg:=sync.WaitGroup{ }Data:=0wg.Add()fori:=0;i<;i++{ gofunc(tint){ //第ä¸æ¬¡è¿è¡åï¼å解éã//循ç¯å°ç¬¬äºæ¬¡æ¶ï¼è¯»éå®åï¼goroutine没æé»å¡ï¼åæ¶è¯»æåãfmt.Println("Locking")rwMutex.RLock()deferrwMutex.RUnlock()fmt.Printf("Readdata:%v\n",Data)wg.Done()time.Sleep(2*time.Second)}(i)gofunc(tint){ //åéå®ä¸æ¯éè¦è§£éåæè½åçrwMutex.Lock()deferrwMutex.Unlock()Data+=tfmt.Printf("WriteData:%v%d\n",Data,t)wg.Done()time.Sleep(2*time.Second)}(i)}wg.Wait()}æ»ç»å¹¶åç¼ç¨ç®æ¯Goçç¹è²ï¼ä¹æ¯æ ¸å¿åè½ä¹ä¸äºï¼æ¶åçç¥è¯ç¹å ¶å®æ¯é常å¤çï¼æ¬æä¹åªæ¯èµ·å°ä¸ä¸ªæç å¼ççä½ç¨èå·²ã
æ¬æå¼å§ä»ç»äºgoroutineçç®åç¨æ³ï¼ç¶åå¼åºäºééçæ¦å¿µã
ééæä¸ç§ï¼
æ ç¼å²éé
ç¼å²éé
ååéé
æåä»ç»äºGoä¸çéæºå¶ï¼åå«æ¯syncå æä¾çsync.Mutexï¼äºæ¥éï¼åsync.RWMutexï¼è¯»åéï¼ã
goroutineå大精深ï¼åé¢çåè¿æ¯è¦æ ¢æ ¢è¸©çã
æç« ä¸çèå¾åæºç é½ä¸ä¼ å°äºGitHubï¼æéè¦çåå¦å¯èªè¡ä¸è½½ã
å°åï¼github.com/yongxinz/gopher/tree/main/sc
ä½è ï¼yongxinz