1、路由模式
路由模式跟订阅模式类似,只不过在订阅模式的基础上加上了类型,订阅模式是分发到所有绑定到交换机的队列,路由模式只分发到绑定在交换机上面指定路由键的队列,即一个消息可被多个消费者获取,并且消息的目标队列可以被生产者绑定。生产者申明一个direct类型交换机,然后发送消息到这个交换机指定路由键。消费者指定消费这个交换机的这个路由键,即可接收到消息,其他消费者收不到。
1、路由模式创建实例
func NewRabbitMQRouting(exchangeName string,routingKey string) *RabbitMQ {
//创建RabbitMQ实例
rabbitmq := NewRabbitMQ("",exchangeName,routingKey)
var err error
//获取connection
rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
rabbitmq.failOnErr(err,"failed to connect rabbitmq!")
//获取channel
rabbitmq.channel, err = rabbitmq.conn.Channel()
rabbitmq.failOnErr(err, "failed to open a channel")
return rabbitmq
}
2、路由模式发送消息
func (r *RabbitMQ) PublishRouting(message string ) {
//1.尝试创建交换机
err := r.channel.ExchangeDeclare(
r.Exchange,
//要改成direct
"direct",
true,
false,
false,
false,
nil,
)
r.failOnErr(err, "Failed to declare an excha"+
"nge")
//2.发送消息
err = r.channel.Publish(
r.Exchange,
//要设置
r.Key,
false,
false,
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(message),
})
}
3、路由模式接收消息
func (r *RabbitMQ) RecieveRouting() {
//1.试探性创建交换机
err := r.channel.ExchangeDeclare(
r.Exchange,
//交换机类型
"direct",
true,
false,
false,
false,
nil,
)
r.failOnErr(err, "Failed to declare an exch"+
"ange")
//2.试探性创建队列,这里注意队列名称不要写
q, err := r.channel.QueueDeclare(
"", //随机生产队列名称
false,
false,
true,
false,
nil,
)
r.failOnErr(err, "Failed to declare a queue")
//绑定队列到 exchange 中
err = r.channel.QueueBind(
q.Name,
//需要绑定key
r.Key,
r.Exchange,
false,
nil)
//消费消息
messges, err := r.channel.Consume(
q.Name,
"",
true,
false,
false,
false,
nil,
)
forever := make(chan bool)
go func() {
for d := range messges {
log.Printf("Received a message: %s", d.Body)
}
}()
fmt.Println("[#]Waiting for messages, to exit press CTRL + C")
<-forever
}
最后提一下主题模式。
其实就是使路由能够模糊匹配
//话题模式
//创建RabbitMQ实例
func NewRabbitMQTopic(exchangeName string,routingKey string) *RabbitMQ {
//创建RabbitMQ实例
rabbitmq := NewRabbitMQ("",exchangeName,routingKey)
var err error
//获取connection
rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
rabbitmq.failOnErr(err,"failed to connect rabbitmq!")
//获取channel
rabbitmq.channel, err = rabbitmq.conn.Channel()
rabbitmq.failOnErr(err, "failed to open a channel")
return rabbitmq
}
//话题模式发送消息
func (r *RabbitMQ) PublishTopic(message string ) {
//1.尝试创建交换机
err := r.channel.ExchangeDeclare(
r.Exchange,
//要改成topic
"topic",
true,
false,
false,
false,
nil,
)
r.failOnErr(err, "Failed to declare an excha"+
"nge")
//2.发送消息
err = r.channel.Publish(
r.Exchange,
//要设置
r.Key,
false,
false,
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(message),
})
}
//话题模式接受消息
//要注意key,规则
//其中“*”用于匹配一个单词,“#”用于匹配多个单词(可以是零个)
//匹配 imooc.* 表示匹配 imooc.hello, 但是imooc.hello.one需要用imooc.#才能匹配到
func (r *RabbitMQ) RecieveTopic() {
//1.试探性创建交换机
err := r.channel.ExchangeDeclare(
r.Exchange,
//交换机类型
"topic",
true,
false,
false,
false,
nil,
)
r.failOnErr(err, "Failed to declare an exch"+
"ange")
//2.试探性创建队列,这里注意队列名称不要写
q, err := r.channel.QueueDeclare(
"", //随机生产队列名称
false,
false,
true,
false,
nil,
)
r.failOnErr(err, "Failed to declare a queue")
//绑定队列到 exchange 中
err = r.channel.QueueBind(
q.Name,
//在pub/sub模式下,这里的key要为空
r.Key,
r.Exchange,
false,
nil)
//消费消息
messges, err := r.channel.Consume(
q.Name,
"",
true,
false,
false,
false,
nil,
)
forever := make(chan bool)
go func() {
for d := range messges {
log.Printf("Received a message: %s", d.Body)
}
}()
fmt.Println("[#]Waiting for messages, to exit press CTRL + C")
<-forever
}