You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
rabbitmq/rabbitmq.go

90 lines
2.0 KiB

4 years ago
package rabbitmq
import (
"fmt"
"github.com/streadway/amqp"
)
var Rabbitmq = &rabbitmq{}
type rabbitmq struct {
2 years ago
Conn *amqp.Connection
4 years ago
}
// InitRabbitmq @Title 初始化rabbitmq
2 years ago
func InitRabbitmq(host string, port int, user, password string) (err error) {
Rabbitmq.Conn, err = amqp.Dial(fmt.Sprintf("amqp://%s:%s@%s:%d/", user, password, host, port))
4 years ago
return
}
// Publish @Title 发送消息
// @Param exchangeName string true "指定发送的交换机"
// @Param msg string true "发送消息内容"
// @Param delay int64 false "延时发送秒数"
func (r *rabbitmq) Publish(exchangeName, msg string, delay ...int64) error {
2 years ago
ch, err := r.Conn.Channel()
4 years ago
if err != nil {
return err
}
3 years ago
defer ch.Close()
4 years ago
header := amqp.Table{}
if len(delay) > 0 && delay[0] > 0 {
header["x-delay"] = delay[0] * 1000
}
return ch.Publish(
exchangeName, //交换
"", //队列名称
false, //强制为
false, //立即
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(msg),
Headers: header,
})
}
// Consume @Title 消费队列
// @Param queueName string true "队列名称"
// @Param autoAck bool true "是否自动确认"
// @Param prefetchCount int 0 "预消费数量"
4 years ago
// @Param callback func(amqp.Delivery)) true "回调处理"
func (r *rabbitmq) Consume(queueName string, autoAck bool, prefetchCount int, callback func(amqp.Delivery)) error {
2 years ago
ch, err := r.Conn.Channel()
4 years ago
if err != nil {
return err
}
2 years ago
defer ch.Close()
if prefetchCount > 0 {
if err := ch.Qos(prefetchCount, 0, false); err != nil {
return err
}
}
4 years ago
//消费队列
msgs, err := ch.Consume(
queueName, //队列名称
"", //消费者
autoAck, //自动确认
false, //排他
false,
false,
nil,
)
if err != nil {
return err
}
forever := make(chan bool)
go func() {
for d := range msgs {
callback(d)
}
}()
<-forever
return nil
}
// Close @Title 关闭资源
func (r *rabbitmq) Close() {
2 years ago
defer Rabbitmq.Conn.Close()
4 years ago
}