【实践】消息队列RabbitMQ从入门安装到精通原理

apt-get install erlang-nox     # 安装erlang
erl    # 查看relang语言版本，成功执行则说明relang安装成功

wget -O- https://www.rabbitmq.com/rabbitmq-release-signing-key.asc | sudo apt-key add -

apt-get update

apt-get install erlang #安装erlang。
apt-get install rabbitmq-server  #安装成功自动启动。

rabbitmqctl status |grep rabbit #检查rabbit版本

systemctl status rabbitmq-server   #Active: active (running) 说明处于运行状态

# service rabbitmq-server status 用service指令也可以查看，同systemctl指令

service rabbitmq-server start    # 启动
service rabbitmq-server stop     # 停止
service rabbitmq-server restart  # 重启 

apt-get remove rabbitmq
apt-get purge erlang
apt-get autoremove

echo "deb http://packages.erlang-solutions.com/ubuntu trusty contrib" | tee -a /etc/apt/sources.list.d/erlang_solutions.list

wget -O- https://dl.bintray.com/rabbitmq/Keys/rabbitmq-release-signing-key.asc |apt-key add -

apt-get update 
apt-get install erlang-nox

wget -c https://bintray.com/rabbitmq/debian/download_file?file_path=pool%2Frabbitmq-server%2Frabbitmq-server_3.8.4-1_all.deb
mv download_file?file_path=pool%2Frabbitmq-server%2Frabbitmq-server_3.8.4-1_all.deb rabbitmq-server_3.8.4-1_all.deb

apt get install socat
dpkg -i ./rabbitmq-server_3.8.4-1_all.deb  

systemctl status rabbitmq-server   #Active: active (running) 说明处于运行状态

# service rabbitmq-server status 用service指令也可以查看，同systemctl指令

service rabbitmq-server start    # 启动
service rabbitmq-server stop     # 停止
service rabbitmq-server restart  # 重启 

rabbitmq-plugins enable rabbitmq_management   # 启用插件
service rabbitmq-server restart    # 重启

rabbitmqctl list_users

rabbitmqctl add_user admin [yourpassword]   # 增加普通用户，密码为admin
rabbitmqctl set_user_tags admin administrator    # 给普通用户分配管理员角色 

**端口信息：**

1. client端通信口amqp: 5672
2. http管理口15672
3. server间内部通信口25672
4. erlang发现口：4369

想要修改默认端口可修改，修改安装目录下 etc/rabbitmq.config文件。

**配置相关信息：**

=INFO REPORT==== 22-Jan-2021::11:32:41 ===
node           : rabbit@JD3
home dir       : /var/lib/rabbitmq
config file(s) : /etc/rabbitmq/rabbitmq.config
cookie hash    : VrYYogHAkySebZfCPZM8DQ==
log            : /var/log/rabbitmq/rabbit@JD3.log
sasl log       : /var/log/rabbitmq/rabbit@JD3-sasl.log
database dir   : /var/lib/rabbitmq/mnesia/rabbit@JD3

# A new style format snippet. This format is used by rabbitmq.conf files.
ssl_options.cacertfile           = /path/to/ca_certificate.pem
ssl_options.certfile             = /path/to/server_certificate.pem
ssl_options.keyfile              = /path/to/server_key.pem
ssl_options.verify               = verify_peer
ssl_options.fail_if_no_peer_cert = true

%% A classic format snippet, now used by advanced.config files.
[
  {rabbit, [{ssl_options, [{cacertfile,           "/path/to/ca_certificate.pem"},
                           {certfile,             "/path/to/server_certificate.pem"},
                           {keyfile,              "/path/to/server_key.pem"},
                           {verify,               verify_peer},
                           {fail_if_no_peer_cert, true}]}]}
].

go get github.com/streadway/amqp

package main

import (
    "fmt"
    "log"

    "github.com/streadway/amqp"
)

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

