Concurrency
## Learning Objectives
- Understand goroutines
- Work with channels
- Master channel operations
- Use select statements
- Handle race conditions
- Implement common patterns
## Goroutines
### What is a Goroutine?
A goroutine is a lightweight thread managed by the Go runtime.
### Basic Syntax
```go
go f(x, y, z) // Starts new goroutine
```
### Example
```go
func main() {
go sayHello()
fmt.Println("Hello from main")
time.Sleep(time.Second) // Wait for goroutine
}
func sayHello() {
fmt.Println("Hello from goroutine")
}
```
### Anonymous Function
```go
go func() {
fmt.Println("Running in goroutine")
}()
time.Sleep(time.Second)
```
## Channels
### Declaration
```go
ch := make(chan int) // Unbuffered
ch := make(chan int, 10) // Buffered with capacity 10
```
### Send and Receive
```go
ch := make(chan int)
// Send
ch <- 42
// Receive
value := <-ch
```
### Channel Operations
```go
ch := make(chan string, 2)
ch <- "Hello" // Send
ch <- "World" // Send
msg1 := <-ch // Receive
msg2 := <-ch // Receive
```
### Closing
```go
ch := make(chan int)
go func() {
ch <- 1
ch <- 2
close(ch)
}()
for v := range ch {
fmt.Println(v)
}
```
## Directional Channels
### Specification
```go
chan T // Can send and receive
chan<- T // Send only
<-chan T // Receive only
```
### Use Cases
```go
// Producer: can only send
func producer(ch chan<- int) {
ch <- 42
}
// Consumer: can only receive
func consumer(ch <-chan int) {
value := <-ch
fmt.Println(value)
}
```
## Select Statement
### Basic select
```go
select {
case msg1 := <-ch1:
fmt.Println("Received from ch1:", msg1)
case msg2 := <-ch2:
fmt.Println("Received from ch2:", msg2)
case sendVal := <-ch3:
fmt.Println("Ready to send:", sendVal)
default:
fmt.Println("No communication")
}
```
### Waiting on Multiple Channels
```go
select {
case msg := <-ch1:
fmt.Println("ch1:", msg)
case msg := <-ch2:
fmt.Println("ch2:", msg)
case <-time.After(time.Second):
fmt.Println("Timeout")
}
```
### Non-blocking Communication
```go
select {
case msg := <-ch:
fmt.Println("Received:", msg)
default:
fmt.Println("No message ready")
}
```
## Channel Patterns
### Pipeline
```go
func generate(nums ...int) <-chan int {
out := make(chan int)
go func() {
for _, n := range nums {
out <- n
}
close(out)
}()
return out
}
func square(in <-chan int) <-chan int {
out := make(chan int)
go func() {
for n := range in {
out <- n * n
}
close(out)
}()
return out
}
func main() {
for n := range square(square(generate(1, 2, 3, 4, 5))) {
fmt.Println(n)
}
}
```
### Fan-out, Fan-in
```go
func merge(channels ...<-chan int) <-chan int {
out := make(chan int)
var wg sync.WaitGroup
for _, ch := range channels {
wg.Add(1)
go func(c <-chan int) {
for v := range c {
out <- v
}
wg.Done()
}(ch)
}
go func() {
wg.Wait()
close(out)
}()
return out
}
```
### Worker Pool
```go
func worker(id int, jobs <-chan int, results chan<- int) {
for j := range jobs {
fmt.Printf("worker %d processing job %d\n", id, j)
results <- j * 2
}
}
func main() {
jobs := make(chan int, 100)
results := make(chan int, 100)
for w := 1; w <= 3; w++ {
go worker(w, jobs, results)
}
for j := 1; j <= 5; j++ {
jobs <- j
}
close(jobs)
for a := 1; a <= 5; a++ {
<-results
}
}
```
## Race Conditions
### Problem
```go
var counter int
func increment() {
counter++ // Not atomic!
}
func main() {
for i := 0; i < 1000; i++ {
go increment()
}
time.Sleep(time.Second)
fmt.Println(counter) // Likely not 1000
}
```
### Mutex Solution
```go
import "sync"
var (
counter int
mu sync.Mutex
)
func increment() {
mu.Lock()
defer mu.Unlock()
counter++
}
func main() {
var wg sync.WaitGroup
for i := 0; i < 1000; i++ {
wg.Add(1)
go func() {
increment()
wg.Done()
}()
}
wg.Wait()
fmt.Println(counter) // 1000
}
```
### WaitGroup
```go
var wg sync.WaitGroup
for i := 0; i < 5; i++ {
wg.Add(1)
go func() {
defer wg.Done()
doWork(i)
}()
}
wg.Wait() // Block until all done
```
## sync Package
### Mutex
```go
varmu sync.Mutex
func criticalSection() {
mu.Lock()
defer mu.Unlock()
// Protected code
}
```
### RWMutex
```go
var (
mu sync.RWMutex
data map[string]int
)
func read(key string) int {
mu.RLock()
defer mu.RUnlock()
return data[key]
}
func write(key string, value int) {
mu.Lock()
defer mu.Unlock()
data[key] = value
}
```
### Once
```go
var (
once sync.Once
single *Config
)
func getConfig() *Config {
once.Do(func() {
single = &Config{}
})
return single
}
```
### Map (Concurrent)
```go
var syncMap sync.Map
syncMap.Store("key", "value")
value, ok := syncMap.Load("key")
syncMap.Delete("key")
syncMap.Range(func(k, v interface{}) bool {
fmt.Printf("%s: %s\n", k, v)
return true
})
```
### Pool
```go
pool := sync.Pool{
New: func() interface{} {
return make([]byte, 1024)
},
}
buf := pool.Get().([]byte)
defer pool.Put(buf)
```
## Context Package
### WithCancel
```go
import "context"
ctx, cancel := context.WithCancel(context.Background())
go func() {
time.Sleep(2 * time.Second)
cancel()
}()
select {
case <-ctx.Done():
fmt.Println("Cancelled!")
case <-time.After(3 * time.Second):
fmt.Println("Timed out")
}
```
### WithTimeout
```go
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
select {
case <-ctx.Done():
fmt.Println("Timed out")
default:
fmt.Println("Proceeding...")
}
```
### WithValue
```go
ctx := context.WithValue(context.Background(), "user", "alice")
if user, ok := ctx.Value("user").(string); ok {
fmt.Println("User:", user)
}
```
## Common Pitfalls
### Sending to Closed Channel
```go
ch := make(chan int)
close(ch)
ch <- 1 // PANIC: send on closed channel
```
### Nil Channel
```go
var ch chan int // nil channel
<-ch // Blocks forever
ch <- 1 // Blocks forever
```
### Deadlock
```go
ch := make(chan int)
ch <- 1 // Blocks (no receiver)
<-ch // Never reaches here
```
## Summary
- Goroutines: lightweight threads via `go` keyword
- Channels: typed pipes for communication
- Buffered channels: `make(chan T, capacity)`
- Unbuffered channels: `make(chan T)` (sends block until received)
- `select`: wait on multiple channels
- `sync.Mutex`, `sync.RWMutex` for mutual exclusion
- `sync.WaitGroup` for waiting on goroutines
- `sync.Map` for concurrent map access
- `context` for cancellation and timeouts
- Never close from receiver; close only from sender
Comments
Comments powered by Giscus
To enable comments, add your Giscus embed code here.
Learn more about Giscus →