场景：多个消费者（worker）竞争消费同一个队列中的消息，实现任务分发和负载均衡

特点：

• 消息默认以 轮询（Round-Robin） 方式分发给消费者。

• 可通过 Prefetch 配置实现 公平分发（防止某个 Worker 积压消息）

RabbitMq版本：4.0.5    go：1.24.1     ffmpeg:7.0.2(win)

准备

RabbitMq的安装就不多讲解了，搜索教程就可以直接安装了

ffmpeg:

Builds - CODEX FFMPEG @ gyan.devhttps://www.gyan.dev/ffmpeg/builds/

下滑找到release bulids部分，选择ffmpeg-7.0.2-essentials_build.zip解压

将D:\Software\ffmpeg-7.0.2-essentials_build\bin目录添加到系统的path目录中,出现下面的就标识成功了

核心代码

消费者

package main

import (
	"encoding/json"
	amqp "github.com/rabbitmq/amqp091-go"
	"log"
	"os/exec"
)

// TranscodeTask 定义视频转码任务结构体
type TranscodeTask struct {
	VideoID    string `json:"video_id"`    // 视频唯一ID
	InputPath  string `json:"input_path"`  // 原始视频路径
	OutputPath string `json:"output_path"` // 转码输出路径
	Format     string `json:"format"`      // 目标格式（如720p/480p）
}

// failOnError 错误处理辅助函数
func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %v", msg, err)
	}
}

// transcodeVideo 调用FFmpeg执行视频转码
func transcodeVideo(inputPath, outputPath, format string) error {
	var args []string
	switch format {
	case "720p":
		// 缩放为720p分辨率
		args = []string{"-i", inputPath, "-vf", "scale=-1:720", outputPath}
	case "480p":
		// 缩放为480p分辨率
		args = []string{"-i", inputPath, "-vf", "scale=-1:480", outputPath}
	default:
		// 默认只进行格式转换，不调整分辨率
		args = []string{"-i", inputPath, outputPath}
	}

	// 执行FFmpeg命令（注意修改为你的FFmpeg路径）
	cmd := exec.Command("D:\\Software\\ffmpeg-7.0.2-essentials_build\\bin\\ffmpeg.exe", args...)
	cmd.Stdout = nil // 可选项：将输出重定向到日志文件
	cmd.Stderr = nil // 可选项：FFmpeg错误输出通常较多
	return cmd.Run()
}

func main() {
	// 连接到RabbitMQ服务器
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "RabbitMQ连接失败")
	defer func(conn *amqp.Connection) {
		err := conn.Close()
		if err != nil {
			log.Printf("关闭连接时出错: %v", err)
		}
	}(conn)

	// 创建通信通道
	ch, err := conn.Channel()
	failOnError(err, "无法打开通道")
	defer func(ch *amqp.Channel) {
		err := ch.Close()
		if err != nil {
			log.Printf("关闭通道时出错: %v", err)
		}
	}(ch)

	// 设置公平分发策略（每个消费者同时只处理一个任务）
	err = ch.Qos(1, 0, false)
	failOnError(err, "无法设置QoS")

	// 开始消费转码队列中的消息
	msgs, err := ch.Consume(
		"transcode_queue", // 队列名称
		"",                // 消费者标识
		false,             // 关闭自动确认
		false,             // 非独占队列
		false,             // 不等待
		false,             // 无额外参数
		nil,
	)
	failOnError(err, "注册消费者失败")

	// 创建永久阻塞的通道
	var forever chan struct{}

	go func() {
		for d := range msgs {
			// 解析JSON格式的任务数据
			var task TranscodeTask
			err := json.Unmarshal(d.Body, &task)
			if err != nil {
				log.Printf("JSON解析失败: %s", err)
				err := d.Nack(false, false) // 丢弃无法解析的消息
				if err != nil {
					log.Printf("消息拒绝失败: %v", err)
				}
				continue
			}

			log.Printf(" [x] 收到转码任务: %+v", task)

			// 开始转码处理
			log.Printf(" [x] 开始转码: %s", task.OutputPath)
			err = transcodeVideo(task.InputPath, task.OutputPath, task.Format)
			if err != nil {
				log.Printf(" [!] 转码失败: %v", err)
				err := d.Nack(false, false) // 转码失败时丢弃消息
				if err != nil {
					log.Printf("消息拒绝失败: %v", err)
				}
				continue
			}

			// 转码成功
			log.Printf(" [x] 转码完成: %s", task.OutputPath)
			err = d.Ack(false) // 手动确认消息处理完成
			if err != nil {
				log.Printf("消息确认失败: %v", err)
			}
		}
	}()

	log.Printf(" [*] 等待转码任务. 按CTRL+C退出")
	<-forever // 永久阻塞主线程
}

生产者

package main

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"log"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

// TranscodeTask 转码任务结构体
type TranscodeTask struct {
	VideoID    string `json:"video_id"`    // 视频唯一ID
	InputPath  string `json:"input_path"`  // 原始视频路径（如 S3 地址）
	OutputPath string `json:"output_path"` // 转码后路径
	Format     string `json:"format"`      // 目标格式（如 "720p"）
}

func main() {
	// 连接到 RabbitMQ
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "连接到 RabbitMQ 失败")
	defer conn.Close()

	// 打开一个通道
	ch, err := conn.Channel()
	failOnError(err, "打开通道失败")
	defer ch.Close()

	// 设置队列参数（包括死信交换机）
	args := amqp.Table{
		"x-dead-letter-exchange": "dlx_transcode",
	}
	// 声明队列
	q, err := ch.QueueDeclare(
		"transcode_queue", // 队列名称
		true,              // 持久化
		false,             // 当没有消费者时删除队列
		false,             // 排他队列
		false,             // 不等待
		args,              // 队列参数
	)
	failOnError(err, "声明队列失败")

	// 生成随机文件名
	randomFileName := generateRandomFileName(".mp4")

	// 创建转码任务
	task := TranscodeTask{
		VideoID:    "video_123",           // 视频唯一ID
		InputPath:  "./original.mp4",      // 原始视频路径
		OutputPath: "./" + randomFileName, // 转码后路径
		Format:     "720p",                // 目标格式
	}
	// 将任务序列化为 JSON
	body, err := json.Marshal(task)
	failOnError(err, "序列化任务失败")

	// 设置超时上下文
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	// 发布消息到队列
	err = ch.PublishWithContext(ctx,
		"",     // 交换机名称（使用默认交换机）
		q.Name, // 队列名称
		false,  // 强制
		false,  // 立即
		amqp.Publishing{
			DeliveryMode: amqp.Persistent,    // 消息持久化
			ContentType:  "application/json", // 内容类型
			Body:         body,               // 消息体
		})
	failOnError(err, "发布任务失败")
	log.Printf(" [x] 发送转码任务: %s -> %s", task.VideoID, task.OutputPath)
}

// failOnError 检查错误并打印日志
func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %v", msg, err)
	}
}

// generateRandomFileName 生成随机字符串的文件名，附加给定的后缀
func generateRandomFileName(extension string) string {
	// 生成随机字节
	bytes := make([]byte, 8)
	if _, err := rand.Read(bytes); err != nil {
		log.Fatalf("生成随机字节失败: %v", err)
	}
	// 将随机字节转换为十六进制字符串并附加后缀
	return hex.EncodeToString(bytes) + extension
}

目录结构，爆红不影响

在目录输入cmd进入先启动生产者,声明队列

go run .\prodcuer.go

然后可以启动多个窗口模拟多个消费者

go run .\consumer.go