Go语言版本rabbitmq消息队列库：simple、worker、Fanout 模型、Direct 模型、Topic模型

package rabbitmq import ( "encoding/json" "github.com/streadway/amqp" "log" ) type RabbitMQ struct { conn *amqp.Connection //连接 channel *amqp.Channel //管道 QueueName string //队列名称 Exchange string //交换机 Key string //key Simple模式 几乎用不到 MqUrl string //连接信息 } // NewRabbitMQ 创建RabbitMQ结构体实例 /* new只需要最基础的连接上rabbitmq服务器即可，也就是创建好客户端 队列和exchange并非是客户行为，应该为管理员配置好，当然我们自己也可以通过代码设置，但是没必要放在new中 */ func NewRabbitMQ(url string) *RabbitMQ { rabbitmq := &RabbitMQ{MqUrl: url} var err error //创建rabbitmq连接 rabbitmq.conn, err = amqp.Dial(rabbitmq.MqUrl) rabbitmq.failOnErr(err, "创建连接错误！") rabbitmq.channel, err = rabbitmq.conn.Channel() rabbitmq.failOnErr(err, "获取channel失败") return rabbitmq } // ExchangeDeclare 创建exchange 这个和创建的rabbitmq没有关系，别的程序创建的exchange，该程序也能使用 /* 创建exchange 根据rabbitmq的原理可以知道，exchange和客户端没多大关系，哪一个客户端都可以绑定exchange 形参:exchange表示声明的ex的名字，以后绑定ex的时候需要用到 kind表示声明的ex的类型 */ func (r *RabbitMQ) ExchangeDeclare(exchange, kind string) { r.Exchange = exchange err := r.channel.ExchangeDeclare( r.Exchange, kind, false, false, false, false, nil) r.failOnErr(err, "创建exchange失败") } // QueueDeclare 创建Queue /* 创建Queue 同ex的道理一样，创建的queue，不一定是非得这个客户端自己使用， 所以，在接收的时候，我们需要选择需要接收的队列 形参:queueName表示声明队列的名称，以后读取数据的时候，需要依靠队列名字来绑定 */ func (r *RabbitMQ) QueueDeclare(queueName string) { r.QueueName = queueName _, err := r.channel.QueueDeclare( queueName, false, false, false, false, nil) r.failOnErr(err, "创建queue失败") } // Bind /* 绑定:在接收消息之前，先要绑定队列以及ex r.QueueName, //绑定要接收的队列 r.Key, //给队列赋bingKey ex, //队列要从哪个ex接收数据 */ func (r *RabbitMQ) Bind(queue, key, ex string) { err := r.channel.QueueBind( queue, key, ex, false, nil) r.failOnErr(err, "绑定exchange失败") } // Send /* 发送，exchange是发送到哪一个ex上，key是routing key 如果在simple或者广播等模式下，exchange,key传入空即可 */ func (r *RabbitMQ) Send(exchange, key string, body interface{}) { str, err := json.Marshal(body) r.failOnErr(err, "send json 失败") err = r.channel.Publish( exchange, key, false, false, amqp.Publishing{ ReplyTo: r.QueueName, Body: []byte(str), }, ) r.failOnErr(err, "send 失败") } // Consume /* 从哪一个队列中读取数据 */ func (r *RabbitMQ) Consume(queue string) <-chan amqp.Delivery { c, err := r.channel.Consume( queue, "", true, false, false, false, nil) r.failOnErr(err, "consume 失败") return c } func (r *RabbitMQ) Destroy() { r.channel.Close() r.conn.Close() } func (r *RabbitMQ) failOnErr(err error, message string) { if err != nil { log.Fatalf("%s:%s", message, err) } }