conn, err := amqp.Dial("amqp://guest:guest@localhost:5672")
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(
    "hello", //name
    false,  //durable
    false,  //delete when unused
    false,  //exclusive
    false,  //no wait
    nil,    //arguments
)
failOnError(err, "Failed to declare q queue")

body := "Hello"
err = ch.Publish(
    "",     //exchange
    q.Name,  // routing key
    false,  //mandatory
    false, //immediate
    amqp.Publishing{
        ContentType: "text/plain",
        Body :      []byte(body),
    }
) 

failOnError(err, "Failed to publish a message")

package main

import {
    "fmt"
    "log"

    "github.com/streadway/amqp"
}

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

conn, err := amqp.Dial("amqp://guest:guest@localhost:5672")
failOnError(err, "Failed to connect to server")
defer conn.Close();

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

q, err := ch.QueueDeclare(
    "hello",    //name
    false,      //durable
    false,      //delete when usused
    false,      // exclusive
    false,      //no-wait
    nil,        // arguments
)

failOnError(err, "Failed to declare a queue")

msgs, err := ch.Consume(
    q.Name,     // queue
    "",         // consumer
    true,       // auto-ack
    false,      // exclusive
    false,      // no-local
    false,      // no-wait
    nil,        // arguments
)
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

func (ch *Channel) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args Table) (<-chan Delivery, error)

 go run receive.go

 go run send.go

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

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)
        dot_count := bytes.Count(d.Body, []byte("."))
        t := time.Duration(dot_count)
        time.Sleep(t* time.Second)
        log.Printf("Done")
    }
}()

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

#shell 1
go run worker.go

#shell 2
go run new_task.go First message.
go run new_task.go Second message..
go run new_task.go Third message...
go run new_task.go Fourth message....
go run new_task.go Fifth message.....

#shell 1
go run worker.go


#shell 2
go run worker.go

#shell 3
go run new_task.go First message..
go run new_task.go Second message....
go run new_task.go Third message......
go run new_task.go Fourth message........
go run new_task.go Fifth message..........
...

msgs, err := ch.Consume(
    q.Name, // queue
    "", // consumer
    false, // 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)
        dot_count := bytes.Count(d.Body, []byte("."))
        t := time.Duration(dot_count)
        time.Sleep(t * time.Second)
        log.Printf("Done")
        d.Ack(false)
    }
}()

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

# rabbitmqctl list_queues
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
name    messages
task_queue  5
hello   0

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

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

err = ch.Publishing(
    "",         //exchange
    q.Name,     // routing key
    false,      // mandatory
    false,
    amqp.Publishing{
        DeliveryMode: amqp.Persistent,
        ContentType:    "text/plain",
        Body:           []byte(body),
    }
)

err = ch.Qos(
    1,      // prefetch count
    0,      // prefetch size
    false,  // global
)
failOnError(err, "Failed to set Qos")

package main

import (
        "fmt"
        "log"
        "os"
        "strings"

        "github.com/streadway/amqp"
)

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

