golang pipeline implementation queue

Overview

As a popular programming language in the industry, Golang has the advantages of lightweight, concurrency safety, built-in GC, fast compilation, etc., and is widely used in cloud computing, Web, web crawlers and other fields. . Golang’s efficient concurrency model is one of the reasons why Golang is so popular. The pipeline mechanism is one of the three communication methods of Golang's concurrency mechanism. Pipes are divided into unbuffered pipes and buffered pipes.

In Golang’s concurrency model, pipelines are usually used to implement the communication mechanism between producers and consumers. As producers flood in data, consumers can get the data from the pipeline and process it. In this model, pipes act as queues. Therefore, Golang's pipeline mechanism is also suitable for queue implementation.

This article will introduce how to use Golang’s pipeline mechanism to implement queues. Specifically, we'll write a buffered queue that supports concurrency and briefly show how to implement a bounded queue using an unbuffered pipe.

Queue with buffered pipe

Queue with buffered pipe allows producers/consumers to still work normally when the speed of production/consumption is inconsistent. It has a fixed size, the producer will be blocked when the queue is full and the consumer will be blocked when the queue is empty. In Golang, we can use the make() function to create buffered pipes.

The following is a simple implementation example:

package main

import "fmt"

type Queue struct {
    // 声明管道
    items chan int
    // 声明队列最大容量
    capacity int
}

func NewQueue(capacity int) *Queue {
    return &Queue{make(chan int, capacity), capacity}
}

func (q *Queue) Enqueue(item int) {
    q.items <- item
}

func (q *Queue) Dequeue() int {
    return <-q.items
}

func main() {
    q := NewQueue(3)

    q.Enqueue(1)
    q.Enqueue(2)
    q.Enqueue(3)

    fmt.Println(q.Dequeue()) // 1
    fmt.Println(q.Dequeue()) // 2
    fmt.Println(q.Dequeue()) // 3
}

In the above code, we use a structure to represent the queue, which contains a pipe and the maximum capacity of the queue. The NewQueue() function is used to create a queue with a specified maximum capacity. In the Enqueue() function, we write data to the pipe and block if the pipe is full. In the Dequeue() function, we read data from the pipe and block if the pipe is empty. In the main() function, we create a queue with a maximum capacity of 3 and add three elements 1, 2, and 3 to the queue. Then, the Dequeue() function is called in sequence to obtain elements from the queue and output them to the console.

Unbuffered pipes to implement bounded queues

In Golang, using unbuffered pipes to implement bounded queues requires the use of the select statement mechanism. We can use the default statement in the select statement to handle the blocking situation when the queue is full or empty.

The following is an example of using an unbuffered pipe to implement a bounded queue:

package main

import (
    "fmt"
    "math/rand"
)

type Queue struct {
    items chan int
}

func NewQueue() *Queue {
    return &Queue{make(chan int)}
}

func (q *Queue) Enqueue(item int) {
    select {
    case q.items <- item:
    default:
        <-q.items
        q.items <- item
    }
}

func (q *Queue) Dequeue() int {
    select {
    case item := <-q.items:
        return item
    default:
        return -1
    }
}

func main() {
    q := NewQueue()

    for i := 0; i < 10; i++ {
        go func() {
            q.Enqueue(rand.Intn(100))
        }()

        go func() {
            fmt.Println(q.Dequeue())
        }()
    }
}

In the above code, we also use a structure to represent a bounded queue. Unlike buffered pipes, we do not pass in the maximum capacity of the queue when creating the pipe. In the Enqueue() function, we use the select statement to insert elements when the pipe is not full; if the pipe is full, we use the default condition, which first removes the first element in the current queue from the pipe, and then adds New elements are inserted. The Dequeue() function also uses the select statement to return the first element in the queue when the pipe is not empty; if the pipe is empty, the default case is used and -1 is returned.

In the main() function, we insert 10 elements into the queue and use 10 coroutines to dequeue the elements in the queue respectively. We can see that since the capacity of the queue is 1, the Enqueue() function continuously inserts elements into the queue, while the Dequeue() function continuously removes elements when the queue is not empty. Therefore, the output is a series of random integers.

Conclusion

Through the introduction of this article, we can see that it is very simple to implement queues using the Golang pipeline mechanism. For a queue with a buffered pipe, its maximum capacity can be specified directly in the make() function, while for an unbuffered pipe to implement a bounded queue, you need to use the select statement mechanism. Due to the advantages of Golang's concurrency model, it is most efficient to use the Golang pipeline mechanism to implement queues.

The above is the detailed content of golang pipeline implementation queue. For more information, please follow other related articles on the PHP Chinese website!