Rabbit MQ เครื่องมือจัดการ Asynchronous อย่างเป็นระบบ

         Synchronous / Asynchronous เข้ามามีบทบาทมากขึ้นเมื่อเรา implement Microservices เมื่อเรามี services เยอะขึ้น การคุยกันก็มีความซับซ้อนมากขึ้น ถ้าพูดในมุมของ client ก็คงต่างกันแค่ รอ และ ไม่รอ เท่านั้นเอง


         Messaging Queue (MQ) ก็เกิดมาเพื่อเป็นตัวกลางระหว่างผู้รับและผู้ส่ง และยังมี protocol เฉพาะอย่าง AMQP ที่กำหนดเป็นมาตรฐาน เพื่อการันตีข้อมูลว่าจะไม่หายระหว่างทางด้วย ซึ่งการใช้งานนั้นจำเป็นต้องอาศัยเครื่องมืออย่าง Message Broker เพื่อควบคุม queue อีกที

Message Broker เป็น software ที่เราต้องติดตั้งบน server ไว้ด้วย เพื่อ tracking & monitoring รวมถึงมี console ให้สามารถทดสอบส่งข้อมูลเข้า queue โดยไม่ต้องเขียนโค้ดเอง ตัวอย่างเช่น Rabbit MQ, Apache Kafka และอื่นๆ อีกมากมาย



         Rabbit MQ เป็น Message Broker ที่ได้รับความนิยมอีกตัวหนึ่ง สามารถใช้งานได้ฟรี และมีความ lightweight กว่าเจ้าอื่น อีกทั้งยังมี web console ให้ผู้ดูแลเข้ามาใช้งานง่ายๆ ด้วย

แนะตัวกันก่อน

  • Producer / Publisher / Sender = ผู้ส่งข้อมูล
  • Consumer / Subscriber / Receiver = ผู้รับข้อมูล

         ส่วนนี้ใช้การเขียนโปรแกรมแบบ PubSub Design Pattern นั่นเอง และในส่วนของ Rabbit MQ จะมี component ดังนี้

  • Exchange = รูปแบบการส่ง
  • Queue = ก้อนข้อมูล
  • Binding / Routing = เส้นทาง

         สิ่งที่ RabbitMQ จะทำให้คือ การจัดคิว และการควบคุมเส้นทางการส่งข้อมูล โดยเราสามารถเลือกใช้ตามประเภทที่เราต้องการได้

         จากภาพจะเห็นว่า Rabbit MQ หรือ Broker ของเรานั้น จะมีหน้าที่ควบคุมเมื่อได้รับข้อมูลมาแล้ว ควรจะส่งต่อให้ใครบ้าง โดยเงื่อนไขการส่งต่อนั้นขึ้นอยู่กับประเภทของ Exchange และ Binding

         ถ้าภาพยังชัดไม่มากพอ ลองดู vdo จาก cloudamqp ก็ได้


เริ่มเข้าสู่โหมด Coding กัน 

ขาส่ง

  1. Producer จะต้องเชื่อมต่อ AMQP ดังนี้ amqp://guest:[email protected]:5672/
  2. Producer จะต้องสร้าง Exchange พร้อมกำหนดประเภท
  3. เมื่อต้องการส่งข้อมูล Producer จะต้อง publish โดยใช้ routing key* และ
package mq

import (
"fmt"
"github.com/streadway/amqp"
"log"
"net/url"
"os"
)

type Publisher struct {
Conn *amqp.Connection
}

func getMQConnection() string {
u := &url.URL{
Scheme: "amqp",
User: url.UserPassword(os.Getenv("MQ_USERNAME"), os.Getenv("MQ_PASSWORD")),
Host: fmt.Sprintf("%s:%s", os.Getenv("MQ_HOST"), os.Getenv("MQ_PORT")),
}
//"amqp://guest:[email protected]:5672/"
return u.String()
}

func openMQConnection() *amqp.Connection {
conn, err := amqp.Dial(getMQConnection())
failOnError(err, "Failed to connect to RabbitMQ")
return conn
}

func CreateMQService() *Publisher {
pub := &Publisher{
Conn: openMQConnection(),
}
return pub
}

func (pub *Publisher) Publish(exchangeName string, routeKey string, body string) {
ch, err := pub.Conn.Channel()
failOnError(err, "Failed to open a channel")
defer ch.Close()

err = ch.ExchangeDeclare(
exchangeName, // name
"direct", // type
true, // durable
false, // auto-deleted
false, // internal
false, // no-wait
nil, // arguments
)
failOnError(err, "Failed to declare an exchange")

err = ch.Publish(
exchangeName, // exchange
routeKey, // routing key
false, // mandatory
false, // immediate
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(body),
})
log.Printf(" [x] Sent %s", body)
failOnError(err, "Failed to publish a message")
}

func failOnError(err error, msg string) {
if err != nil {
log.Fatalf("%s: %s", msg, err)
}
}


ขารับ

  1. Consumer จะต้องเชื่อมต่อ AMQP เช่น amqp://guest:[email protected]:5672/
  2. Consumer จะต้องสร้าง Queue 
  3. Bind ระหว่าง่ Queue กับ Exchange ด้วย routing key ตามเงื่อนไข
package mq

import (
"github.com/streadway/amqp"
"log"
)

func Consume(queueName string) {
conn, err := amqp.Dial(getMQConnection())
failOnError(err, "Failed to connect to RabbitMQ")
defer conn.Close()

ch, err := conn.Channel()
failOnError(err, "Failed to open a channel")
defer ch.Close()

q, err := ch.QueueDeclare(
queueName, // name
true, // durable
false, // delete when unused
false, // exclusive
false, // no-wait
nil, // arguments
)
failOnError(err, "Failed to declare a queue")

err = ch.QueueBind(
queueName, // queue name
"key", // routing key
"my-ex", // exchange
false,
nil,
)
failOnError(err, "Failed to bind a queue")

msgs, err := ch.Consume(
q.Name, // queue
"", // consumer
true, // auto-ack
false, // exclusive
false, // no-local
false, // no-wait
nil, // args
)
failOnError(err, "Failed to register a consumer")

forever := make(chan bool)

go func() {
for d := range msgs {
log.Printf("Received a message: %s", d.Body)
}
}()

log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
<-forever
}

         จากตัวอย่างเป็นภาษา Go หากใช้ภาษาอื่นๆ เว็บ official ก็มีตัวอย่าง source code ให้ประยุกต์ใช้งานเช่นกัน ลองใช้งานกันดูครับ 😁

ref.

Facebook Comment

Recent Posts

Popular post of 7 days

ลืมรหัสปลดล็อค Android เข้าเครื่องไม่ได้ มีทางออกครับ

[Android] เปิดเครื่องไม่ได้ โลโก้ค้าง (Boot Loop)

Portal คืออะไรกันนะ???

Popular

ลืมรหัสปลดล็อค Android เข้าเครื่องไม่ได้ มีทางออกครับ

ความเร็วเน็ตที่แท้จริง Mbps MBps ???

[Android] เปิดเครื่องไม่ได้ โลโก้ค้าง (Boot Loop)

Portal คืออะไรกันนะ???

[Java] ความแตกต่างระหว่าง Overloading กับ Overriding

พิมพ์ซองจากรายชื่อ excel ด้วยวิธี Mail Merge

[Java] Java Static Variable และ Static Method

Push Notification คืออะไร มีวิธีทำอย่างไร (No Code)

[Android] Download Mode & Recovery Mode

เรื่องของ ++i กับ i++