简介
rabbitmq是一个消息代理系统，为应用提供一个通用得消息发布，接受平台，为应用提供非阻塞的消息系统，方便进行异步处理。
优点
- 消息的可靠性。持久化消息，消息接受确认，消息重传等可靠机制。
- 灵活的路由。交换机可以根据广播，或者根据路由键或匹配符匹配到不同的队列。
- 高可用的集群。
应用场景
1.异步处理
减少用户对不必要的耗时操作的等待，处理结果以异步方式（邮件，消息推送）进行提醒。
2.应用解耦
当某个应用发展到一定规模的时候，需要把里面的模块分别拆出来进行解耦，而模块之间的通讯方式是多样的，常见的有rpc，消息队列，http请求。其中消息队列在内部模块通信是更为稳定。
3.流量削峰
如果突发遇到大量的数据请求的时候，服务器如果不做队列处理，一下子处理全部的请求，会很容易造成宕机，如果把请求的数据都放入队列里，之后再逐个逐个地进行处理，可以平缓地渡过流量高峰期。
工作方式
rabbitmq的工作方式如下，生产者（publisher）发送消息到交换机，交换机（exchange）根据自己的类型以及消息的路由键，路由到对应的队列里，队列分发消息到消费者（consumer）
初尝rabbitmq
现在我们假设有这个场景，客服A需要发送客户的下单信息给库存人员B，客服A有一个订单信息发送器，库存人员B拥有消息接收器。
首先库存人员B建立连接并接受消息，伪代码：
// 建立连接
conn, _ := amqp.Dial("amqp://localhost")
ch, _ := conn.Channel()
// 声明队列，不存在则创建，存在则不会进行任何操作
queue, _ := ch.QueueDeclare("order")
// 从队列里面获取消息
deliver, _ := ch.Consume(q.Name)
for d:= range deliver {
// 输出消息主体
log.Printf("B Received a message: %s", d.Body)
// 返回获取成功标识给队列
d.Ack(true)
}
然后客服A也建立连接并发送消息，伪代码：
// 建立连接
conn, _ := amqp.Dial("amqp://localhost")
ch, _ := conn.Chanenel()
// 声明队列，不存在则创建，存在则不会进行任何操作
queue, _ := ch.QueueDeclare("order")
// 发布消息
ch.Publish(
q.Name, // 队列名字
amqp.Publishing{
Body:[]byte("new order" + product.String()),
})
上面就是一种简单的直接通过队列进行连接的方法，可能会有人看出来，为什么没有交换机的参与，其实上面的操作其实是通过默认交换机进行消息传递，可以不指定交换机名字直接指定队列名字进行交互。
模块介绍
通过上面的简单例子，我们可以更进一步地了解到rabbitmq的工作方式，下面我会更详细地讲解各个模块。
消息
消息是通信内容的主体，消息对象有点像http的request，除了可以携带消息内容，还可以带有各种属性，如：
- ContentType（内容类型）
- ContentEncoding（内容编码）
- RemoteKey（路由键）
- DeliveryMode（ 投递模式，消息是否持久化）
- 等等...
有些属性只是约定规范，如ContentType，ContentEncoding，需要程序自己做处理，有些属性rabbitmq会根据值来进行处理，如RemoteKey，交换机会根据消息的RemoteKey和自身的类型来决定投递到哪些队列，DeliveryMode可以决定是否持久化消息。
#消息投递
ch.Publish(
"", # exchange名字，空为默认交换机
key, # routingkey 路由键
false,
false,
# 消息
amqp.Publishing{
DeliveryMode:amqp.Persistent,
ContentType:"text/plain",
Body:[]byte("hello world"),
})
队列
队列是存储消息的主体，队列本身所拥有的一些属性：
- Name 队列名字，不同的队列名字应该保持唯一性
- Durable rabbitmq重启后，队列是否依旧存在，需要注意消息要持久化的要另外设置消息
- exclusive 当前队列只能被一个消费者连接使用，关闭连接后删除队列。
- auto-delete 最后一个消费者退订后删除队列。
在代码里面队列声明，如果队列不存在则新建队列，如果已存在相同名字的队列且属性不同的话则会报错。可以选择让系统自动生成队列，然后返回队列名字。
# 队列声明，参数依次为name，Durable，auto-delete，exclusive，no-wait.args
amqp.QueueDeclare("queuename", true, false, flase, false, nil)
消费者（consumer）
消费者用以消费队列里的消息的自定义程序片段，消费者获取队列里的消息有两种方式，一种是拉取（pull）的方式，通过channel.basicget方法，一种是订阅方式，队列推送消息到rabbitmq，这种方式用的最多。
消息处理
消息处理，消费者端连接队列后，可以得到一个类似于句柄的东西，如果没有消息就会一直阻塞。
消费者在收到消息之后处理的情况可能是成功的，也有可能是失败的，为了确保消息已经成功处理然后队列删除消息，如果失败则进行其他机制，以免消息一直重复在队列里面，或消息因消费者宕机而丢失。
消息确认（ack）
如果消息成功地被消费者处理的话，需要有一个消息确认的机制。
rabbitmq提供两种确认机制：
- 自动确认模式，队列将消息发送给消费者之后立即删除消息（basic.deliver或basic.get-ok）
- 显式确认模式，待消费者发送接受成功之后删除（basic.ack）
一般而言我们用的更多的是显式确认模式，如果消费者接收到消息没有进行确认之后就宕机了，队列里面的该消息还是会存在的，然后会把消息转发到其他消费者。
消息拒绝（basic.reject）
如果消费者对消息的处理出现了一些问题，可以调用rabbitmq的basic.reject来拒绝消息，拒绝消息之后，可以做的是把消息放回到队列里面，或者直接删除消息。
其实如果出现问题的消息，即便是交给其他的消费者，很会很大概率继续出现问题，这时候我们可以把消息放到其他专门处理记录问题的队列里面，交由另外的消费者处理。
交换机
交换机更像是消息的路由层，队列绑定到交换机，然后发布者可以发送的消息都是经过交换机的，然后经由消息的remote key（路由键）路由到交换机所绑定的队列里。
交换机分为4种类型：
1.直连交换机（direct）
直连型交换机（direct exchange）是根据消息携带的路由键（routing key）将消息投递给对应队列的。
- 将一个队列绑定到某个交换机上，同时赋予该绑定一个路由键（routing key）
- 当一个携带着路由键为R的消息被发送给直连交换机时，交换机会把它路由给绑定值同样为R的队列。
其实初尝rabbitmq的例子里面，看上去没有绑定交换机，实际上也是绑定了直连交换机，只是是一个特殊的预先声明好的，名字为空字符串的交换机，叫默认交换机，每个队列都会自动绑定到默认交换机上。
2.扇形交换机（funout）
扇型交换机（funout exchange）将消息路由给绑定到它身上的所有队列，而不理会绑定的路由键。如果N个队列绑定到某个扇型交换机上，当有消息发送给此扇型交换机时，交换机会将消息的拷贝分别发送给这所有的N个队列。扇型用来交换机处理消息的广播路由（broadcast routing）。
3.主题交换机（topic）
主题交换机（topic exchanges）通过对消息的路由键和队列到交换机的绑定模式之间的匹配，将消息路由给一个或多个队列。主题交换机经常用来实现各种分发/订阅模式及其变种。主题交换机通常用来实现消息的多播路由（multicast routing）。
主题交换机在我看来就像添加了简单的通配符+字符串来达到一个路由的规则。
4.头交换机（headers）
头交换机用的不是很多，有时消息的路由操作会涉及到多个属性，此时使用消息头就比用路由键更容易表达，头交换机（headers exchange）就是为此而生的。头交换机使用多个消息属性来代替路由键建立路由规则。通过判断消息头的值能否与指定的绑定相匹配来确立路由规则。
扇形路由器实现广播
生产者代码
conn, _ := amqp.Dial("amqp://localhost")
ch, _ := conn.Channel()
ch.ExchangeDeclare(
"hello",
"fanout",
true,
false,
false,
false,
nil,
)
ch.Publish(
"hello",
"", // 由于是广播，所以可以不填写路由键
false,
false,
amqp.Publishing{
DeliveryMode:amqp.Persistent,
Body:[]byte("hello"+time.Now().String()),
})
消费者代码
conn, _ := amqp.Dial("amqp://localhost")
ch, _ := conn.Channel()
ch.ExchangeDeclare(
"hello", // 交换机名字
"fanout", // 交换机类型
true, // durable
false, // autoDelete
false, // internal
false, // noWait
nil, // args
)
q, _ := ch.QueueDeclare(
"",
false, // durable
false, // autoDelete
true, // exclusive
false, // noWait
nil, //
)
ch.QueueBind(
q.Name, // queuename
"", // remote key，由于是广播，可以不填写路由键
"hello", // exchange name
false, // nowait
nil,
)
msgs, _ := ch.Consume(q.Name,"", true, false, false,false,nil)
for msg := range msgs {
log.Printf("%s", msg.Body)
}
主题交换机实现路由匹配
设有如下场景：设计一个日志收集系统，日志有不同的级别，debug，info，warn，error，日志格式为：
级别.模块名字 如：info.login
有不同的队列负责收集不同级别的日志，其中有个队列专门收集收集warn和error的数据，设计如下：
生产者
func main() {
conn, _ := amqp.Dial("amqp://localhost")
ch, _ := conn.Channel()
ch.ExchangeDeclare(
"logs",
"topic",
true,
false,
false,
false,
nil,
)
ch.Publish(
"logs",
"debug.123",
false,
false,
amqp.Publishing{
DeliveryMode:amqp.Persistent,
Body:[]byte("hello"),
},
)
}
消费者
func main() {
conn, _ := amqp.Dial("amqp://localhost")
ch, _ := conn.Channel()
ch.ExchangeDeclare(
"logs",
"topic",
true,
false,
false,
false,
nil,
)
q, _ := ch.QueueDeclare(
"log1",
true,
false,
false,
false,
nil,
)
// 队列绑定的remote key
keys := []string{"error.*", "warn.*"}
for _, key := range keys{
ch.QueueBind(
q.Name,
key,
"logs",
false,
nil,
)
}
deliver, _ := ch.Consume(
q.Name,
"",
false,
false,
false,
false,
nil,
)
for d:= range deliver {
fmt.Println(string(d.Body))
d.Ack(true)
}
}