手册:http://www.rabbitmq.com/getstarted.html

安装:http://www.rabbitmq.com/download.html

参考：http://blog.csdn.net/whycold/article/details/41119807

一.介绍

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。

AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Go、Python、Ruby。用于在分布式系统中存储转发消息。

二.安装

ubuntu直接下载deb文件安装，默认已经启动，sudo敲入：

sudo  rabbitmq-server start
sudo lsof -i:5672

启用插件，进入UI：

sudo rabbitmq-plugins enable rabbitmq_management

用户名:密码=guest:guest

三.使用

# 敲入查看帮助
sudo rabbitmqctl

# 创建用户
sudo rabbitmqctl add_user  登录用户名  密码

# 可以创建管理员用户，负责整个MQ的运维
sudo rabbitmqctl set_user_tags 登录用户名 administrator
# 可以创建RabbitMQ监控用户，负责整个MQ的监控
sudo rabbitmqctl set_user_tags 登录用户名 monitoring
# 可以创建某个项目的专用用户，只能访问项目自己的virtual hosts
sudo rabbitmqctl set_user_tags 登录用户名 management

# 查看用户
sudo rabbitmqctl list_users 

# 授权
# 该命令使用户具有/这个virtual host中所有资源的配置、写、读权限以便管理其中的资源
# set_permissions [-p <vhostpath>] <user> <conf> <write> <read>
# 其中，<conf> <write> <read>的位置分别用正则表达式来匹配特定的资源，如‘^(amq\.gen.*|amq\.default)$‘可以匹配server生成的和默认的exchange，‘^$‘不匹配任何资源
sudo rabbitmqctl  set_permissions -p / 登录用户名 ‘.*‘ ‘.*‘ ‘.*‘

四.概念入门

1.ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。

Connection是RabbitMQ的socket链接，它封装了socket协议相关部分逻辑。

ConnectionFactory为Connection的制造工厂。

Channel是我们与RabbitMQ打交道的最重要的一个接口，我们大部分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

2.Queue

Queue（队列）是RabbitMQ的内部对象，用于存储消息，用下图表示。

RabbitMQ中的消息都只能存储在Queue中，生产者（下图中的P）生产消息并最终投递到Queue中，消费者（下图中的C）可以从Queue中获取消息并消费。

多个消费者可以订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。

3.消息的一些机制

3.1.消息确认Message acknowledgment

在实际应用中，可能会发生消费者收到Queue中的消息，但没有处理完成就宕机（或出现其他意外）的情况，这种情况下就可能会导致消息丢失。为了避免这种情况发生，我们可以要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message acknowledgment）后才将该消息从Queue中移除；如果RabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开，则RabbitMQ会将该消息发送给其他消费者（如果存在多个消费者）进行处理。这里不存在timeout概念，一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者，除非它的RabbitMQ连接断开。

这里会产生另外一个问题，如果我们的开发人员在处理完业务逻辑后，忘记发送回执给RabbitMQ，这将会导致严重的bug——Queue中堆积的消息会越来越多；消费者重启后会重复消费这些消息并重复执行业务逻辑…

另外pub message是没有ack的。（？？）

3.2.消息持久Message durability

如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。但依然解决不了小概率丢失事件的发生（比如RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），如果我们需要对这种小概率事件也要管理起来，那么我们要用到事务。由于这里仅为RabbitMQ的简单介绍，所以这里将不讲解RabbitMQ相关的事务。

3.3.提前取机制Prefetch count

前面我们讲到如果有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时如果每个消息的处理时间不同，就有可能会导致某些消费者一直在忙，而另外一些消费者很快就处理完手头工作并一直空闲的情况。我们可以通过设置prefetchCount来限制Queue每次发送给每个消费者的消息数，比如我们设置prefetchCount=1，则Queue每次给每个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。就是变慢而已。订阅模式如何平摊？这种模式是一个消费者一次性拿很多条消息？

4.Exchange

在上一节我们看到生产者将消息投递到Queue中，实际上这在RabbitMQ中这种事情永远都不会发生。实际的情况是，生产者将消息发送到Exchange（交换器，下图中的X），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。

