The core concept of concurrent programming is synchronous communication.
1.sync.Mutex achieves synchronization:
package main
import (
"fmt"
"sync"
)
func main() {
var mu sync.Mutex
go func() {
fmt.Println("hello world")
mu.Lock()
}()
mu.Unlock()
}
Execution result:
Because the main thread and the created coroutine are not in the same Goroutine, the lock release operation cannot be performed. Therefore, an exception is thrown.
2.sync.Mutex implements synchronization (repair):
package main
import (
"fmt"
"sync"
)
func main() {
var mu sync.Mutex
mu.Lock()
go func() {
fmt.Println("hello world")
mu.Unlock()
}()
mu.Lock()
}
Execution result:
The way to fix it is to perform two locking operations in the main function. The second locking will be blocked because the lock is already occupied (not a recursive lock). The main thread will wait for the execution of the coroutine. When the coroutine execution completes the logic of releasing the lock, unlocking will also unblock the main thread.
3. No cache channel to achieve synchronization:
package main
import "fmt"
func main() {
done := make(chan int)
go func() {
fmt.Println("Hello,world")
<-done
}()
done <- 1
}
Execution result:
According to the Go language memory model specification, reception from an unbuffered channel occurs before the send to the channel is completed.
4. No cache channel limit size:
package main
import "fmt"
func main() {
done := make(chan int, 1)
go func() {
fmt.Println("Hello,world")
<-done
}()
done <- 1
}
5. The cache channel size is N:
package main
import (
"fmt"
"time"
)
func main() {
done := make(chan int, 10)
for i := 0; i < cap(done); i++ {
go func() {
fmt.Println("Hello,worldi")
}()
done <- 1
}
for i := 0; i < cap(done); i++ {
<-done
}
//Block main thread.see print results.
time.Sleep(1 * time.Second)
}
Execution result:
6.sync.WaitGroup synchronization:
package main
import (
"fmt"
"sync"
)
func main() {
var wg sync.WaitGroup
for i := 0; i < 10; i++ {
wg.Add(1)
go func() {
fmt.Println("Hello,world", i)
wg.Done()
}()
}
wg.Wait()
}
Execution result:
wg.Add(1) is used to increase the number of waiting events, and must be ensured to be executed before the background thread is started. Calling wg.done indicates the completion of this event. wg.Wait indicates the completion of the waiting event.
7. Producer-consumer model:
package main
import (
"fmt"
"time"
)
func main() {
ch := make(chan int, 64)
go Producer(3, ch)
go Producer(5, ch)
go Consumer(ch)
//Exit after running for a certain time.
time.Sleep(5 * time.Second)
}
// producer.
func Producer(factor int, out chan int) {
for i := 0; ; i++ {
out <- i * factor
}
}
// consumer.
func Consumer(in <-chan int) {
for v := range in {
fmt.Println(v)
}
}
8. Optimize the producer-consumer model:
package main
import (
"fmt"
"os"
"os/signal"
"syscall"
)
func main() {
ch := make(chan int, 64)
go Producer(3, ch)
go Consumer(ch)
sig := make(chan os.Signal, 1)
signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
fmt.Printf("quit (%v)\n", <-sig)
}
// producer.
func Producer(factor int, out chan int) {
for i := 0; ; i++ {
out <- i * factor
}
}
// consumer.
func Consumer(in <-chan int) {
for v := range in {
fmt.Println(v)
}
}
9. Publish and subscribe model:
package pubsub
import (
"sync"
"time"
)
type (
//Subscriber is a channel.
subscriber chan interface{}
//Topic as a filter.
topicFunc func(v interface{}) bool
)
// publisher object.
type Publisher struct {
//read-write lock.
m sync.Mutex
//Cache size of subscription queue.
buffer int
//Post timeout.
timeOut time.Duration
//Subscription information
subscribers map[subscriber]topicFunc
}
// Construct a publisher object.You can set the publishing timeout and the length of the cache queue.
func NewPublisher(publishTimeOut time.Duration, buff int) *Publisher {
return &Publisher{
buffer: buff,
timeOut: publishTimeOut,
subscribers: make(map[subscriber]topicFunc),
}
}
// Add a new subscriber,Subscribe to all topics.
func (p *Publisher) Subscribe() chan interface{} {
return p.SubscribeTopic(nil)
}
// Add a new subscriber.Subscribe to topics filtered by filters.
func (p *Publisher) SubscribeTopic(topicFunc topicFunc) chan interface{} {
ch := make(chan interface{}, p.buffer)
p.m.Lock()
p.subscribers[ch] = topicFunc
p.m.Unlock()
return ch
}
// Unsubscribe.
func (p *Publisher) Evict(sub chan interface{}) {
p.m.Lock()
defer p.m.Unlock()
delete(p.subscribers, sub)
close(sub)
}
// Post a topic.
func (p *Publisher) Publish(v interface{}) {
p.m.Lock()
defer p.m.Unlock()
var wg sync.WaitGroup
for sub, topic := range p.subscribers {
wg.Add(1)
go p.sendTopic(sub, topic, v, &wg)
}
wg.Wait()
}
// Send topic.
func (p *Publisher) sendTopic(sub subscriber, topic topicFunc, v interface{}, s *sync.WaitGroup) {
defer s.Done()
if topic != nil && !topic(v) {
return
}
select {
case sub <- v:
case <-time.After(p.timeOut):
}
}
func (p *Publisher) Close() {
p.m.Lock()
defer p.m.Unlock()
for sub := range p.subscribers {
delete(p.subscribers, sub)
close(sub)
}
}
package main
import (
"fmt"
"gomodule/pubsub"
"strings"
"time"
)
func main() {
publisher := pubsub.NewPublisher(100*time.Millisecond, 10)
defer publisher.Close()
all := publisher.Subscribe()
golang := publisher.SubscribeTopic(func(v interface{}) bool {
if s, ok := v.(string); ok {
return strings.Contains(s, "golang")
}
return false
})
publisher.Publish("hello world")
publisher.Publish("golang")
go func() {
for msg := range all {
fmt.Println(msg)
}
}()
go func() {
for msg := range golang {
fmt.Println(msg)
}
}()
//Exit after running for a certain time.
time.Sleep(3 * time.Second)
}