When looking through some Go code I found the following:
ch := make(chan int)
I looked up in a online tutorial how Go Channels work:
https://tour.golang.org/concurrency/2
But I find this example unclear.
Can someone give me a easy explanation and an example of the use of channels?
chan is a channel in Golang. In simple word you can think it as a box in which you put a item at one end and then pick it from other end.
Unbuffered Channels
Buffered Channel
This is the small code I have written for you to understand channels. Now change order of go routines and see the outputs. Each time output may differ.
package main
import (
"fmt"
"time"
)
func main() {
messages := make(chan int)
go func() {
time.Sleep(time.Second * 3)
messages <- 1
}()
go func() {
time.Sleep(time.Second * 2)
messages <- 2
}()
go func() {
time.Sleep(time.Second * 1)
messages <- 3
}()
go func() {
for i := range messages {
fmt.Println(i)
}
}()
go func() {
time.Sleep(time.Second * 1)
messages <- 4
}()
go func() {
time.Sleep(time.Second * 1)
messages <- 5
}()
time.Sleep(time.Second * 5)
}
For best understanding visit this blog where go routines and channels are described in GUI.
Visit http://divan.github.io/posts/go_concurrency_visualize/
I think the spec is pretty clear on this. Spec: Channel types:
A channel provides a mechanism for concurrently executing functions to communicate by sending and receiving values of a specified element type.
When you have multiple goroutines which are executed concurrently, channels provide the easiest way to allow the goroutines to communicate with each other.
One way for communication would be via a "shared" variable which is visible to both goroutines, but that would require proper locking / synchronized access.
Instead, Go favors channels. Quoting from Effective Go: Share by communicating:
Do not communicate by sharing memory; instead, share memory by communicating.
So instead of putting messages into a shared slice for example, you can create a channel (visible to both goroutines), and without any external synchronization / locking, one goroutine can send messages (values) via the channel, and the other goroutine can receive them.
Only one goroutine has access to the value at any given time. Data races cannot occur, by design.
So in fact, any number of goroutines can send values on the same channel, and any number of goroutines can receive values from it, still without any further synchronization. See related question for more details: If I am using channels properly should I need to use mutexes?
Let's see an example where we start 2 additional goroutines for concurrent computation purposes. We pass a number to the first, which adds 1 to it, and delivers the result on a 2nd channel. The 2nd goroutine will receive a number, multiply it by 10 and deliver it to the result channel:
func AddOne(ch chan<- int, i int) {
i++
ch <- i
}
func MulBy10(ch <-chan int, resch chan<- int) {
i := <-ch
i *= 10
resch <- i
}
This is how it can be called / used:
func main() {
ch := make(chan int)
resch := make(chan int)
go AddOne(ch, 9)
go MulBy10(ch, resch)
result := <-resch
fmt.Println("Result:", result)
}
Communicating via channels also takes care of goroutines waiting for each other. In this example it means MulBy10() will wait until AddOne() delivers the incremented number, and main() will wait for MulBy10() before printing the result. Output as expected (try it on the Go Playground):
Result: 100
There are a number of language constructs which are for easy use of channels, for example:
for ... range on a channel loops over values received from a channel, until the channel is closed.select statement can be used to list multiple channel operations such as send on a channel and receive from a channel, and the one that can proceed without blocking will be selected (randomly if there are multiple ones that can proceed; and will block if none is ready).v, ok := <-chlen() function tells the number of elements queued (unread); the builting cap() function returns the channel buffer capacity.For a more practical example, see how channels can be used to implement a worker pool. A similar use is distributing values from a producer to consumer(s).
Another practical example is to implement a memory pool using buffered channels.
And yet another practical example is an elegant implementation of a broker.
A channel is often used to timeout some blocking operation, utilizing a channel returned by time.After() which "fires" after the specified delay / duration ("fires" means a value will be sent on it). See this example for demonstration (try it on the Go Playground):
ch := make(chan int)
select {
case i := <-ch:
fmt.Println("Received:", i)
case <-time.After(time.Second):
fmt.Println("Timeout, no value received")
}
It can be used to wait for a maximum time amount for some value, but if other goroutines cannot supply the value by that time, we may decide to do something else instead.
Also a special form of communication may be just to signal completion of some operation (without actually sending any "useful" data). Such case may be implemented by a channel with any element type, e.g. chan int, and sending any value on it, e.g. 0. But since the sent value holds no information, you may declare it like chan struct{}. Or even better, if you only need a one-time signalling, you may just close the channel which can be intercepted on the other side using for ... range, or receiving from it (as receiving from a closed channel proceeds immediately, yielding the zero value of the element type). Also know that even though a channel may be used for this kind of signalling, there's a better alternative for this: sync.WaitGroup.
It's worth knowing about the channel axioms to avoid suprising behaviour: How does a non initialized channel behave?
The Go Blog: Share Memory By Communicating
The Go Blog: Go Concurrency Patterns: Pipelines and cancellation
The concept is very similar to something that's been in Unix/Linux since the beginning: pipes.
These are a reliable inter-process / inter-thread communication facility that is built-in to the language. Very convenient.
Use channels if you want goroutines to signal each other. There are multiple reasons you might want to want this signalling.
Use channels if you want goroutines to communicate. This could be with or without data.
package main
/*
Simulation for sending messages from threads for processing,
and getting a response (processing result) to the thread
*/
import (
"context"
"fmt"
"math/rand"
"sync"
"time"
)
type (
TMsg struct { // message
name string // thread name, owner name
backId int // index of response Back channel, or -1
msg string // message
note string // comment
}
TTh struct { // for thread
name string // thread name
jobId int // index of central Job chanel
backId int // index of response Back channel, or -1
}
TChans map[int]chan TMsg
)
var gChans TChans //global variable, all channels map
func main() {
gChans = make(TChans) // all channels map
jobIndex, job := NewChan() // chanel for send mesage to central Job (from threads)
_, worker := NewChan() // channel for send message to Worker (from Job receiver)
for i := 1; i <= 5; i++ { // 5 threads
backIndex, _ := NewChan() // channel index for response back the thread
go ping(TTh{name: fmt.Sprint(i) + "th", jobId: jobIndex, backId: backIndex}) //start threads
}
//go Job(job, worker) // central receiver and start workers
Job(job, worker) // central receiver and start workers
for LenChan() > 2 { // 2 = job and worker channels
SleepM(5000)
}
}
func Job(job, worker chan TMsg) { //central receiver
var v TMsg
ctx := context.Background()
ctx, cancelWorkers := context.WithCancel(ctx)
for i := 1; i <= 3; i++ { // start workers , sql simulation
go Worker(i, worker, ctx)
}
for {
select {
case v = <-job: // receive message
if v.note == "sql" { // sql simulation
worker <- v
} else {
if v.note == "end" {
FreeChan(v.backId)
fmt.Println(v.name, "FREE")
} else {
ch := GetChan(v.backId)
if ch != nil {
ch <- TMsg{name: v.name, backId: v.backId, msg: v.msg, note: "receiver"}
}
}
}
default:
if LenChan() <= 2 {
cancelWorkers() // cancel workers
return
} else {
SleepM(2)
}
}
}
}
func Worker(id int, worker <-chan TMsg, ctx context.Context) { // simulate sql or auther process
var v TMsg
for {
select {
case v = <-worker:
{
SleepM(rand.Intn(50))
v.note = "worker:" + fmt.Sprint(id)
ch := GetChan(v.backId)
if ch != nil {
ch <- v
}
}
case <-ctx.Done(): //return
default:
{
//fmt.Println("worker", id)
SleepM(2)
}
}
}
}
func waitResponse(d chan TMsg, pTimeout int) (bool, TMsg) {
var v TMsg
for {
select {
case v = <-d:
return true, v
case <-time.After(time.Duration(pTimeout) * time.Second):
return false, v
}
}
}
func ping(pTh TTh) {
SleepM(10)
var v TMsg
ok := true
i := 0
job := GetChan(pTh.jobId) // central Job receiver chanel
back := GetChan(pTh.backId) // response Back channel
for i < 50 {
if ok {
ok = false
job <- TMsg{name: pTh.name, backId: pTh.backId, msg: fmt.Sprint(i), note: "sql"}
i++
}
if back != nil {
if !ok {
ok, v = waitResponse(back, 10) //with timeout 10 sec
if ok {
fmt.Println(v.name, "msg:", v.msg, v.note)
SleepM(1)
} else {
fmt.Println(pTh.name, "response timeout")
}
}
} else {
SleepM(1)
}
}
fmt.Println(v.name, "---- end ----")
v.note = "end"
job <- v
}
func NewChan() (int, chan TMsg) {
mux := &sync.RWMutex{}
mux.Lock()
defer mux.Unlock()
index := len(gChans)
gChans[index] = make(chan TMsg)
return index, gChans[index]
}
func GetChan(pIndex int) chan TMsg {
mux := &sync.RWMutex{}
mux.Lock()
defer mux.Unlock()
ch, ok := gChans[pIndex]
if ok {
return ch
} else {
return nil
}
}
func LenChan() int {
return len(gChans)
}
func FreeChan(pIndex int) bool {
ch := GetChan(pIndex)
if ch != nil {
mux := &sync.RWMutex{}
mux.Lock()
defer mux.Unlock()
close(gChans[pIndex]) //close channel
gChans[pIndex] = nil
delete(gChans, pIndex)
return true
} else {
return false
}
}
func SleepM(pMilliSec int) { // sleep millisecounds
time.Sleep(time.Duration(pMilliSec) * time.Millisecond)
}