Exchange是按照什么逻辑将消息路由到Queue的？这个将在Binding一节介绍。

RabbitMQ中的Exchange有四种类型，不同的类型有着不同的路由策略，这将在Exchange Types一节介绍。

5.Routing key

生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，而这个routing key需要与Exchange Type及binding key联合使用才能最终生效。

在Exchange Type与binding key固定的情况下（在正常使用时一般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。

RabbitMQ为routing key设定的长度限制为255 bytes。

6.Binding

RabbitMQ中通过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。

6.1.Binding key

在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key；消费者将消息发送给Exchange时，一般会指定一个routing key；当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。这个将在Exchange Types章节会列举实际的例子加以说明。

在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的binding key。 binding key 并不是在所有情况下都生效，它依赖于Exchange Type，比如fanout类型(广播)的Exchange就会无视binding key，而是将消息路由到所有绑定到该Exchange的Queue。

7.Exchange Types

RabbitMQ常用的Exchange Type有fanout、direct、topic、headers这四种（AMQP规范里还提到两种Exchange Type，分别为system与自定义，这里不予以描述），下面分别进行介绍。

7.1.fanout

fanout类型的Exchange路由规则非常简单，它会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中。

7.2.direct

direct类型的Exchange路由规则也很简单，它会把消息路由到那些binding key与routing key完全匹配的Queue中。

以上图的配置为例，我们以routingKey=”error”发送消息到Exchange，则消息会路由到Queue1（amqp.gen-S9b…，这是由RabbitMQ自动生成的Queue名称）和Queue2（amqp.gen-Agl…）；如果我们以routingKey=”info”或routingKey=”warning”来发送消息，则消息只会路由到Queue2。如果我们以其他routingKey发送消息，则消息不会路由到这两个Queue中。

7.3.topic

前面讲到direct类型的Exchange路由规则是完全匹配binding key与routing key，但这种严格的匹配方式在很多情况下不能满足实际业务需求。topic类型的Exchange在匹配规则上进行了扩展，它与direct类型的Exchage相似，也是将消息路由到binding key与routing key相匹配的Queue中，但这里的匹配规则有些不同，它约定：

1. routing key为一个句点号“. ”分隔的字符串（我们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”
2. binding key与routing key一样也是句点号“. ”分隔的字符串
3. binding key中可以存在两种特殊字符“*”与“#”，用于做模糊匹配，其中“*”用于匹配一个单词，“#”用于匹配多个单词（可以是零个）

以上图中的配置为例，routingKey=”quick.orange.rabbit”的消息会同时路由到Q1与Q2，routingKey=”lazy.orange.fox”的消息会路由到Q1与Q2，routingKey=”lazy.brown.fox”的消息会路由到Q2，routingKey=”lazy.pink.rabbit”的消息会路由到Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配）；routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的消息将会被丢弃，因为它们没有匹配任何bindingKey

7.4.headers

headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。

在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。

该类型的Exchange没有用到过（不过也应该很有用武之地），所以不做介绍。

8.RPC

MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）。

但实际的应用场景中，我们很可能需要一些同步处理，需要同步等待服务端将我的消息处理完成后再进行下一步处理。这相当于RPC（Remote Procedure Call，远程过程调用）。在RabbitMQ中也支持RPC。

RabbitMQ中实现RPC的机制是：

客户端发送请求（消息）时，在消息的属性（MessageProperties，在AMQP协议中定义了14中properties，这些属性会随着消息一起发送）中设置两个值replyTo（一个Queue名称，用于告诉服务器处理完成后将通知我的消息发送到这个Queue中）和correlationId（此次请求的标识号，服务器处理完成后需要将此属性返还，客户端将根据这个id了解哪条请求被成功执行了或执行失败）
服务器端收到消息并处理
服务器端处理完消息后，将生成一条应答消息到replyTo指定的Queue，同时带上correlationId属性
客户端之前已订阅replyTo指定的Queue，从中收到服务器的应答消息后，根据其中的correlationId属性分析哪条请求被执行了，根据执行结果进行后续业务处理

五.Go 接口

