Direct Exchange

routing แบบ match เป๊ะ

direct exchange route message ไปยัง queue ที่มี binding key ตรงกับ routing key ของ message เป๊ะ ๆ ไม่มี pattern ไม่มี wildcard — เทียบ string ให้เท่ากันตรง ๆ นี่คือ exchange type ที่ง่ายที่สุด และเป็นตัวที่หยิบมาใช้เมื่อ message มีปลายทางเป็นหมวดหมู่ที่ชัดเจนหมวดเดียว

จริง ๆ คุณเคยใช้ direct exchange มาแล้วโดยไม่รู้ตัว: default exchange คือ direct exchange ที่ไม่มีชื่อ ซึ่งทุก queue จะถูก bind เข้าไปด้วยชื่อของตัวเองโดยอัตโนมัติ นั่นคือเหตุผลที่ publish ไปที่ routing key hello ด้วย exchange เป็น string ว่าง แล้ว message ไปลงที่ queue ชื่อ hello

ตัวอย่าง: log ตาม severity

สมมติคุณส่ง log message และอยากให้ log ระดับ error ไปเข้า queue หนึ่ง (ไว้ alert) และ info ไปอีก queue หนึ่ง (ไว้ archive) ก็ประกาศ direct exchange, bind แต่ละ queue ด้วย severity เป็น binding key แล้ว publish โดยใช้ severity เป็น routing key

flowchart LR
  p["publish
key: error"] --> x["direct exchange
'logs'"]
  x -->|"bind key: error"| qe["queue: errors"]
  x -->|"bind key: info"| qi["queue: info"]
  x -. "ไม่ match 'error'" .-> qi

direct exchange match routing key กับ binding key

message ที่ publish ด้วย routing key error จะ match แค่ binding ที่ key เป็น error เท่านั้น จึงไปเข้า queue errors ไม่ใช่ info

const ex = 'logs';
await channel.assertExchange(ex, 'direct', { durable: true });

// consumer side: bind a queue for errors only
const q = await channel.assertQueue('errors', { durable: true });
await channel.bindQueue(q.queue, ex, 'error');

// producer side: publish with the severity as the routing key
channel.publish(ex, 'error', Buffer.from('disk full on node-3'));
channel.publish(ex, 'info', Buffer.from('user logged in'));

ex = "logs"
channel.exchange_declare(exchange=ex, exchange_type="direct", durable=True)

# consumer side: bind a queue for errors only
channel.queue_declare(queue="errors", durable=True)
channel.queue_bind(queue="errors", exchange=ex, routing_key="error")

# producer side: publish with the severity as the routing key
channel.basic_publish(exchange=ex, routing_key="error", body="disk full on node-3")
channel.basic_publish(exchange=ex, routing_key="info", body="user logged in")

ex := "logs"
ch.ExchangeDeclare(ex, "direct", true, false, false, false, nil)

// consumer side: bind a queue for errors only
ch.QueueDeclare("errors", true, false, false, false, nil)
ch.QueueBind("errors", "error", ex, false, nil)

// producer side: publish with the severity as the routing key
ch.PublishWithContext(ctx, ex, "error", false, false,
    amqp.Publishing{Body: []byte("disk full on node-3")})
ch.PublishWithContext(ctx, ex, "info", false, false,
    amqp.Publishing{Body: []byte("user logged in")})

หลาย binding, key เดียวกัน

direct exchange ไม่ได้จำกัดที่ queue เดียวต่อ key ถ้ามี queue สองตัว bind ด้วย key error เหมือนกัน message error ตัวเดียวจะถูก copy ไป ทั้งสอง queue — คุณจึงผสม routing แบบ match เป๊ะเข้ากับการ fan-out นิด ๆ ได้ ในทางกลับกัน queue เดียวก็ bind ได้หลาย key (bind errors กับทั้ง error และ critical) เพื่อรวบหลายหมวดมาไว้ด้วยกัน