Implementing a Concurrent Task Queue in Go

Table of Contents

  1. Introduction
  2. Prerequisites
  3. Setting Up the Project
  4. Implementing the Task Queue - Creating the Task Struct - Creating the Task Queue Struct - Adding Tasks to the Queue - Processing Tasks Concurrently
  5. Running and Testing the Concurrent Task Queue
  6. Recap and Conclusion

Introduction

In this tutorial, we will learn how to implement a concurrent task queue in Go. A task queue allows us to schedule and process tasks concurrently, optimizing the usage of system resources and improving overall efficiency.

By the end of this tutorial, you will understand the concepts behind concurrent task queues and be able to implement your own in Go. We will go step by step, starting with the basics and gradually building upon them to create a fully functional concurrent task queue.

Prerequisites

Before starting this tutorial, you should have a basic understanding of the Go programming language, including how to define structs, write functions, and use goroutines and channels for concurrency. Additionally, make sure you have Go installed on your machine.

Setting Up the Project

To begin, let’s set up a new Go module for our project. Open your terminal and create a new directory:

mkdir concurrent-task-queue
cd concurrent-task-queue

Initialize a new Go module:

go mod init github.com/your-username/concurrent-task-queue

Implementing the Task Queue

Creating the Task Struct

A task in our task queue will be represented by a struct that encapsulates the work to be done. Let’s create a new file called task.go and define the Task struct:

package main

import (
	"fmt"
	"time"
)

type Task struct {
	ID      int
	Payload string
}

func (t *Task) Execute() {
	// Simulating task execution time
	time.Sleep(1 * time.Second)
	fmt.Printf("Task %d executed with payload: %s\n", t.ID, t.Payload)
}

The Task struct has two fields: ID to uniquely identify the task, and Payload to store any data associated with the task. We also define an Execute method that simulates the execution of the task by sleeping for 1 second and printing a message.

Creating the Task Queue Struct

Next, let’s create a new file called task_queue.go to define the TaskQueue struct, which will manage the tasks and the execution flow:

package main

type TaskQueue struct {
	Queue chan *Task
	Stop  chan bool
}

func NewTaskQueue() *TaskQueue {
	return &TaskQueue{
		Queue: make(chan *Task),
		Stop:  make(chan bool),
	}
}

func (q *TaskQueue) AddTask(task *Task) {
	q.Queue <- task
}

func (q *TaskQueue) Start(workerCount int) {
	for i := 0; i < workerCount; i++ {
		go q.processTasks()
	}
}

func (q *TaskQueue) StopProcessing() {
	q.Stop <- true
	close(q.Queue)
	close(q.Stop)
}

func (q *TaskQueue) processTasks() {
	for {
		select {
		case task := <-q.Queue:
			task.Execute()
		case <-q.Stop:
			return
		}
	}
}

The TaskQueue struct contains two channels: Queue to store incoming tasks and Stop to signal the workers to stop processing tasks. We define a NewTaskQueue function to create a new instance of TaskQueue and initialize the channels.

The AddTask method is used to add a new task to the queue. The Start method starts the task processing by spawning a fixed number of goroutines, each running the processTasks method. The StopProcessing method stops the processing of tasks by closing the channels.

The processTasks method runs in an infinite loop and uses a select statement to handle tasks from the Queue channel and the stop signal from the Stop channel. When a task is received, it is executed by calling the Execute method on the task. If a stop signal is received, the goroutine terminates.

Adding Tasks to the Queue

Now that we have our task queue implementation ready, let’s create a new file called main.go to demonstrate how to add tasks to the queue and initiate the processing:

package main

func main() {
	taskQueue := NewTaskQueue()
	taskQueue.Start(3)

	// Add tasks to the queue
	taskQueue.AddTask(&Task{ID: 1, Payload: "Task 1"})
	taskQueue.AddTask(&Task{ID: 2, Payload: "Task 2"})
	taskQueue.AddTask(&Task{ID: 3, Payload: "Task 3"})
	taskQueue.AddTask(&Task{ID: 4, Payload: "Task 4"})
	taskQueue.AddTask(&Task{ID: 5, Payload: "Task 5"})

	// Wait for task processing to complete
	time.Sleep(3 * time.Second)

	// Stop task processing
	taskQueue.StopProcessing()
}

In the main function, we create a new instance of the TaskQueue using NewTaskQueue and start three workers to process the tasks concurrently by calling Start(3).

We then add five tasks to the queue using the AddTask method. Finally, we wait for 3 seconds to allow the tasks to be processed and call StopProcessing to stop the processing of tasks.

Running and Testing the Concurrent Task Queue

To test our concurrent task queue, run the following command in your terminal:

go run main.go task.go task_queue.go

You should see the tasks being executed by the workers concurrently. The output might be in a different order due to the concurrent nature of the processing. Here’s an example of the expected output:

Task 1 executed with payload: Task 1
Task 2 executed with payload: Task 2
Task 5 executed with payload: Task 5
Task 3 executed with payload: Task 3
Task 4 executed with payload: Task 4

Recap and Conclusion

In this tutorial, we learned how to implement a concurrent task queue in Go. We started by defining the Task struct and the TaskQueue struct, which manages the tasks and the execution flow. We then demonstrated how to add tasks to the queue and process them concurrently.

You should now have a good understanding of how to implement a concurrent task queue in Go and leverage goroutines and channels for concurrent programming. You can further extend this implementation by adding error handling, task priority, or other features based on your specific requirements.

Remember to practice and experiment with the code to deepen your understanding. Happy coding!

Published:

Related Articles