http://www.rabbitmq.com/tutorials/tutorial-one-go.html

请看官方示例：

https://github.com/rabbitmq/rabbitmq-tutorials/blob/master/go/send.go

上面的例子仔细看，有必要看源码！

GO接口库

六.实例解释

四种模式

DIRECT 默认点对点模式
TOPIC 话题模式
FANOUT 广播模式
RPC RPC模式

工作队列：默认点对点模式

发布方，一个！

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:[email protected]: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()

        // 申明一个队列
        // https://godoc.org/github.com/streadway/amqp#Channel.QueueDeclare
        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 默认模式，exchange为空
                q.Name,       // routing key 默认模式路由到同名队列，即是task_queue
                false,        // mandatory
                false,
                amqp.Publishing{
                        // 持久性的发布，因为队列被声明为持久的，发布消息必须加上这个（可能不用），但消息还是可能会丢，如消息到缓存但MQ挂了来不及持久化。
                        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:[email protected]: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")

        // Fair dispatch 预取，每个工作方每次拿一个消息，确认后才拿下一次，缓解压力
        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")

                        // 确认消息被收到！！如果为真的，那么同在一个channel,在该消息之前未确认的消息都会确认，适合批量处理
                        // 真时场景：每十条消息确认一次，类似
                        d.Ack(false)
                }
        }()

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

发布-订阅：广播模式

发布方

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:[email protected]: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:[email protected]: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
}

高级路由发布-订阅新版：默认点对点模式

发布-订阅每个绑定的队列都收到一样的消息，现在不想！使用路由功能，队列绑定进行分发。

发布方

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:[email protected]: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", // name
                "direct",      // 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_direct",         // exchange   发到这个交换机
                severityFrom(os.Args), // routing key  且路由key是由命令行指定，如下方，指定了error
                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) < 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) || os.Args[1] == "" {
                s = "info"
        } else {
                s = os.Args[1]
        }
        return s
}

发消息到交换机，路由key为error

go run *.go error "Run. Run. Or it will explode."

消费方

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:[email protected]: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", // name
                "direct",      // 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")

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

        # 绑定队列和交换机，绑定多个路由key，见下方
        for _, s := range os.Args[1:] {
                log.Printf("Binding queue %s to exchange %s with routing key %s",
                        q.Name, "logs_direct", s)

                // 下面同个队列可以收到不同路由key的消息 ，广播模式除外！
                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, // 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
}

消费这些key:info warning error

go run *.go info warning error

话题发布-订阅新新版:话题模式

上面的路由都是标准的，就是固定字符串名字，话题模式可以使用类正则的路由，这样模糊匹配更棒！！

路由类似于这样 *.love.*

* (star) can substitute for exactly one word.
# (hash) can substitute for zero or more words.

发布方

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:[email protected]: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", // name
                "topic",      // 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_topic",          // exchange
                severityFrom(os.Args), // 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) < 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) || os.Args[1] == "" {
                s = "anonymous.info"
        } else {
                s = os.Args[1]
        }
        return s
}

命令运行

To receive all the logs:

go run *.go "#"

To receive all logs from the facility “kern”:

go run *.go "kern.*"

Or if you want to hear only about “critical” logs:

go run *.go "*.critical"

You can create multiple bindings:

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

