RPC over RabbitMQ

ขอคำตอบกลับผ่าน queue

messaging เป็น fire-and-forget โดยธรรมชาติ — แต่บางครั้งคุณอยากได้ คำตอบกลับ RabbitMQ รองรับ pattern request/reply (RPC) ด้วย message property สองตัวที่คุณเจอมาแล้ว:

reply_to — client สร้าง callback queue แล้วใส่ชื่อ queue นั้นตรงนี้ บอก server ว่า “ส่งคำตอบไปที่ queue นี้”
correlation_id — id ที่ไม่ซ้ำที่ client แนบกับแต่ละ request และคาดว่าจะได้ echo กลับมา เพื่อจับคู่ reply กับ request ต้นทาง (reply อาจมาไม่เรียงลำดับ)

sequenceDiagram
  participant C as Client
  participant RQ as rpc_queue
  participant S as Server
  participant CB as callback queue
  C->>RQ: request (reply_to=cb, correlation_id=abc)
  RQ->>S: deliver
  S->>S: compute result
  S->>CB: reply (correlation_id=abc)
  CB->>C: deliver
  Note over C: match abc to the pending request

RPC round-trip ที่จับคู่ด้วย correlation_id

สองฝั่ง

server consume จาก request queue ที่รู้จักกันดี ทำงาน แล้ว publish ผลลัพธ์ไปที่ reply_to queue ของ request โดย echo correlation_id กลับ ส่วน client publish request พร้อม correlation_id ใหม่และ reply_to แล้วรอบน callback queue ของตัวเองเพื่อรับ reply ที่มี id ตัวเดียวกันนั้น

// Server
await channel.assertQueue('rpc_queue', { durable: false });
await channel.prefetch(1);
channel.consume('rpc_queue', (msg) => {
  const n = parseInt(msg.content.toString(), 10);
  const result = Buffer.from(String(fib(n)));
  channel.sendToQueue(msg.properties.replyTo, result, {
    correlationId: msg.properties.correlationId,
  });
  channel.ack(msg);
});

// Client (sketch)
const { queue: cbQueue } = await channel.assertQueue('', { exclusive: true });
const correlationId = randomUUID();
channel.consume(cbQueue, (msg) => {
  if (msg.properties.correlationId === correlationId) {
    console.log('result:', msg.content.toString());
  }
}, { noAck: true });
channel.sendToQueue('rpc_queue', Buffer.from('30'), {
  correlationId, replyTo: cbQueue,
});

# Server
channel.queue_declare(queue="rpc_queue")
channel.basic_qos(prefetch_count=1)
def on_request(ch, method, props, body):
    result = str(fib(int(body)))
    ch.basic_publish(
        exchange="", routing_key=props.reply_to,
        properties=pika.BasicProperties(correlation_id=props.correlation_id),
        body=result,
    )
    ch.basic_ack(delivery_tag=method.delivery_tag)
channel.basic_consume(queue="rpc_queue", on_message_callback=on_request)

# Client (sketch)
cb = channel.queue_declare(queue="", exclusive=True).method.queue
corr_id = str(uuid.uuid4())
channel.basic_publish(
    exchange="", routing_key="rpc_queue",
    properties=pika.BasicProperties(reply_to=cb, correlation_id=corr_id),
    body="30",
)

// Server
ch.QueueDeclare("rpc_queue", false, false, false, false, nil)
ch.Qos(1, 0, false)
msgs, _ := ch.Consume("rpc_queue", "", false, false, false, false, nil)
for d := range msgs {
    result := strconv.Itoa(fib(atoi(d.Body)))
    ch.PublishWithContext(ctx, "", d.ReplyTo, false, false, amqp.Publishing{
        CorrelationId: d.CorrelationId,
        Body:          []byte(result),
    })
    d.Ack(false)
}

// Client (sketch)
cb, _ := ch.QueueDeclare("", false, false, true, false, nil)
corrId := uuid.NewString()
ch.PublishWithContext(ctx, "", "rpc_queue", false, false, amqp.Publishing{
    ReplyTo:       cb.Name,
    CorrelationId: corrId,
    Body:          []byte("30"),
})

คุณควรทำแบบนี้จริงไหม?

ถามตัวเองตรง ๆ ก่อนจะหยิบ RPC-over-RabbitMQ มาใช้ คุณกำลัง สร้าง synchronous call ขึ้นมาใหม่บน async infrastructure — พร้อม callback queue, การจัดการ correlation และ timeout handling ของตัวเอง — เพื่อให้ได้สิ่งที่ HTTP หรือ gRPC call ธรรมดาให้ฟรี ๆ สำหรับ request/reply ส่วนใหญ่ direct call ง่ายกว่า debug ง่ายกว่า และ latency ต่ำกว่า

RPC-over-messaging คุ้มเมื่อคุณอยากได้สิ่งที่ broker เพิ่มให้จริง ๆ: load-balance request ข้าม pool ของ worker (RPC server หลายตัวแข่งกันบน rpc_queue), location transparency (client ไม่รู้และไม่สนว่า worker ตัวไหนตอบ), หรือ buffering request ตอน worker ยุ่งชั่วครู่ ถ้าคุณไม่ต้องการสิ่งเหล่านี้ ก็อย่าสร้างขึ้นมา