Go with buffered chan implements message queue function

1. Channels definition

A channel is a pipeline that supports multiple types through which you can use the channel operator <- to send and receive values.

Data flows in the direction of the arrow.

ch <- v    // Send v to channel ch.
v := <-ch  // Receive from ch, and
           // assign value to v.

Like maps and slices, channels must be created before use:

ch := make(chan int)

By default, sending and receiving blocks until the other party is ready.
This allows goroutines to be synchronized without explicit locks or condition variables.

package main

import "fmt"

func sum(s []int, c chan int) {
    sum := 0
    for _, v := range s {
        sum += v
    }
    c <- sum // send sum to c
}

func main() {
    s := []int{7, 2, 8, -9, 4, 0}

    c := make(chan int)
    go sum(s[:len(s)/2], c)
    go sum(s[len(s)/2:], c)
    x, y := <-c, <-c // receive from c

    (x, y, x+y)
}

2. chan Common operations

No buffer: Save and read once, and it is not fetched after being stored. If it is stored again, it will be blocked. Also, if it is not saved, it will be blocked if it is fetched.
Buffer: Only when the buffer is full will it be blocked; only when the buffer is empty will it be blocked.
len(channel) Returns the existing data length of the buffer
cap(channel) Returns the size of the buffer
close(channel) Close the channel, and after closing, the data cannot be read. As follows, if other coroutines turn off the channel, the loop will be jumped

3. Use buffered chan to implement message queue function

// Monitor data structuretype Msg struct {
    Timestamp    int64
    Content        string        
}

// Use chan to simulate a queue, the elements of the queue are of type Msgvar SyncQueen chan Msg

// Must be initialized to use.  Initialize a chan with a capacity of 1024.  chan will block when fullfunc init() {
    SyncQueen = make(chan Msg, 1024)
}

// Queue Consumerfunc Consumer() {
    defer func() {
        if err := recover(); err != nil {
            (err)
        }
    }()

    for {
        // If there is no message in chan, it will block        msg := &lt;-SyncQueen

        ()
    }
}

// Queue Producerfunc Producer() {
    for {
        msg := Msg(().Unix(), "hello")
        
        // Send a message to chan        SyncQueen &lt;- msg
        (2 )
    }
}

Focus

Multi-coroutines are concurrently safe to use chan. Here is a simple example:

// Define chan of type intvar chanNums chan int

// chan consumers, users follow-up multiple coroutines// Purpose: 10,000 numbers are stored in the array, and after multiple coroutines are calculated in parallel, the sum is added upfunc consumer(sum *int) int {
    for {
        v := &lt;-chanNums
        *sum += v
    }
}

//-------------------------------------

func main() {
    var a [10000]int

    for i := 0; i &lt; 10000; i++ {
        a[i] = i + 1
    }

    chanNums = make(chan int, 10000)

    for i := 0; i &lt; 10000; i++ {
        chanNums &lt;- (i + 1)
    }
    var s1, s2, s3, s4, s5 int = 0, 0, 0, 0, 0
    go consumer(&amp;s1)
    go consumer(&amp;s2)
    go consumer(&amp;s3)
    go consumer(&amp;s4)
    go consumer(&amp;s5)

    for {
        (5 * )
        break
    }

    ("s1=", s1, "s2=", s2, "s3=", s3, "s4=", s4, "s5=", s5)
    ("sum=", s1+s2+s3+s4+s5)
}

// Outputs1= 10818438 s2= 12073966 s3= 9044041 s4= 11509634 s5= 6558921
sum= 50005000

This is the end of this article about Go buffered chan implementing the message queue function. For more related Go buffered chan message queue content, please search for my previous articles or continue browsing the related articles below. I hope everyone will support me in the future!