golang rabbitmq的使用（五）

先说一个实际的业务场景：
Client端有一个请求需要进行耗时处理或者查询，这个处理在Server端做。Server 端处理完后通知给请求的Client端。
这种场景可以称之为RPC(Remote Procedure Call)

有两个点说明一下：

• <1>Client端发送请求给Server端可以简单定义一个Queue。Client作为Producer发布消息，Server端作为Cosumer消费消息

• <2>Server端处理完耗时处理后需要将处理结果返回给请求的客户端。

  • 可以在Client声明一个不指定名称的Queue，系统会自动生成一个随机名称的Queue。将Queue的名称在publish是发送给Server端
  • 因为Server端要将处理结果返回给对应的请求，所以在Client端需要生成一个CorrelationId发送给Server端

处理流程

Client端
(1)声明从Server返回消息用的queue

respQueue, err := ch.QueueDeclare(
        "",    // name
        false, // durable
        false, // delete when unused
        true,  // exclusive
        false, // noWait
        nil,
    )

(2)发送请求消息到rpc_queue

err = ch.Publish(
        "",               //exchange
        config.QUEUENAME, //routing key
        false,
        false,
        amqp.Publishing{
            ContentType:   "text/plain",
            CorrelationId: correlationID,
            ReplyTo:       respQueue.Name,
            Body:          []byte(msgBody),
        })

corrId为自己随机生成的Id

(2)Server端
(3)声明rpc_queue,从rpc_queue中消费消息

q, err := ch.QueueDeclare(
                config.QUEUENAME,
                false,
                false,
                false,
                false,
                nil,
            )

            msgs, err := ch.Consume(
                q.Name,
                "",
                false, // auto ack
                false,
                false,
                false,
                nil,
            )

(4)执行处理后使用msg中的ReplyTo返回处理结果给Client

err = ch.Publish(
        "",
        msg.ReplyTo,
        false,
        false,
        amqp.Publishing{
            ContentType:   "text/plain",
            CorrelationId: msg.CorrelationId,
            Body:          []byte(bookName),
        })

    msg.Ack(false)

(5)Client端从reply queue中接收从Server端来的response

respMsgs, err := ch.Consume(
        respQueue.Name,
        "",
        true,  // auto-ack
        true,  // exclusive
        false, // noLocal
        false, // nowait
        nil,
    )

详细代码如下
conf.go

package config

const (
    RMQADDR            = "amqp://guest:guest@172.17.84.205:5672/"
    QUEUENAME          = "rpc_queue"
    SERVERINSTANCESCNT = 5
)

client.go

package main

import (
    config "binTest/rabbitmqTest/t1/l6/conf"
    "fmt"
    "log"
    "math/rand"
    "os"

    "github.com/streadway/amqp"
)

func main() {

    if len(os.Args) < 2 {
        log.Println("Arguments error")
        return
    }

    conn, err := amqp.Dial(config.RMQADDR)
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

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

    msgBody := os.Args[1]

    respQueue, err := ch.QueueDeclare(
        "",    // name
        false, // durable
        false, // delete when unused
        true,  // exclusive
        false, // noWait
        nil,
    )
    failOnError(err, "Failed to declare a response queue")

    correlationID := randomID(32)

    err = ch.Publish(
        "",               //exchange
        config.QUEUENAME, //routing key
        false,
        false,
        amqp.Publishing{
            ContentType:   "text/plain",
            CorrelationId: correlationID,
            ReplyTo:       respQueue.Name,
            Body:          []byte(msgBody),
        })

    log.Printf(" [x] Sent %s", msgBody)
    failOnError(err, "Failed to publish a message")

    respMsgs, err := ch.Consume(
        respQueue.Name,
        "",
        true,  // auto-ack
        true,  // exclusive
        false, // noLocal
        false, // nowait
        nil,
    )

    for item := range respMsgs {
        if item.CorrelationId == correlationID {
            fmt.Println("response:", string(item.Body))
            break
        }
    }
}

func failOnError(err error, msg string) {
    if err != nil {
        fmt.Printf("%s: %s\n", msg, err)
    }
}

func randomID(length int) string {
    if length <= 0 {
        return ""
    }

    bytes := make([]byte, length)
    for i := 0; i < length; i++ {
        bytes[i] = byte(rand.Intn(9))
    }

    return string(bytes)
}

server.go

package main

import (
    config "binTest/rabbitmqTest/t1/l6/conf"
    "fmt"
    "log"
    "math/rand"
    "time"

    "github.com/streadway/amqp"
)

func main() {

    conn, err := amqp.Dial(config.RMQADDR)
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

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

    err = ch.Qos(
        config.SERVERINSTANCESCNT,
        0,
        false,
    )

    forever := make(chan bool)

    for routine := 0; routine < config.SERVERINSTANCESCNT; routine++ {
        go func(routineNum int) {

            q, err := ch.QueueDeclare(
                config.QUEUENAME,
                false,
                false,
                false,
                false,
                nil,
            )
            failOnError(err, "Failed to declare a queue")

            msgs, err := ch.Consume(
                q.Name,
                "",
                false, // auto ack
                false,
                false,
                false,
                nil,
            )

            for msg := range msgs {
                log.Printf("In %d start consuming message: %s\n", routineNum, msg.Body)

                bookName := queryBookID(string(msg.Body))

                err = ch.Publish(
                    "",
                    msg.ReplyTo,
                    false,
                    false,
                    amqp.Publishing{
                        ContentType:   "text/plain",
                        CorrelationId: msg.CorrelationId,
                        Body:          []byte(bookName),
                    })

                if err != nil {
                    fmt.Println("Failed to reply msg to client")
                } else {
                    fmt.Println("Response to client:", bookName)
                }
                msg.Ack(false)
            }
        }(routine)
    }

    <-forever
}

func failOnError(err error, msg string) {
    if err != nil {
        fmt.Printf("%s: %s\n", msg, err)
    }
}

func queryBookID(bookID string) string {
    bookName := "QUERIED_" + bookID
    time.Sleep(time.Duration(rand.Intn(9)) * time.Second)

    return bookName
}

执行效果
Client端

Server端

全部代码可以在如下处取得
https://github.com/BinWang-sh...