开发笔记:消息队列rabbitmq的五种工作模式（go语言版本）

package main
import (
"RabbitMQ"
"time"
)
func main(){
ch := rabbitMQ.Connect("amqp://user:password@ip/")
rabbitMQ.NewExchange("amqp://user:password@ip/","exchange","topic")
for{
time.Sleep(1)
ch.Publish("exchange","hello world","lazy.brown.fox")
}
}

package main
import (
rabbitMQ "RabbitMQ"
"log"
)
func main(){
// 1.接收者，首先自己队列
// 2.创建交换机
// 3.将自己绑定到交换机上
// 4.接收交换机上发过来的消息
//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
//1
receive_mq := rabbitMQ.New("amqp://user:password@ip:port/","hello1")
//2
//第一个参数：rabbitmq服务器的链接，第二个参数：交换机名字，第三个参数：交换机类型
rabbitMQ.NewExchange("amqp://user:password@ip:port/","exchange","topic")
//3
receive_mq.Bind("exchange","*.orange.*")
//4
for{
//接收消息时，指定
msgs := receive_mq .Consume()
go func() {
for d := range msgs {
log.Printf("recevie1 Received a message: %s", d.Body)
}
}()
}
}

package main
import (
rabbitMQ "RabbitMQ"
"log"
)
func main(){
// 1.接收者，首先自己队列
// 2.创建交换机
// 3.将自己绑定到交换机上
// 4.接收交换机上发过来的消息
//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
//1
receive_mq := rabbitMQ.New("amqp://user:password@ip:port/","hello2")
//2
//第一个参数：rabbitmq服务器的链接，第二个参数：交换机名字，第三个参数：交换机类型
rabbitMQ.NewExchange("amqp://user:password@ip:port/","exchange","topic")
//3
receive_mq.Bind("exchange","*.*.rabbit")
receive_mq.Bind("exchange","lazy.#")
//4
for{
//接收消息时，指定
msgs := receive_mq .Consume()
go func() {
for d := range msgs {
log.Printf("recevie2 Received a message: %s", d.Body)
}
}()
}
}

package rabbitMQ
import (
"encoding/json"
"github.com/streadway/amqp"
"log"
)
//声明队列类型
type RabbitMQ struct {
channel *amqp.Channel
Name string
exchange string
}
//连接服务器
func Connect(s string) * RabbitMQ{
//连接rabbitmq
conn,e := amqp.Dial(s)
failOnError(e,"连接Rabbitmq服务器失败！")
ch ,e :=conn.Channel()
failOnError(e,"无法打开频道！")
mq := new(RabbitMQ)
mq.channel =ch
return mq
}
//初始化单个消息队列
//第一个参数：rabbitmq服务器的链接，第二个参数：队列名字
func New(s string,name string) * RabbitMQ{
//连接rabbitmq
conn,e := amqp.Dial(s)
failOnError(e,"连接Rabbitmq服务器失败！")
ch ,e :=conn.Channel()
failOnError(e,"无法打开频道！")
q,e := ch.QueueDeclare(
name,//队列名
false,//是否开启持久化
true,//不使用时删除
false, //排他
false, //不等待
nil, //参数
)
failOnError(e,"初始化队列失败！")
mq := new(RabbitMQ)
mq.channel =ch
mq.Name =q.Name
return mq
}
//批量初始化消息队列
//第一个参数：rabbitmq服务器的链接，第二个参数：队列名字列表
//配置队列参数
func (q *RabbitMQ)Qos(){
e := q.channel.Qos(1,0,false)
failOnError(e,"无法设置QoS")
}
//配置交换机参数
//初始化交换机
//第一个参数：rabbitmq服务器的链接，第二个参数：交换机名字，第三个参数：交换机类型
func NewExchange(s string,name string,typename string){
//连接rabbitmq
conn,e := amqp.Dial(s)
failOnError(e,"连接Rabbitmq服务器失败！")
ch ,e :=conn.Channel()
failOnError(e,"无法打开频道！")
e = ch.ExchangeDeclare(
name, // name
typename, // type
true, // durable
false, // auto-deleted
false, // internal
false, // no-wait
nil, // arguments
)
failOnError(e,"初始化交换机失败！")
}
//删除交换机
func (q *RabbitMQ)ExchangeDelete(exchange string){
e := q.channel.ExchangeDelete(exchange,false,true)
failOnError(e,"绑定队列失败！")
}
//绑定消息队列到哪个exchange
func (q *RabbitMQ)Bind(exchange string,key string){
e := q.channel.QueueBind(
q.Name,
key,
exchange,
false,
nil,
)
failOnError(e,"绑定队列失败！")
q.exchange = exchange
}
//向消息队列发送消息
//Send方法可以往某个消息队列发送消息
func (q *RabbitMQ) Send(body interface{}){
str,e := json.Marshal(body)
failOnError(e,"消息序列化失败！")
e = q.channel.Publish(
"",//交换
q.Name,//路由键：当前队列的名字
false, //必填
false, //立即
amqp.Publishing{
ReplyTo:q.Name,
Body:[]byte(str),
})
msg := "向队列:"+q.Name+"发送消息失败！"
failOnError(e,msg)
}
//向exchange发送消息
//Publish方法可以往某个exchange发送消息
func (q *RabbitMQ) Publish(exchange string,body interface{},key string) {
str,e := json.Marshal(body)
failOnError(e,"消息序列化失败！")
e = q.channel.Publish(
exchange,
key,
false,
false,
amqp.Publishing{ReplyTo:q.Name,
Body:[]byte(str)},
)
failOnError(e,"向路由发送消息失败！")
}
//接收某个消息队列的消息
func (q * RabbitMQ) Consume() <-chan amqp.Delivery{
c,e :=q.channel.Consume(
q.Name,//指定从哪个队列中接收消息
"",
true,
false,
false,
false,
nil,
)
failOnError(e,"接收消息失败！")
return c
}
//关闭队列连接
func (q *RabbitMQ) Close() {
q.channel.Close()
}
//错误处理函数
func failOnError(err error, msg string) {
if err != nil {
log.Fatalf("%s: %s", msg, err)
}
}