merch-tracker-v2/mt-rabbit
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

simple rate limiter

Browse source
This commit is contained in:
nquidox 2026-02-20 14:39:24 +03:00
parent d2af3b1183
commit 8b1bc57cd0
3 changed files with 9 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

7
consumer.go
View file

@ -3,6 +3,7 @@ package rabbit
import ( import (
"context" "context"
amqp "github.com/rabbitmq/amqp091-go" amqp "github.com/rabbitmq/amqp091-go"
"golang.org/x/time/rate"
"log" "log"
"time" "time"
) )
@ -11,6 +12,8 @@ func runConsumer(ctx context.Context, queue *Client, msgCh chan []byte) {
runCtx, cancel := context.WithCancel(ctx) runCtx, cancel := context.WithCancel(ctx)
defer cancel() defer cancel()
limiter := rate.NewLimiter(rate.Every(time.Millisecond*500), 100)
deliveries, err := queue.Consume() deliveries, err := queue.Consume()
if err != nil { if err != nil {
log.Printf("Could not start consuming: %s\n", err) log.Printf("Could not start consuming: %s\n", err)
@ -44,11 +47,11 @@ func runConsumer(ctx context.Context, queue *Client, msgCh chan []byte) {
case delivery := <-deliveries: case delivery := <-deliveries:
msgCh <- delivery.Body msgCh <- delivery.Body
log.Printf("Received message: %s\n", delivery.Body) log.Printf("Received message: %s\n", delivery.Body)
if err = delivery.Ack(false); err != nil { if err = delivery.Ack(false); err != nil {
log.Printf("Error acknowledging message: %s\n", err) log.Printf("Error acknowledging message: %s\n", err)
} }
<-time.After(time.Second * 2) limiter.Wait(runCtx)
} }
} }
} }

2
go.mod
View file

@ -3,3 +3,5 @@ module repo.nqws.ru/merch-tracker-v2/mt-rabbit
go 1.25.0 go 1.25.0
require github.com/rabbitmq/amqp091-go v1.10.0 require github.com/rabbitmq/amqp091-go v1.10.0
require golang.org/x/time v0.14.0 // indirect

2
go.sum
View file

@ -2,3 +2,5 @@ github.com/rabbitmq/amqp091-go v1.10.0 h1:STpn5XsHlHGcecLmMFCtg7mqq0RnD+zFr4uzuk
github.com/rabbitmq/amqp091-go v1.10.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= github.com/rabbitmq/amqp091-go v1.10.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI=
golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4=