如何为 RabbitMQ RPC 请求设置超时？

Posted 2023-02-16

【中文标题】如何为 RabbitMQ RPC 请求设置超时？

【英文标题】：How to set a timeout to a RabbitMQ RPC request?

【发布时间】：2020-04-13 21:45:25

【问题描述】：

在 AMQP (RabbitMQ) RPC 模型中向主题发布消息是否存在超时？

我不想等待很长时间（超时后）消费者对生产者消息的回答。

参考：RPC (Go RabbitMQ Client)

【参考方案1】：

（示例代码使用streadway/amqp）

您可以通过使用计时器来实现这一点，它可以尽可能短。然后使用 select 语句等待 RPC 响应和计时器通道：

func doRPC() ([]byte, error) 
    // ...RPC setup code
    timer := time.NewTimer(2 * time.Second)
    for 
        // waits until either a response from the RPC server
        // is available or the timer expires
        select 
        case msg := <-msgs: // msgs is of type <-chan amqp.Delivery
            if msg.CorrelationId == correlationID 
                return msg.Body, nil
            

        case <-timer.C:
            return nil, errors.New("waiting for RPC response timed out or was cancelled")
        
    

要让调用者控制超时，您可以使用上下文来代替计时器。它的工作方式相同，只是现在您在上下文Done() 频道上选择：

func caller() 
    // the caller can create a context with the desired timeout
    ctx, cancel := context.WithTimeout(context.Background(), 2 * time.Second)
    defer cancel()

    doRPC(ctx)


func doRPC(ctx context.Context) ([]byte, error) 
    // ...RPC setup code
    for 
        select 
        case msg := <-msgs: // msgs is of type <-chan amqp.Delivery
            if msg.CorrelationId == correlationID 
                return msg.Body, nil
            

        case <-ctx.Done():
            return nil, errors.New("waiting for RPC response timed out or was cancelled")