feat(video): VideoSummary consumer 添加延迟重试和失败重传次数

src/services/publish/handler.go

@@ -50,14 +50,17 @@ func (a PublishServiceImpl) New() {
 	channel, err = conn.Channel()
 	exitOnError(err)
 
+	exchangeArgs := amqp.Table{
+		"x-delayed-type": "topic",
+	}
 	err = channel.ExchangeDeclare(
 		strings.VideoExchange,
-		"fanout",
+		"x-delayed-message", //"topic",
 		true,
 		false,
 		false,
 		false,
-		nil,
+		exchangeArgs,
 	)
 	exitOnError(err)
 
@@ -83,7 +86,7 @@ func (a PublishServiceImpl) New() {
 
 	err = channel.QueueBind(
 		strings.VideoPicker,
-		"",
+		strings.VideoPicker,
 		strings.VideoExchange,
 		false,
 		nil,
@@ -92,7 +95,7 @@ func (a PublishServiceImpl) New() {
 
 	err = channel.QueueBind(
 		strings.VideoSummary,
-		"",
+		strings.VideoSummary,
 		strings.VideoExchange,
 		false,
 		nil,
@@ -265,20 +268,24 @@ func (a PublishServiceImpl) CreateVideo(ctx context.Context, request *publish.Cr
 	// Context 注入到 RabbitMQ 中
 	headers := rabbitmq.InjectAMQPHeaders(ctx)
 
-	err = channel.Publish(strings.VideoExchange, "", false, false,
-		amqp.Publishing{
-			DeliveryMode: amqp.Persistent,
-			ContentType:  "text/plain",
-			Body:         marshal,
-			Headers:      headers,
-		})
-
-	if err != nil {
-		resp = &publish.CreateVideoResponse{
-			StatusCode: strings.VideoServiceInnerErrorCode,
-			StatusMsg:  strings.VideoServiceInnerError,
+	routingKeys := []string{strings.VideoPicker, strings.VideoSummary}
+	for _, key := range routingKeys {
+		// Send raw video to all queues bound the exchange
+		err = channel.Publish(strings.VideoExchange, key, false, false,
+			amqp.Publishing{
+				DeliveryMode: amqp.Persistent,
+				ContentType:  "text/plain",
+				Body:         marshal,
+				Headers:      headers,
+			})
+
+		if err != nil {
+			resp = &publish.CreateVideoResponse{
+				StatusCode: strings.VideoServiceInnerErrorCode,
+				StatusMsg:  strings.VideoServiceInnerError,
+			}
+			return
 		}
-		return
 	}
 
 	resp = &publish.CreateVideoResponse{

src/services/videoprocessor/main.go

@@ -57,14 +57,17 @@ func main() {
 		exitOnError(err)
 	}(ch)
 
+	exchangeArgs := amqp.Table{
+		"x-delayed-type": "topic",
+	}
 	err = ch.ExchangeDeclare(
 		strings.VideoExchange,
-		"fanout",
+		"x-delayed-message", //"topic",
 		true,
 		false,
 		false,
 		false,
-		nil,
+		exchangeArgs,
 	)
 	exitOnError(err)
 
@@ -90,7 +93,7 @@ func main() {
 
 	err = ch.QueueBind(
 		strings.VideoPicker,
-		"",
+		strings.VideoPicker,
 		strings.VideoExchange,
 		false,
 		nil,
@@ -99,7 +102,7 @@ func main() {
 
 	err = ch.QueueBind(
 		strings.VideoSummary,
-		"",
+		strings.VideoSummary,
 		strings.VideoExchange,
 		false,
 		nil,
@@ -178,7 +181,13 @@ func Consume(channel *amqp.Channel) {
 				"err":        err,
 			}).Errorf("Error when updating file information to database")
 			logging.SetSpanError(span, result.Error)
-			errorHandler(d, true, logger, &span)
+			err = d.Nack(false, true)
+			if err != nil {
+				logger.WithFields(logrus.Fields{
+					"err": err,
+				}).Errorf("Error when resending the video to queue...")
+				logging.SetSpanError(span, err)
+			}
 			span.End()
 			continue
 		}