Controlling Data Flow with Backpressure using Buffered Channels
Owner: SnippetBot
Created: 2026-08-02 00:00:46
Size: 1.76 KB
Expires: Never
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
package main
import (
"fmt"
"time"
)
// producer generates data and sends it to the channel.
// It will block if the channel is full, applying backpressure.
func producer(dataCh chan<- int, stopCh <-chan struct{}) {
i := 0
for {
select {
case <-stopCh:
fmt.Println("Producer: Stopping.")
return
case dataCh <- i:
fmt.Printf("Producer: Sent %d
", i)
i++
time.Sleep(50 * time.Millisecond) // Simulate work/data generation
}
}
}
// consumer receives data from the channel and processes it slowly.
func consumer(dataCh <-chan int, doneCh chan<- struct{}) {
for data := range dataCh {
fmt.Printf("Consumer: Received %d, processing...
", data)
time.Sleep(200 * time.Millisecond) // Simulate slow processing
}
fmt.Println("Consumer: Data channel closed. Shutting down.")
close(doneCh) // Signal that consumer is done
}
func main() {
// Create a buffered channel for data with capacity 3.
// This buffer size dictates how much data can be "in flight" before
// the producer blocks, creating backpressure.
dataChannel := make(chan int, 3)
stopProducer := make(chan struct{})
consumerDone := make(chan struct{})
go producer(dataChannel, stopProducer)
go consumer(dataChannel, consumerDone)
// Let the system run for a while
time.Sleep(2 * time.Second)
// Signal producer to stop
close(stopProducer)
// Wait for producer to finish sending remaining data and then close the data channel
// A more robust solution might involve waiting for all items to be processed
// before closing the data channel. For this example, we simply wait for a bit.
time.Sleep(500 * time.Millisecond)
close(dataChannel)
// Wait for consumer to finish processing all data and signal its completion
<-consumerDone
fmt.Println("Main: All operations complete. Exiting.")
}