消费方

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:[email protected]: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", // name
                "topic",      // 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")

        if len(os.Args) < 2 {
                log.Printf("Usage: %s [binding_key]...", 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,       // queue name
                        s,            // routing key
                        "logs_topic", // 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 *.go "kern.critical" "A critical kernel error"

rpc：RPC模式

应答方

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:[email protected]:5672/")
        failOnError(err, "Failed to connect to RabbitMQ")
        defer conn.Close()

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

        // 声明匿名队列
        q, err := ch.QueueDeclare(
                "rpc_queue", // name
                false,       // durable
                false,       // delete when usused
                false,       // exclusive
                false,       // no-wait
                nil,         // arguments
        )
        failOnError(err, "Failed to declare a queue")

        // 公平分发 没有这个则round-robbin：https://segmentfault.com/a/1190000004492447
        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 {
                        n, err := strconv.Atoi(string(d.Body))
                        failOnError(err, "Failed to convert body to 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 requests")
        <forever
}

请教方

package main

import (
        "fmt"
        "log"
        "math/rand"
        "os"
        "strconv"
        "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 fibonacciRPC(n int) (res int, err error) {
        // 拨号
        conn, err := amqp.Dial("amqp://guest:[email protected]: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, // durable
                false, // delete when usused
                true,  // exclusive 为真即连接断开就删除
                false, // noWait
                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,    // args
        )
        failOnError(err, "Failed 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)
}

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
}

http://www.rabbitmq.com/tutorials/tutorial-six-go.html

七.属性详解，测试！

package main

import (
	"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))
	}
}

// Queue属性测试
//
// durable属性和auto-delete属性可以同时生效；
// durable属性和exclusive属性会有性质上的冲突,两者同时设置时,仅exclusive属性生效；
// auto_delete属性和exclusive属性可以同时生效；
//
// auto_delete如果有连接存在消费者订阅该http://www.lenggirl.com/tool/RabbitMQ.htmlqueue,正常,如果消费者全部消失,自动删除队列
// 可以在没有创建consumer的情况下,创建出具有auto-delete属性的queue。
//
// exclusive,如果声明该队列的连接断开,自动删除队列
// queue的存在条件是在声明该队列的连接上存在某个consumer订阅了该queue。
//
func main() {
	conn, err := amqp.Dial("amqp://guest:[email protected]: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()

	// 默认模式有默认交换机,广播自己定义一个交换机,交换机可与队列进行绑定
	/*

		ExchangeDeclare declares an exchange on the server. If the exchange does not
		already exist, the server will create it.  If the exchange exists, the server
		verifies that it is of the provided type, durability and auto-delete flags.

        ExchangeDeclare方法在服务器声明一个exchange。如果不存在，新建一个，存在的话则确认type和durability和auto-delete的标志是否一致。

		Errors returned from this method will close the channel.

		如果方法返回错误，channel会被关闭。

		Exchange names starting with "amq." are reserved for pre-declared and
		standardized exchanges. The client MAY declare an exchange starting with
		"amq." if the passive option is set, or the exchange already exists.  Names can
		consists of a non-empty sequence of letters, digits, hyphen, underscore,
		period, or colon.

		名字以"amq."开头的Exchange是为之前已经声明和标准化的exchange们保留的，
		在exchange已经存在的情况下，或者passive选项设置为真，客户端才有可能声明一个这样的exchange。

		exchange的名字是一个非空序列，仅能包含字母，数字，连字符-，下划线_，句号.，冒号：

		另外的方法ExchangeDeclarePassive主要用来检测exchange是否已经存在。

		Each exchange belongs to one of a set of exchange kinds/types implemented by
		the server. The exchange types define the functionality of the exchange - i.e.
		how messages are routed through it. Once an exchange is declared, its type
		cannot be changed.  The common types are "direct", "fanout", "topic" and
		"headers".

		Durable and Non-Auto-Deleted exchanges will survive server restarts and remain
		declared when there are no remaining bindings.  This is the best lifetime for
		long-lived exchange configurations like stable routes and default exchanges.

		Non-Durable and Auto-Deleted exchanges will be deleted when there are no
		remaining bindings and not restored on server restart.  This lifetime is
		useful for temporary topologies that should not pollute the virtual host on
		failure or after the consumers have completed.

		Non-Durable and Non-Auto-deleted exchanges will remain as long as the server is
		running including when there are no remaining bindings.  This is useful for
		temporary topologies that may have long delays between bindings.

		Durable and Auto-Deleted exchanges will survive server restarts and will be
		removed before and after server restarts when there are no remaining bindings.
		These exchanges are useful for robust temporary topologies or when you require
		binding durable queues to auto-deleted exchanges.

		Note: RabbitMQ declares the default exchange types like ‘amq.fanout‘ as
		durable, so queues that bind to these pre-declared exchanges must also be
		durable.

		Exchanges declared as `internal` do not accept accept publishings. Internal
		exchanges are useful for when you wish to implement inter-exchange topologies
		that should not be exposed to users of the broker.

		When noWait is true, declare without waiting for a confirmation from the server.
		The channel may be closed as a result of an error.  Add a NotifyClose listener
		to respond to any exceptions.

		Optional amqp.Table of arguments that are specific to the server‘s implementation of
		the exchange can be sent for exchange types that require extra parameters.

		func (me *Channel) ExchangeDeclare(name, kind string, durable, autoDelete, internal, noWait bool, args Table) error {
	*/
	err = ch.ExchangeDeclare("logs", "fanout", true, false, false, false, nil)
	failOnError(err, "Failed to declare an exchange")

	// test-1 无属性,机器重启删除
	/*
		QueueDeclare declares a queue to hold messages and deliver to consumers.
		Declaring creates a queue if it doesn‘t already exist, or ensures that an
		existing queue matches the same parameters.

		Every queue declared gets a default binding to the empty exchange "" which has
		the type "direct" with the routing key matching the queue‘s name.  With this
		default binding, it is possible to publish messages that route directly to
		this queue by publishing to "" with the routing key of the queue name.

		  QueueDeclare("alerts", true, false, false, false, nil)
		  Publish("", "alerts", false, false, Publishing{Body: []byte("...")})

		  Delivery       Exchange  Key       Queue
		  -----------------------------------------------
		  key: alerts -> ""     -> alerts -> alerts

		The queue name may be empty, in which the server will generate a unique name
		which will be returned in the Name field of Queue struct.

		Durable and Non-Auto-Deleted queues will survive server restarts and remain
		when there are no remaining consumers or bindings.  Persistent publishings will
		be restored in this queue on server restart.  These queues are only able to be
		bound to durable exchanges.

		Non-Durable and Auto-Deleted queues will not be redeclared on server restart
		and will be deleted by the server after a short time when the last consumer is
		canceled or the last consumer‘s channel is closed.  Queues with this lifetime
		can also be deleted normally with QueueDelete.  These durable queues can only
		be bound to non-durable exchanges.

		Non-Durable and Non-Auto-Deleted queues will remain declared as long as the
		server is running regardless of how many consumers.  This lifetime is useful
		for temporary topologies that may have long delays between consumer activity.
		These queues can only be bound to non-durable exchanges.

		Durable and Auto-Deleted queues will be restored on server restart, but without
		active consumers, will not survive and be removed.  This Lifetime is unlikely
		to be useful.

		Exclusive queues are only accessible by the connection that declares them and
		will be deleted when the connection closes.  Channels on other connections
		will receive an error when attempting declare, bind, consume, purge or delete a
		queue with the same name.

		When noWait is true, the queue will assume to be declared on the server.  A
		channel exception will arrive if the conditions are met for existing queues
		or attempting to modify an existing queue from a different connection.

		When the error return value is not nil, you can assume the queue could not be
		declared with these parameters and the channel will be closed.

		func (me *Channel) QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args Table) (Queue, error) {

	*/
	ch.QueueDeclare("test-1", false, false, false, false, nil)

	/*
		QueueBind binds an exchange to a queue so that publishings to the exchange will
		be routed to the queue when the publishing routing key matches the binding
		routing key.

		  QueueBind("pagers", "alert", "log", false, nil)
		  QueueBind("emails", "info", "log", false, nil)

		  Delivery       Exchange  Key       Queue
		  -----------------------------------------------
		  key: alert --> log ----> alert --> pagers
		  key: info ---> log ----> info ---> emails
		  key: debug --> log       (none)    (dropped)

		If a binding with the same key and arguments already exists between the
		exchange and queue, the attempt to rebind will be ignored and the existing
		binding will be retained.

		In the case that multiple bindings may cause the message to be routed to the
		same queue, the server will only route the publishing once.  This is possible
		with topic exchanges.

		  QueueBind("pagers", "alert", "amq.topic", false, nil)
		  QueueBind("emails", "info", "amq.topic", false, nil)
		  QueueBind("emails", "#", "amq.topic", false, nil) // match everything

		  Delivery       Exchange        Key       Queue
		  -----------------------------------------------
		  key: alert --> amq.topic ----> alert --> pagers
		  key: info ---> amq.topic ----> # ------> emails
		                           \---> info ---/
		  key: debug --> amq.topic ----> # ------> emails

		It is only possible to bind a durable queue to a durable exchange regardless of
		whether the queue or exchange is auto-deleted.  Bindings between durable queues
		and exchanges will also be restored on server restart.

		If the binding could not complete, an error will be returned and the channel
		will be closed.

		When noWait is true and the queue could not be bound, the channel will be
		closed with an error.

		func (me *Channel) QueueBind(name, key, exchange string, noWait bool, args Table) error {
	*/
	ch.QueueBind("test-1", "test-1", "logs", false, nil)

	// test-2 D,不会删除
	ch.QueueDeclare("test-2", true, false, false, false, nil)
	ch.QueueBind("test-2", "test-2", "logs", false, nil)

	// test-3 AD,机器重启删除,消费者连接一断开,队列自动删除
	ch.QueueDeclare("test-3", false, true, false, false, nil)
	ch.QueueBind("test-3", "test-3", "logs", false, nil)

	// test-4 D,AD,消费者连接一断开,队列自动删除
	ch.QueueDeclare("test-4", true, true, false, false, nil)
	ch.QueueBind("test-4", "test-4", "logs", false, nil)

	// test-5 无属性,本连接一断开就删除了,只有exclusive属性有效
	// 在其他连接消费报错：Exception (405) Reason: "RESOURCE_LOCKED - cannot obtain exclusive access to locked queue ‘test-7‘ in vhost ‘/‘"
	// 只能在当前连接进行消费
	ch.QueueDeclare("test-5", true, false, true, false, nil)
	ch.QueueBind("test-5","test-5","logs",false,nil)// test-6 D,AD,消费者连接一断开,队列自动删除ch.QueueDeclare("test-6",true,true,false,false,nil)ch.QueueBind("test-6","test-6","logs",false,nil)// test-7 AD,本连接一断开就删除了,只有autoDelete属性和exclusive属性有效// 在其他连接消费报错：Exception (405) Reason: "RESOURCE_LOCKED - cannot obtain exclusive access to locked queue ‘test-7‘ in vhost ‘/‘"// 只能在当前连接进行消费ch.QueueDeclare("test-7",true,true,true,false,nil)ch.QueueBind("test-7","test-7","logs",false,nil)// 发布/*
	Publish sends a Publishing from the client to an exchange on the server.

	When you want a single message to be delivered to a single queue, you can
	publish to the default exchange with the routingKey of the queue name.  This is
	because every declared queue gets an implicit route to the default exchange.

	Since publishings are asynchronous, any undeliverable message will get returned
	by the server.  Add a listener with Channel.NotifyReturn to handle any
	undeliverable message when calling publish with either the mandatory or
	immediate parameters as true.

	Publishings can be undeliverable when the mandatory flag is true and no queue is
	bound that matches the routing key, or when the immediate flag is true and no
	consumer on the matched queue is ready to accept the delivery.

	This can return an error when the channel, connection or socket is closed.  The
	error or lack of an error does not indicate whether the server has received this
	publishing.

	It is possible for publishing to not reach the broker if the underlying socket
	is shutdown without pending publishing packets being flushed from the kernel
	buffers.  The easy way of making it probable that all publishings reach the
	server is to always call Connection.Close before terminating your publishing
	application.  The way to ensure that all publishings reach the server is to add
	a listener to Channel.NotifyPublish and put the channel in confirm mode with
	Channel.Confirm.  Publishing delivery tags and their corresponding
	confirmations start at 1.  Exit when all publishings are confirmed.

	When Publish does not return an error and the channel is in confirm mode, the
	internal counter for DeliveryTags with the first confirmation starting at 1.

	func (me *Channel) Publish(exchange, key string, mandatory, immediate bool, msg Publishing) error {
	*/err=ch.Publish("logs","",false,false,amqp.Publishing{ContentType:"text/plain",Body:[]byte("hi"),// 下面这条没加,机器重启消息清零DeliveryMode:amqp.Persistent,})failOnError(err,"Failed to publish a message")// exclusive只能在这个连接消费/*
	Consume immediately starts delivering queued messages.

	Begin receiving on the returned chan Delivery before any other operation on the
	Connection or Channel.

	Continues deliveries to the returned chan Delivery until Channel.Cancel,
	Connection.Close, Channel.Close, or an AMQP exception occurs.  Consumers must
	range over the chan to ensure all deliveries are received.  Unreceived
	deliveries will block all methods on the same connection.

	All deliveries in AMQP must be acknowledged.  It is expected of the consumer to
	call Delivery.Ack after it has successfully processed the delivery.  If the
	consumer is cancelled or the channel or connection is closed any unacknowledged
	deliveries will be requeued at the end of the same queue.
	/*
	Publish sends a Publishing from the client to an exchange on the server.

	When you want a single message to be delivered to a single queue, you can
	publish to the default exchange with the routingKey of the queue name.  This is
	because every declared queue gets an implicit route to the default exchange.

	Since publishings are asynchronous, any undeliverable message will get returned
	by the server.  Add a listener with Channel.NotifyReturn to handle any
	undeliverable message when calling publish with either the mandatory or
	immediate parameters as true.

	Publishings can be undeliverable when the mandatory flag is true and no queue is
	bound that matches the routing key, or when the immediate flag is true and no
	consumer on the matched queue is ready to accept the delivery.

	This can return an error when the channel, connection or socket is closed.  The
	error or lack of an error does not indicate whether the server has received this
	publishing.

	It is possible for publishing to not reach the broker if the underlying socket
	is shutdown without pending publishing packets being flushed from the kernel
	buffers.  The easy way of making it probable that all publishings reach the
	server is to always call Connection.Close before terminating your publishing
	application.  The way to ensure that all publishings reach the server is to add
	a listener to Channel.NotifyPublish and put the channel in confirm mode with
	Channel.Confirm.  Publishing delivery tags and their corresponding
	confirmations start at 1.  Exit when all publishings are confirmed.

	When Publish does not return an error and the channel is in confirm mode, the
	internal counter for DeliveryTags with the first confirmation starting at 1.

	func (me *Channel) Publish(exchange, key string, mandatory, immediate bool, msg Publishing) error {
	*/Theconsumerisidentifiedbyastringthatisuniqueandscopedforallconsumersonthischannel.Ifyouwishtoeventuallycanceltheconsumer,usethesamenon-emptyidenfitierinChannel.Cancel.Anemptystringwillcausethelibrarytogenerateauniqueidentity.TheconsumeridentitywillbeincludedineveryDeliveryintheConsumerTagfieldWhenautoAck(alsoknownasnoAck)istrue,theserverwillacknowledgedeliveriestothisconsumerpriortowritingthedeliverytothenetwork.WhenautoAckistrue,theconsumershouldnotcallDelivery.Ack.Automaticallyacknowledgingdeliveriesmeansthatsomedeliveriesmaygetlostiftheconsumerisunabletoprocessthemaftertheserverdeliversthem.Whenexclusiveistrue,theserverwillensurethatthisisthesoleconsumerfromthisqueue.Whenexclusiveisfalse,theserverwillfairlydistributedeliveriesacrossmultipleconsumers.WhennoLocalistrue,theserverwillnotdeliverpublishingsentfromthesameconnectiontothisconsumer.It‘sadvisabletouseseparateconnectionsforChannel.PublishandChannel.ConsumesonottohaveTCPpushbackonpublishingaffecttheabilitytoconsumemessages,sothisparameterisheremostlyforcompleteness.WhennoWaitistrue,donotwaitfortheservertoconfirmtherequestandimmediatelybegindeliveries.Ifitisnotpossibletoconsume,achannelexceptionwillberaisedandthechannelwillbeclosed.Optionalargumentscanbeprovidedthathavespecificsemanticsforthequeueorserver.Whenthechannelorconnectioncloses,alldeliverychanswillalsoclose.Deliveriesonthereturnedchanwillbebufferedindefinitely.Tolimitmemoryofthisbuffer,usetheChannel.Qosmethodtolimittheamountofunacknowledged/buffereddeliveriestheserverwilldeliveronthisChannel.func(me*Channel)Consume(queue,consumerstring,autoAck,exclusive,noLocal,noWaitbool,argsTable)(<-chanDelivery,error){*/c,e:=ch.Consume("test-5","test-5-c",false,false,false,false,nil)ife!=nil{fmt.Println(e.Error())}else{a:=<-cfmt.Println("test-5:"+string(a.Body))//a.Ack(false)}time.Sleep(1000*time.Second)//ch.Consume("test-1", "test-1-c", false, false, false, false, nil)}

package main

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

// Queue属性测试
func main() {
	conn, _ := amqp.Dial("amqp://guest:[email protected]:5672/")
	defer conn.Close()

	ch, _ := conn.Channel()
	defer ch.Close()

	ch.QueueDelete("test-1", false, false, false)
	ch.QueueDelete("test-2", false, false, false)
	ch.QueueDelete("test-3", false, false, false)
	ch.QueueDelete("test-4", false, false, false)
	ch.QueueDelete("test-5", false, false, false)
	ch.QueueDelete("test-6", false, false, false)
	ch.QueueDelete("test-7", false, false, false)
}

package main

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

// Queue属性测试
func main() {
	conn, _ := amqp.Dial("amqp://guest:[email protected]:5672/")
	defer conn.Close()

	ch, _ := conn.Channel()
	defer ch.Close()

	c, e := ch.Consume("test-3", "test-2-c", false, false, false, false, nil)
	if e != nil {
		fmt.Println(e.Error())
	} else {
		a := <-c
		fmt.Println("test-1:" + string(a.Body))
		//a.Ack(false)
	}
	time.Sleep(time.Second * 1000)
}

时间： 2024-10-29 06:21:34

RabbitMQ安装|使用|概念|Golang开发

一.介绍

二.安装

三.使用

四.概念入门

1.ConnectionFactory、Connection、Channel

2.Queue

3.消息的一些机制

3.1.消息确认Message acknowledgment

3.2.消息持久Message durability

3.3.提前取机制Prefetch count

4.Exchange

5.Routing key

6.Binding

6.1.Binding key

7.Exchange Types

7.1.fanout

7.2.direct

7.3.topic

7.4.headers

8.RPC

五.Go 接口

六.实例解释

工作队列：默认点对点模式

发布-订阅：广播模式

高级路由发布-订阅新版：默认点对点模式

话题发布-订阅新新版:话题模式

rpc：RPC模式

七.属性详解，测试！

RabbitMQ安装|使用|概念|Golang开发的相关文章

转：windows 下的 eclipse安装goclipse 配置golang开发环境

windows 下的 eclipse安装goclipse 配置golang开发环境

Spring Boot（十三）RabbitMQ安装与集成

RabbitMQ安装配置使用指南

RabbitMQ安装和配置

RabbitMQ 安装和配置

：Windows下RabbitMQ安装及入门

RabbitMQ安装教程

RabbitMQ系列(一)：Windows下RabbitMQ安装及入门

RabbitMQ安装|使用|概念|Golang开发

一.介绍

二.安装

三.使用

四.概念入门

1.ConnectionFactory、Connection、Channel

2.Queue

3.消息的一些机制

3.1.消息确认Message acknowledgment

3.2.消息持久Message durability

3.3.提前取机制Prefetch count

4.Exchange

5.Routing key

6.Binding

6.1.Binding key

7.Exchange Types

7.1.fanout

7.2.direct

7.3.topic

7.4.headers

8.RPC

五.Go 接口

六.实例解释

工作队列：默认点对点模式

发布-订阅：广播模式

高级路由 发布-订阅 新版：默认点对点模式

话题 发布-订阅 新新版:话题模式

rpc：RPC模式

七.属性详解，测试！

RabbitMQ安装|使用|概念|Golang开发的相关文章

高级路由发布-订阅新版：默认点对点模式

话题发布-订阅新新版:话题模式