本文旨在帮助开发者理解和解决 Go 并发程序中常见的死锁问题。通过分析一个包含三个 Goroutine 相互通信的示例程序，我们将深入探讨死锁产生的原因，并提供有效的调试和修复策略，包括使用 runtime.Gosched() 和缓冲 Channel 来避免死锁，同时强调并发程序设计的复杂性和潜在的非确定性行为。
理解 Go 中的死锁
在 Go 语言中，死锁通常发生在多个 Goroutine 试图通过 Channel 进行通信时，由于某种原因，它们都在等待对方释放资源或发送/接收数据，从而导致所有 Goroutine 都被阻塞，程序无法继续执行。Go 运行时会检测到这种状态，并抛出 “fatal error: all goroutines are asleep – deadlock!” 错误。
死锁的常见原因包括：
- 循环等待： 多个 Goroutine 互相等待对方释放资源。
- Channel 容量不足： 使用无缓冲 Channel 时，发送方必须等待接收方准备好才能发送数据，如果接收方被阻塞，发送方也会被阻塞，从而可能导致死锁。
- 错误的 Channel 操作： 例如，向已关闭的 Channel 发送数据，或者从已关闭且没有数据的 Channel 接收数据。
示例代码分析与死锁排查
下面是一个可能导致死锁的示例代码，它包含三个 Goroutine，它们通过 Channel 互相发送整数：
package main import ( "fmt" "math/rand" "runtime" ) func Routine1(command12 chan int, response12 chan int, command13 chan int, response13 chan int) { z12 := 200 z13 := 200 m12 := false m13 := false y := 0 for i := 0; i < 20; i++ { y = rand.Intn(100) if y == 0 { fmt.Println(z12, " z12 STATE SAVED") fmt.Println(z13, " z13 STATE SAVED") y = 0 command12 <- y command13 <- y for m12 != true || m13 != true { select { case cmd1 := <-response12: { z12 = cmd1 if z12 != 0 { fmt.Println(z12, " z12 Channel Saving.... ") y = rand.Intn(100) command12 <- y } if z12 == 0 { m12 = true fmt.Println(" z12 Channel Saving Stopped ") } } case cmd2 := <-response13: { z13 = cmd2 if z13 != 0 { fmt.Println(z13, " z13 Channel Saving.... ") y = rand.Intn(100) command13 <- y } if z13 == 0 { m13 = true fmt.Println(" z13 Channel Saving Stopped ") } } } } m12 = false m13 = false } if y != 0 { if y%2 == 0 { command12 <- y } if y%2 != 0 { command13 <- y } select { case cmd1 := <-response12: { z12 = cmd1 fmt.Println(z12, " z12") } case cmd2 := <-response13: { z13 = cmd2 fmt.Println(z13, " z13") } } } } close(command12) close(command13) } func Routine2(command12 chan int, response12 chan int, command23 chan int, response23 chan int) { z21 := 200 z23 := 200 m21 := false m23 := false for i := 0; i < 20; i++ { select { case x, open := <-command12: { if !open { return } if x != 0 && m23 != true { z21 = x fmt.Println(z21, " z21") } if x != 0 && m23 == true { z21 = x fmt.Println(z21, " z21 Channel Saving ") } if x == 0 { m21 = true if m21 == true && m23 == true { fmt.Println(" z21 and z23 Channel Saving Stopped ") m23 = false m21 = false } if m21 == true && m23 != true { z21 = x fmt.Println(z21, " z21 Channel Saved ") } } } case x, open := <-response23: { if !open { return } if x != 0 && m21 != true { z23 = x fmt.Println(z23, " z21") } if x != 0 && m21 == true { z23 = x fmt.Println(z23, " z23 Channel Saving ") } if x == 0 { m23 = true if m21 == true && m23 == true { fmt.Println(" z23 Channel Saving Stopped ") m23 = false m21 = false } if m23 == true && m21 != true { z23 = x fmt.Println(z23, " z23 Channel Saved ") } } } } if m23 == false && m21 == false { y := rand.Intn(100) if y%2 == 0 { if y == 0 { y = 10 response12 <- y } } if y%2 != 0 { if y == 0 { y = 10 response23 <- y } } } if m23 == true && m21 != true { y := rand.Intn(100) response12 <- y } if m23 != true && m21 == true { y := rand.Intn(100) command23 <- y } } close(response12) close(command23) } func Routine3(command13 chan int, response13 chan int, command23 chan int, response23 chan int) { z31 := 200 z32 := 200 m31 := false m32 := false for i := 0; i < 20; i++ { select { case x, open := <-command13: { if !open { return } if x != 0 && m32 != true { z31 = x fmt.Println(z31, " z21") } if x != 0 && m32 == true { z31 = x fmt.Println(z31, " z31 Channel Saving ") } if x == 0 { m31 = true if m31 == true && m32 == true { fmt.Println(" z21 Channel Saving Stopped ") m31 = false m32 = false } if m31 == true && m32 != true { z31 = x fmt.Println(z31, " z31 Channel Saved ") } } } case x, open := <-command23: { if !open { return } if x != 0 && m31 != true { z32 = x fmt.Println(z32, " z32") } if x != 0 && m31 == true { z32 = x fmt.Println(z32, " z32 Channel Saving ") } if x == 0 { m32 = true if m31 == true && m32 == true { fmt.Println(" z32 Channel Saving Stopped ") m31 = false m32 = false } if m32 == true && m31 != true { z32 = x fmt.Println(z32, " z32 Channel Saved ") } } } } if m31 == false && m32 == false { y := rand.Intn(100) if y%2 == 0 { response13 <- y } if y%2 != 0 { response23 <- y } } if m31 == true && m32 != true { response13 <- y } if m31 != true && m32 == true { response23 <- y } } close(response13) close(response23) } func main() { command12 := make(chan int) response12 := make(chan int) command13 := make(chan int) response13 := make(chan int) command23 := make(chan int) response23 := make(chan int) go Routine1(command12, response12, command13, response13) go Routine2(command12, response12, command23, response23) Routine3(command13, response13, command23, response23) }
登录后复制
这段代码创建了三个 Goroutine，它们通过六个 Channel 进行通信。Routine1 是发起者，它可以发送 0 值来请求其他 Goroutine 保存状态。每个 Goroutine 都有一个内部状态，并通过 Channel 交换数据。
死锁分析：
这段代码的复杂性使得静态分析变得困难，但我们可以推断出潜在的死锁点：
- 在 Routine1 中，当 y == 0 时，它会向 command12 和 command13 发送 0，并进入一个循环，等待从 response12 和 response13 接收 0。如果 Routine2 和 Routine3 由于某种原因无法发送 0，Routine1 将永远被阻塞。
- 在 Routine2 和 Routine3 中，它们都在 select 语句中等待从两个 Channel 接收数据。如果它们都在等待对方发送数据，就会形成循环等待。
解决死锁的策略
-
使用 runtime.Gosched()：
runtime.Gosched() 函数可以让当前 Goroutine 放弃 CPU 的使用权，让其他 Goroutine 有机会运行。在 select 语句中添加 default 分支并调用 runtime.Gosched() 可以避免 Goroutine 一直占用 CPU，从而降低死锁的风险。
select { case cmd1 := <-response12: // ... case cmd2 := <-response13: // ... default: runtime.Gosched() }登录后复制 -
使用缓冲 Channel：
将无缓冲 Channel 替换为缓冲 Channel 可以缓解死锁问题。缓冲 Channel 允许发送方在 Channel 未满时发送数据，而无需等待接收方准备好。这可以避免发送方因为等待接收方而被阻塞。
command12 := make(chan int, 10) // 创建一个容量为 10 的缓冲 Channel
登录后复制注意： 缓冲 Channel 只是缓解死锁，并不能完全避免死锁。如果缓冲 Channel 满了，发送方仍然会被阻塞。
-
代码重构：
最根本的解决方案是重新设计并发程序，避免复杂的 Channel 交互和循环等待。可以考虑使用更高级的并发模式，例如 sync.WtGroup、context 等，或者使用更简单的通信方式，例如共享内存和锁。
修改后的代码示例
下面是修改后的代码，使用了 runtime.Gosched() 和缓冲 Channel：
package main import ( "fmt" "math/rand" "runtime" ) const bufferSize = 4 // 缓冲大小 func Routine1(command12 chan int, response12 chan int, command13 chan int, response13 chan int) { z12 := 200 z13 := 200 m12 := false m13 := false y := 0 for i := 0; i < 20; i++ { y = rand.Intn(100) if y == 0 { fmt.Println(z12, " z12 STATE SAVED") fmt.Println(z13, " z13 STATE SAVED") y = 0 command12 <- y command13 <- y for m12 != true || m13 != true { select { case cmd1 := <-response12: { z12 = cmd1 if z12 != 0 { fmt.Println(z12, " z12 Channel Saving.... ") y = rand.Intn(100) command12 <- y } if z12 == 0 { m12 = true fmt.Println(" z12 Channel Saving Stopped ") } } case cmd2 := <-response13: { z13 = cmd2 if z13 != 0 { fmt.Println(z13, " z13 Channel Saving.... ") y = rand.Intn(100) command13 <- y } if z13 == 0 { m13 = true fmt.Println(" z13 Channel Saving Stopped ") } } default: runtime.Gosched() } } m12 = false m13 = false } if y != 0 { if y%2 == 0 { command12 <- y } if y%2 != 0 { command13 <- y } select { case cmd1 := <-response12: { z12 = cmd1 fmt.Println(z12, " z12") } case cmd2 := <-response13: { z13 = cmd2 fmt.Println(z13, " z13") } default: runtime.Gosched() } } } close(command12) close(command13) } func Routine2(command12 chan int, response12 chan int, command23 chan int, response23 chan int) { z21 := 200 z23 := 200 m21 := false m23 := false for i := 0; i < 20; i++ { select { case x, open := <-command12: { if !open { return } if x != 0 && m23 != true { z21 = x fmt.Println(z21, " z21") } if x != 0 && m23 == true { z21 = x fmt.Println(z21, " z21 Channel Saving ") } if x == 0 { m21 = true if m21 == true && m23 == true { fmt.Println(" z21 and z23 Channel Saving Stopped ") m23 = false m21 = false } if m21 == true && m23 != true { z21 = x fmt.Println(z21, " z21 Channel Saved ") } } } case x, open := <-response23: { if !open { return } if x != 0 && m21 != true { z23 = x fmt.Println(z23, " z21") } if x != 0 && m21 == true { z23 = x fmt.Println(z23, " z23 Channel Saving ") } if x == 0 { m23 = true if m21 == true && m23 == true { fmt.Println(" z23 Channel Saving Stopped ") m23 = false m21 = false } if m23 == true && m21 != true { z23 = x fmt.Println(z23, " z23 Channel Saved ") } } } default: runtime.Gosched() } if m23 == false && m21 == false { y := rand.Intn(100) if y%2 == 0 { if y == 0 { y = 10 response12 <- y } } if y%2 != 0 { if y == 0 { y = 10 response23 <- y } } } if m23 == true && m21 != true { y := rand.Intn(100) response12 <- y } if m23 != true && m21 == true { y := rand.Intn(100) command23 <- y } } close(response12) close(command23) } func Routine3(command13 chan int, response13 chan int, command23 chan int, response23 chan int) { z31 := 200 z32 := 200 m31 := false m32 := false for i := 0; i < 20; i++ { select { case x, open := <-command13: { if !open { return } if x != 0 && m32 != true { z31 = x fmt.Println(z31, " z21") } if x != 0 && m32 == true { z31 = x fmt.Println(z31, " z31 Channel Saving ") } if x == 0 { m31 = true if m31 == true && m32 == true { fmt.Println(" z21 Channel Saving Stopped ") m31 = false m32 = false } if m31 == true && m32 != true { z31 = x fmt.Println(z31, " z31 Channel Saved ") } } } case x, open := <-command23: { if !open { return } if x != 0 && m31 != true { z32 = x fmt.Println(z32, " z32") } if x != 0 && m31 == true { z32 = x fmt.Println(z32, " z32 Channel Saving ") } if x == 0 { m32 = true if m31 == true && m32 == true { fmt.Println(" z32 Channel Saving Stopped ") m31 = false m32 = false } if m32 == true && m31 != true { z32 = x fmt.Println(z32, " z32 Channel Saved ") } } } default: runtime.Gosched() } if m31 == false && m32 == false { y := rand.Intn(100) if y%2 == 0 { response13 <- y } if y%2 != 0 { response23 <- y } } if m31 == true && m32 != true { response13 <- y } if m31 != true && m32 == true { response23 <- y } } close(response13) close(response23) } func main() { command12 := make(chan int, bufferSize) response12 := make(chan int, bufferSize) command13 := make(chan int, bufferSize) response13 := make(chan int, bufferSize) command23 := make(chan int, bufferSize) response23 := make(chan int, bufferSize) go Routine1(command12, response12, command13, response13) go Routine2(command12, response12, command23, response23) Routine3(command13, response13, command23, response23) // 为了防止 main 函数过早退出，可以等待一段时间 // 或者使用 sync.WaitGroup 等待所有 Goroutine 完成 runtime.Gosched() runtime.Gosched() runtime.Gosched() runtime.Gosched() runtime.Gosched() }
登录后复制
注意事项：
- 修改后的代码仍然可能存在非确定性行为，因为 Goroutine 的执行顺序是不确定的。
- 缓冲 Channel 的大小需要根据实际情况进行调整。过小的缓冲可能无法完全避免死锁，过大的缓冲可能会浪费内存。
- runtime.Gosched() 的使用应该谨慎，过度使用可能会降低程序的性能。
- 在实际开发中，应该尽量避免编写复杂的并发程序，并使用更高级的并发模式来简化代码。
总结
Go 并发程序中的死锁是一个常见的问题，但通过理解死锁的原因和掌握调试和修复策略，我们可以有效地避免死锁。在设计并发程序时，应该尽量避免复杂的 Channel 交互和循环等待，并使用更高级的并发模式来简化代码。同时，需要注意并发程序的非确定性行为，并进行充分的测试，以确保程序的正确性和稳定性。
以上就是Go 并发程序死锁排查与避免：深入剖析与实践的详细内容，更多请关注php中文网其它相关文章！
微信扫一扫打赏
支付宝扫一扫打赏
