diff --git a/consumer.go b/consumer.go index 2127b65..8a17f4a 100644 --- a/consumer.go +++ b/consumer.go @@ -3,6 +3,7 @@ package rabbit import ( "context" amqp "github.com/rabbitmq/amqp091-go" + "golang.org/x/time/rate" "log" "time" ) @@ -11,6 +12,8 @@ func runConsumer(ctx context.Context, queue *Client, msgCh chan []byte) { runCtx, cancel := context.WithCancel(ctx) defer cancel() + limiter := rate.NewLimiter(rate.Every(time.Millisecond*500), 100) + deliveries, err := queue.Consume() if err != nil { 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: msgCh <- delivery.Body log.Printf("Received message: %s\n", delivery.Body) - + if err = delivery.Ack(false); err != nil { log.Printf("Error acknowledging message: %s\n", err) } - <-time.After(time.Second * 2) + limiter.Wait(runCtx) } } } diff --git a/go.mod b/go.mod index d18bf43..0a3c615 100644 --- a/go.mod +++ b/go.mod @@ -3,3 +3,5 @@ module repo.nqws.ru/merch-tracker-v2/mt-rabbit go 1.25.0 require github.com/rabbitmq/amqp091-go v1.10.0 + +require golang.org/x/time v0.14.0 // indirect diff --git a/go.sum b/go.sum index 024eebe..64cd225 100644 --- a/go.sum +++ b/go.sum @@ -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= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= 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=