高效的图像处理：Golang、Asynq、Redis 和 Fiber 用于异步队列处理

介绍

在这篇简短的文章中，我将解释一种加速 Web 应用程序的常用方法。它涉及将任务从主线程中移开并将它们放在队列中以进行异步处理，使用队列单独组织和处理这些任务。

在服务器端应用程序中，有效处理图像处理等资源密集型任务是一项重大挑战。这些操作通常占用大量计算量，可能会使主线程陷入困境，从而导致性能下降和可伸缩性差。

眼前的挑战

在服务器端应用程序中，管理使用大量计算机能力的图像处理是一个巨大的挑战。如果操作不当，这些任务可能会使应用程序的主线程过载，使其速度缓慢且让用户感到沮丧。我的目标是创建一个以简单的方式提高性能的解决方案。主要目标是保持应用程序的核心部分平稳运行，即使它正在进行大量图像处理。

使用的工具和先决条件

我选择 Golang 作为这个图像处理平台的主要语言，因为它擅长同时处理多个任务，从而实现快速的图像处理。Golang 有助于提高服务的速度和可靠性。

Fiber 是一个基于 Golang 构建的轻量级 Web 框架，我用它来处理与 Web 相关的任务。其高效的设计确保了无缝的互联网通信和可靠的网络操作。光纤可有效保持流畅的 Web 服务性能。

为了处理后台任务，我使用 Asynq。它安排了这些任务，因此它们不会压倒服务的主要部分。这对于保持服务正常运行而不会出现任何延迟或问题非常重要。

Redis 是一种快速键值存储，我将其用作队列系统，这是一种流行的使用方式。它对 Asynq 包至关重要，因为 Asynq 依赖于它来管理任务。Redis 有助于保持一切井井有条，并确保高效可靠地完成任务。

此外，我还使用 Goland 作为编码和测试的主要 IDE。它是专门为 Go 设计的，对于这个项目来说非常方便。使用 Goland，编码变得简化和高效，促进了 Go 应用程序的开发。

我使用 Postman 来管理和测试 API 请求。它非常适合 API 工作，使工作更轻松、更快捷。

实施解决方案：Asynq 和 Redis

为了实现这一目标，我求助于 Asynq 和 Redis。Async 是一个强大的任务队列，允许我将繁重的图像处理任务从主线程中卸载。这意味着这些任务可以异步处理，确保应用程序的核心功能不受影响。以高性能著称的 Redis 充当了主干，为 Asynq 提供了快速的内存数据存储功能。

图像处理服务概述

我构建了一个服务，使处理图像变得更加容易。它是用一种叫做 Go 的编程语言编写的，它非常适合同时做很多事情而不会混淆。这项服务可以非常快速地更改图像大小，将它们切换到不同的格式，并使它们看起来更好。它可以在不变慢的情况下处理大量工作，这意味着使用此服务的应用程序的主要部分不必自己完成所有艰苦的工作，并且可以平稳运行。

文件夹结构

以下是图像处理服务的文件夹结构：

docker-compose.yaml 此文件包含使用 Docker Compose 运行服务的配置。

go.mod and go.sum 这些文件用于管理 Go 依赖项和版本控制。

handlers 此目录包含用于处理服务不同方面的代码，例如图像处理。

images 此目录是存储已处理图像的位置。

routes 指定如何处理不同的 HTTP 请求。

server 此目录中的 server.go 文件包含用于设置和运行服务器的代码。

tasks 在此目录中，管理任务（包括排队和处理）相关的代码。

worker 这里的 worker.go 文件处理后台任务，确保任务高效执行。

这种有组织的文件夹结构使服务的组件保持独立，从而更容易管理和维护代码库。

image-processing-service/
├── docker-compose.yaml
├── go.mod
├── go.sum
├── handlers
│   └── handlers.go
├── images
├── routes
│   └── routes.go
├── server
│   └── server.go
├── tasks
│   ├── queue.go
│   └── tasks.go
└── worker
    └── worker.go

docker-compose.yaml 文件：

version: "3.9"

services:
  redis:
    image: redis:latest
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data
    networks:
      - image-processing-network

  asynqmon:
    image: hibiken/asynqmon:latest
    environment:
      - REDIS_ADDR=redis:6379
    ports:
      - "8080:8080"
    depends_on:
      - redis
    networks:
      - image-processing-network

networks:
  image-processing-network:

volumes:
  redis-data:

Redis 服务：此服务使用最新版本的 Redis 镜像。它就像一个运行 Redis 数据库的容器。它连接到 image-processing-network 网络，允许它与同一网络上的其他服务进行通信。此外，它还使用名为存储数据的 redis-data 卷。

Asynqmon 服务：此服务使用最新版本的 hibiken/asynqmon 映像。它是 Asynq 的监控工具。它还连接到 image-processing-network 网络以与其他服务进行通信。它依赖于 Redis 服务，这意味着它会等待 Redis 服务启动，然后再启动。

总而言之，此 docker-compose.yaml 文件定义了两个服务（Redis 和 Asynqmon），它们是“图像处理网络”网络的一部分，并使用名为“redis-data”的卷进行数据存储。

通过命令 docker-compose up -d 运行底层基础设施

看一下应用程序的主要入口点，即 server/server.go

此代码使用 Fiber 设置图像处理服务的服务器，配置路由，连接到 Redis，并在端口 3000 上侦听请求。

package main

import (
 "github.com/gofiber/fiber/v2"
 "github.com/qahta0/image-processing-service/routes"
 "github.com/qahta0/image-processing-service/tasks"
 "log"
)

const redisAddress = "127.0.0.1:6379"

func main() {
 app := fiber.New()
 routes.Setup(app)
 tasks.Init(redisAddress)
 defer tasks.Close()
 log.Fatal(app.Listen(":3000"))
}

看一下应用程序的路由，它们是 routes/routes.go

此代码使用 Fiber 框架在“/image-process”处设置单个 POST 路由，允许该 handlers.UploadImage 函数处理图像上传。

package routes

import (
    "github.com/gofiber/fiber/v2"
    "github.com/qahta0/image-processing-service/handlers"
)

func Setup(app *fiber.App) {
    app.Post("/image-process", handlers.UploadImage)
}

在包中 tasks ，我们使用库 hibiken/asynq 创建一个集中式任务队列。该队列可高效管理后台任务，连接到 Redis 进行无缝处理，并简化图像处理服务中的任务管理。它确保对同一队列对象的一致访问。

package tasks

import (
 "github.com/hibiken/asynq"
 "sync"
)

var (
 client *asynq.Client
 once   sync.Once
)

func Init(redisAddress string) {
 once.Do(func() {
  client = asynq.NewClient(asynq.RedisClientOpt{Addr: redisAddress})
 })
}

func Close() {
 if client != nil {
  client.Close()
 }
}

func GetClient() *asynq.Client {
 return client
}

应用程序的路由，它们是 handlers/handlers.go

此代码定义用于上传和处理图像的处理程序。以下是简明扼要的总结：

该 UploadImage 函数通过 POST 请求处理图像上传。它首先检查上传是否成功，并相应地处理错误。如果上传成功，它将读取上传的图像数据，创建图像大小调整任务，并将其排入队列进行处理。最后，它以成功消息进行响应。

package handlers

import (
 "fmt"
 "github.com/gofiber/fiber/v2"
 "github.com/qahta0/image-processing-service/tasks"
 "io"
)

func UploadImage(c *fiber.Ctx) error {
 file, err := c.FormFile("file")
 if err != nil {
  return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "Upload failed"})
 }
 fileData, err := file.Open()
 if err != nil {
  return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "Failed to open the file"})
 }
 defer fileData.Close()
 data, err := io.ReadAll(fileData)
 if err != nil {
  return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "Failed to read the file"})
 }
 resizeTasks, err := tasks.NewImageResizeTasks(data)
 if err != nil {
  return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "Could not create image resize tasks"})
 }
 client := tasks.GetClient()
 for _, task := range resizeTasks {
  if _, err := client.Enqueue(task); err != nil {
   fmt.Printf("Error enqueuing task: %v\n", err)
   return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "Could not enqueue image resize task"})
  }
 }
 return c.JSON(fiber.Map{"message": "Image uploaded and resizing tasks started"})
}

在下一个关键步骤中，该函数将创建图像大小调整任务，并有效地将它们排入队列以进行异步执行。这种方法可确保图像处理不会延迟主线程，从而允许它继续处理用户的请求。这种任务分离有助于提高图像处理服务的整体响应能力和效率。

通过以下命令 go run ./server/server.go 运行主服务器

通过访问 http://localhost:8080 来检查 Asynqmon 门户，我们注意到这里没有什么特别之处，因为现在没有正在执行的作业！

用 Postman 请求在 . http://127.0.0.1:3000/process-image 可以将图像上传到 image 字段