func main() {
        conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
        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(
                "task_queue", // name
                true,         // durable
                false,        // delete when unused
                false,        // exclusive
                false,        // no-wait
                nil,          // arguments
        )
        failOnError(err, "Failed to declare a queue")

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

func bodyFrom(args []string) string {
        var s string
        if (len(args) < 2) || os.Args[1] == "" {
                s = "hello"
        } else {
                s = strings.Join(args[1:], " ")
        }
        return s
}

package main

import (
        "bytes"
        "fmt"
        "github.com/streadway/amqp"
        "log"
        "time"
)

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

func main() {
        conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
        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(
                "task_queue", // name
                true,         // durable
                false,        // delete when unused
                false,        // exclusive
                false,        // no-wait
                nil,          // arguments
        )
        failOnError(err, "Failed to declare a queue")

        err = ch.Qos(
                1,     // prefetch count
                0,     // prefetch size
                false, // global
        )
        failOnError(err, "Failed to set QoS")

        msgs, err := ch.Consume(
                q.Name, // queue
                "",     // consumer
                false,  // 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)
                        dot_count := bytes.Count(d.Body, []byte("."))
                        t := time.Duration(dot_count)
                        time.Sleep(t * time.Second)
                        log.Printf("Done")
                        d.Ack(false)
                }
        }()

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

err = ch.ExchangeDeclare(
    "logs",     //name
    "fanout",   //type
    "true",     //durable
    false,      //auto-deleted
    false,      //internal
    false,      //no-wait
    nil,        //arguments
)

>     err = ch.Publish(
>          "",     //exchange
>          q.Name, //routing key
>          false,  //mandatory
>          false,  //immediate
>          amqp.Publishing(
>              ContentType: "text/plain",
>              Body:        []byte(body),
>          )
>     )

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

body:= bodyForm(os.Args)
err = ch.Publish(
    "logs",     //exchange
    "",         //routing key
    false,      //mandatory
    false,      //immediate
    amqp.Publishing(
        ContentType: "text/plain",
        Body:       []byte(body),
    )
)

q, err := ch.QueueDeclare(
    "",     //name
    false,  //durable
    false,  //delete when unused
    true,   //exclusive
    false,  //no-wait
    nil,    //arguments
)

 err = ch.QueueBind(
     q.Name,    //queue name
     "",        //routing key
     "logs",    //exchange
     false,
     nil
 )

package main

import (
        "fmt"
        "log"
        "os"
        "strings"

        "github.com/streadway/amqp"
)

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

func main() {
        conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
        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.ExchangeDeclare(
                "logs",   // name
                "fanout", // type
                true,     // durable
                false,    // auto-deleted
                false,    // internal
                false,    // no-wait
                nil,      // arguments
        )
        failOnError(err, "Failed to declare an exchange")

        body := bodyFrom(os.Args)
        err = ch.Publish(
                "logs", // exchange
                "",     // routing key
                false,  // mandatory
                false,  // immediate
                amqp.Publishing{
                        ContentType: "text/plain",
                        Body:        []byte(body),
                })
        failOnError(err, "Failed to publish a message")

        log.Printf(" [x] Sent %s", body)
}

func bodyFrom(args []string) string {
        var s string
        if (len(args) < 2) || os.Args[1] == "" {
                s = "hello"
        } else {
                s = strings.Join(args[1:], " ")
        }
        return s
}

package main

import (
        "fmt"
        "log"

        "github.com/streadway/amqp"
)

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

func main() {
        conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
        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.ExchangeDeclare(
                "logs",   // name
                "fanout", // type
                true,     // durable
                false,    // auto-deleted
                false,    // internal
                false,    // no-wait
                nil,      // arguments
        )
        failOnError(err, "Failed to declare an exchange")

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

        err = ch.QueueBind(
                q.Name, // queue name
                "",     // routing key
                "logs", // 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(" [x] %s", d.Body)
                }
        }()

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

go run receive_logs.go > logs_from_rabbit.log

go run receive_logs.go

go run emit_log.go

sudo rabbitmqctl list_bindings
# => Listing bindings ...
# => logs    exchange        amq.gen-JzTY20BRgKO-HjmUJj0wLg  queue           []
# => logs    exchange        amq.gen-vso0PVvyiRIL2WoV3i48Yg  queue           []
# => ...done.

err = ch.QueueBind(
    q.Name,     //queue name
    "",         //routing key
    "logs",     //exchange
    false,
    nil)

err = ch.QueueBind(
    q.Name,     //queue name
    "black",    //routing key
    "logs",     //exchange
    false,
    nil)

err = ch.ExchangeDeclare(
    "logs_direct",  //name
    "direct",       //type
    true,           //durable
    false,          //auto-deleted
    false,          //internal
    false,          //no-wait
    nil,            //arguments
)

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

body := bodyFrom(os.Args)
err = ch.Publish(
    "logs_direct",      //exchange
    severityFrom(os.Args),  //routing key
    false,              //mandatory
    false,              //immediate
    amqp.Publishing{
        ContentType: "text/plain",
        Body:          []byte(body),
    }
)

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

if len(os.Args) < 2 {
    log.Printf("Usage: %s [info] [warnint] [error]", os.Args[0])
    os.Exit(0)
}

for _, s := range os.Args[1:] {
    log.Printf("Binding queue %s to exchange %s with routing key %s",
        q.Name, "logs_direct", s)
    err = ch.QueueBind(
        q.Name,     // queue name
        s,          // routing key
        "logs_direct", //exchange
        false,
        nil)
    failOnError(err, "Failed to bind a queue")
}

package main

import (
    "fmt"
    "log"
    "os"
    "strings"

    "github.com/streadway/amqp"
)

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

func main(){
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

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

    err = ch.ExchangeDeclare(
        "logs_direct",     //name
        "direct",         //type
        true,    
        false,
        false,
        false,
        nil,
    )
    failOnError(err, "Failed to declare an exchange")

    body := bodyFrom(os.Args)
    err = ch.Publish(
        "logs_direct",            // exchange
        severityFrom(os.Args),    //routing key
        false,
        false,
        amqp.Publishing{
            ContentType: "text/plain",
            Body:        []byte(body),
        })
    failOnError(err, "Failed to publish a message")

    log.Printf(" [x] sent %s", body)
}

func bodyFrom(args []string) string{
    var s string
    if(len(args) < 3) || os.Args[2] == "" {
        s = "hello"
    }else{
        s = strings.Join(args[2:], " ")
    }

    return s
}

func severityFrom(args []string) string {
    var s string
    if len(args) < 2 || args[1] == "" {
        s = "info"
    }else {
        s = os.Args[1]
    }

    return s
}

package main

import(
    "fmt"
    "log"
    "os"

    "github.com/streadway/amqp"
)

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

func main(){
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    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.ExchangeDeclare(
        "logs_direct",
        "direct",
        true,
        false, 
        false,
        false,
        nil,
    )
    failOnError(err, "Failed to declare an exchange")

    q, err := ch.QueueDeclare(
        "",     //name
        false,
        false,
        true,
        false,
        nil,
    )
    failOnError(err, "Failed to declare a queue")

    if len(os.Args) < 2 {
        log.Printf("Usage: %s [info] [warning] [error]")
        os.Exit(0)
    }

    for _, s := range os.Args[1:] {
        log.Printf("Binding queue %s to exchange %s with routing key %s",
        q.Name, "logs_direct", s)
        err = ch.QueueBind(
            q.Name, //queue name
            s,                //routing key
            "logs_direct",    //exchange
            false,
            nil,
        )
        failOnError(err, "Failed to bind a queue")
    }

    msgs, err := ch.Consume(
        q.Name,     //name
        "",         //consumer
        true,
        false,
        false,
        false,
        nil,
    )
    failOnError(err, "Failed to register a consumer")

    forever := make(chan bool)

    go func(){
        for d:= range msgs{
            log.Printf(" [x] %s", d.Body)
        }
    }()

    log.Printf(" [*] Waiting for logs. To Exit press Ctrl+c")
    <-forever
}

go run receive_logs_direct.go warning error>logs_from_rabbit.log

go run receive_logs_direct.go warning error info

go run emit_log_direct.go error "this is a log message"

*   表示一个单词
#   表示0或多个单词

Q1对orange颜色的动物感兴趣；
Q2则对所有物种是rabbit的、速度是lazy的所有动物感兴趣；

"quick.orange.rabbit":      分发到Q1和Q2；
"lazy.orange.elephant":     分发到Q1和Q2;
"quick.orange.fox":         分发到Q1；
"lazy.brown.fox":           分发到Q2;
"lazy.pink.rabbit":         分发到Q2，且只会分发一次，虽然它匹配了Q2的两个绑定；
"quick.brown.fox":          Q1、Q2都不分发，因为没有与之匹配的绑定规则，将会被忽略；
"orange":                   无匹配规则，丢失
"quick.orange.male.rabbit": 无匹配规则，丢失
"lazy.orange.male.rabbit":  分发到Q2，虽然有4个单词，但符合"lazy.#"规则的通配符

package main

import (
    "fmt"
    "log"
    "os"
    "strings"

    "github.com/streadway/amqp"
)

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

func main(){
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

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

    err = ch.ExchangeDeclare(
        "logs_topic",     //name
        "topic",         //type
        true,    
        false,
        false,
        false,
        nil,
    )
    failOnError(err, "Failed to declare an exchange")
    body := bodyFrom(os.Args)
    err = ch.Publish(
        "logs_topic",            // exchange
        severityFrom(os.Args),    //routing key
        false,
        false,
        amqp.Publishing{
            ContentType: "text/plain",
            Body:        []byte(body),
        })
    failOnError(err, "Failed to publish a message")

    log.Printf(" [x] sent %s", body)
}
func bodyFrom(args []string) string{
    var s string
    if(len(args) < 3) || os.Args[2] == "" {
        s = "hello"
    }else{
        s = strings.Join(args[2:], " ")
    }

    return s
}

func severityFrom(args []string) string {
    var s string
    if len(args) < 2 || args[1] == "" {
        s = "info"
    }else {
        s = os.Args[1]
    }

    return s
}

import(
    "fmt"
    "log"
    "os"

    "github.com/streadway/amqp"
)

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

func main(){
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    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.ExchangeDeclare(
        "logs_topic",
        "topic",
        true,
        false, 
        false,
        false,
        nil,
    )
    failOnError(err, "Failed to declare an exchange")

    q, err := ch.QueueDeclare(
        "",     //name
        false,
        false,
        true,
        false,
        nil,
    )
    failOnError(err, "Failed to declare a queue")

    if len(os.Args) < 2 {
        log.Printf("Usage: %s [info] [warning] [error]", os.Args[0])
        os.Exit(0)
    }

    for _, s := range os.Args[1:] {
        log.Printf("Binding queue %s to exchange %s with routing key %s",
            q.Name, "logs_topic", s)
        err = ch.QueueBind(
            q.Name, 
            s,
            "logs_topic",
            false,
            nil)
        failOnError(err, "Failed to bind a queue")
    }

    msgs, err := ch.Consume(
        q.Name,
        "",
        true,
        false,
        false, 
        false,
        nil,
    )
    failOnError(err, "Failed to register a consumer")

    forever := make(chan bool)

    go func(){
        for d:= range msgs {
            log.Printf(" [x] %s", d.Body)
        }
    }()

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

go run receive_logs_topic.go "#"

go run receive_logs_topic.go "kern.*"

go run receive_logs_topic.go "*.critical"

go run receive_logs_topic.go "kern.*" "*.critical"

go run emit_log_topic.go "kern.critical" "A critical kernal error"

q, err := ch.QueueDeclare(
    "",     //name
    false,  //durable
    false,  //delete when usused
    true,   //exclusive
    false,  //nowait
    nil,    //argments
)

err = ch.Publish(
    "",     //exchange
    "rpc_queue" //routing key
    false,  //mandatory
    false,  //immediate
    amqp.Publishing{
        ContentType:    "text/plain",
        CorrelationId:  corrId,
        ReplyTo:        q.Name,
        Body:           []byte(strconv.Itoa(n)),
    }
)

当Client启动时，会创建一个匿名的、独有的回调队列；

对每一个RPC Request，Client都会设置两个参数：用于标识回调队列的reply_to和用于唯一标识的correlation_id;

Request被发送到rpc_queue队列。

RPC服务器等待rpc_queue的消息，一旦消息到达，处理任务后将响应结果消息发送到reply_to指定的队列；

Client等待callback队列的消息，一旦消息到达，查找与correlation_id匹配的request，然后返回给应用程序。

func fib(n int) int {
    if n== 0 {
        return 0
    }else if n==1 {
        return 1
    }else {
        return fib(n-1) + fib(n-2)
    }
}

package main

import (
    "fmt"
    "log"
    "strconv"

    "github.com/streadway/amqp"
)

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

func fib(n int) int {
    if n== 0 {
        return 0
    }else if n==1 {
        return 1
    }else {
        return fib(n-1) + fib(n-2)
    }
}

func main(){
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    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(
        "rpc_queue",     //name
        false,            //durables
        false,            //delete when unused
        false,            //exclusive
        false,            //no wait
        nil,            //args
    )
    failOnError(err, "Failed to declare a queue")

    err = ch.Qos(
            1,     // prefetch count
            0,  // prefetch size
            false,    //global
    )
    failOnError(err, "Failed to set Qos")

    msgs, err := ch.Consume(
        q.Name,        //queue
        "",            //exchange
        false,        // 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 {
            n, err := strconv.Atoi(string(d.Body))
            failOnError(err, "Failed to convert body to an integer")

            log.Printf(" [.] fib(%d)", n)
            response := fib(n)

            err = ch.Publish(
                "",        //exchange
                d.ReplyTo,        //routing key
                false,        //mandatory
                false,        //immediate
                amqp.Publishing{
                    ContentType :    "text/plain",
                    CorrelationId:    d.CorrelationId,
                    Body:            []byte(strconv.Itoa(response)),
                })
            failOnError(err, "Failed to publish a message")

            d.Ack(false)
        }
    }()

    log.Printf(" [*] Awaiting RPC reqeusts")
    <-forever
}

package main

import (
    "fmt"
    "log"
    "strconv"
    "os"
    "math/rand"
    "strings"
    "time"

    "github.com/streadway/amqp"
)

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

func randomString(l int) string {
    bytes := make([]byte, l)
    for i:=0; i<l; i++ {
        bytes[i] = byte(randInt(65, 90))
    }
    return string(bytes)
}

func randInt(min int, max int) int {
    return min + rand.Intn(max - min)
}

func bodyFrom(args []string) int {
    var s string
    if(len(args) < 2 || os.Args[1]==""){
        s = "30"
    }else{
        s = strings.Join(args[1:], " ")
    }

    n, err := strconv.Atoi(s)
    failOnError(err, "Failed to convert arg to integer")
    return n
}

func fibonacciRPC(n int) (res int, err error) {
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    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(
        "",             //name
        false,            //durables
        false,            //delete when unused
        true,            //exclusive
        false,            //no wait
        nil,            //args
    )
    failOnError(err, "Failed to declare a queue")

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

    corrId := randomString(32)

    err = ch.Publish(
        "",            //exchange
        "rpc_queue",    //routing key
        false,        //mandatory
        false,        //immediate
        amqp.Publishing{
            ContentType:        "text/plain",
            CorrelationId:        corrId,
            ReplyTo:            q.Name,
            Body:                []byte(strconv.Itoa(n)),
        })
    failOnError(err, "Failed to publish a message")

    for d:= range msgs {
        if corrId == d.CorrelationId {
            res, err = strconv.Atoi(string(d.Body))
            failOnError(err, "Failed to convert body to integer")
            break
        }
    }

    return
}

func main(){
    rand.Seed(time.Now().UTC().UnixNano())

    n:= bodyFrom(os.Args)

    log.Printf(" [x] Requesting fib(%d)", n)
    res, err := fibonacciRPC(n)
    failOnError(err, "Failed to handle RPC request")

    log.Printf(" [.] Got %d", res)
}

go run rpc_server.go
# => [x] Awaiting RPC requests

go run rpc_client.go 30
# => [x] Requesting fib(30)

* 如果单台RPC服务器性能缓慢，可以很容易的进行扩展，只需在新窗口运行一个rpc_server.go脚本即可；

* 在客户端，RPC模式要求对消息进行一次发送和接收操作，因此只需要一次网络往返即可完成一次RPC请求；

* 客户端如何处理服务端掉线的情况？
* 客户端如何处理服务端超时情况？
* 如果服务端故障导致异常，是否需要将异常转发给客户端处理？
* 在对消息处理前未对消息的合法性进行检查，如边界值、类型信息等；

> rabbitmqctl add_vhost qa1