发送请求后，我们观察到处理时间非常快，因为 Redis 中的排队任务几乎是即时发生的，复杂度为 O（1）。随后，我们返回 Asyncmon，发现目前有 10 个任务在队列中等待。

现在，为了异步使用任务，我们将利用以下代码片段。此任务旨在使用 Asynq 库处理异步图像大小调整任务。它定义 image:resize 了图像 data 、 、 width height 和 的任务类型和 filename 有效负载结构。

该函数为各种标准宽度生成多个调整大小任务，而 HandleResizeImageTask 该 NewImageResizeTasks 函数通过调整图像大小、使用唯一文件名保存图像并显示输出 UUID 来处理这些任务。此代码有助于在应用程序中高效且并发地调整图像大小。

package tasks

import (
 "bytes"
 "context"
 "encoding/json"
 "fmt"
 "github.com/google/uuid"
 "github.com/hibiken/asynq"
 "github.com/nfnt/resize"
 "image"
 "image/jpeg"
 _ "image/jpeg"
 _ "image/png"
 "os"
 "path/filepath"
 "time"
)

const TypeResizeImage = "image:resize"

type ResizeImagePayload struct {
 ImageData []byte
 Width     uint
 Height    uint
 FileName  string
}

var StandardWidths = []uint{16, 32, 128, 240, 320, 480, 540, 640, 800, 1024}

func NewImageResizeTasks(imageData []byte, fileName string) ([]*asynq.Task, error) {
 img, _, err := image.Decode(bytes.NewReader(imageData))
 if err != nil {
  return nil, err
 }
 originalBounds := img.Bounds()
 originalWidth := uint(originalBounds.Dx())
 originalHeight := uint(originalBounds.Dy())
 var tasks []*asynq.Task
 for _, width := range StandardWidths {
  height := (width * originalHeight) / originalWidth
  payload := ResizeImagePayload{
   ImageData: imageData,
   Width:     width,
   Height:    height,
   FileName:  fileName,
  }
  payloadBytes, err := json.Marshal(payload)
  if err != nil {
   return nil, err
  }
  task := asynq.NewTask(TypeResizeImage, payloadBytes)
  tasks = append(tasks, task)
 }
 return tasks, nil
}

func HandleResizeImageTask(ctx context.Context, t *asynq.Task) error {
 var payload ResizeImagePayload
 if err := json.Unmarshal(t.Payload(), &payload); err != nil {
  return fmt.Errorf("failed to parse resize image task payload: %v", err)
 }
 img, _, err := image.Decode(bytes.NewReader(payload.ImageData))
 if err != nil {
  return fmt.Errorf("image decode failed: %v", err)
 }
 resizedImg := resize.Resize(payload.Width, payload.Height, img, resize.Lanczos3)
 outputUUID := uuid.New()
 outputFileName := fmt.Sprintf("images/%s/%s%s", time.Now().Format("2006-01-02"), outputUUID.String(), filepath.Ext(payload.FileName))
 outputDir := filepath.Dir(outputFileName)
 if _, err := os.Stat(outputDir); os.IsNotExist(err) {
  if err := os.MkdirAll(outputDir, 0755); err != nil {
   return err
  }
 }
 outFile, err := os.Create(outputFileName)
 if err != nil {
  return err
 }
 defer outFile.Close()
 if err := jpeg.Encode(outFile, resizedImg, nil); err != nil {
  return err
 }
 fmt.Printf("Output UUID for the processed image: %s\n", outputUUID.String())
 return nil
}

接下来，我们应该通过运行命令 go run worker/worker.go 来执行工作线程，我们观察到 10 个任务是异步高效处理的，与主线程分开。

在运行工作线程时，我们观察到队列中先前的 10 个任务现已处理并完成。

该函数会以一条消息进行响应，确认图像上传成功。此外，它还通知用户调整大小任务已在后台启动。这种无缝过程不仅提高了用户满意度，还展示了图像处理服务中异步任务处理的强大功能。

未来增强功能：通过 GCP Bucket 进行云集成

展望未来，我计划通过集成云存储功能来增强服务的功能，特别是针对 Google Cloud Platform （GCP） 存储桶。这将允许上传和存储处理后的图像，进一步简化工作流程并增加一层便利性和安全性。

结论

本文解决了在 Web 开发中优化图像处理的关键挑战。通过利用 Golang、Asynq、Redis 和 Fiber，图像处理服务通过高效的异步任务处理来提高性能。本文展示了经过深思熟虑的软件技术在解决 Web 开发中的关键瓶颈方面的力量，从而带来响应速度更快、更高效的用户体验。