写在前面
在我们的生产环境中搭了两台rabbitmq， 前面架设了一台HAProxy做负载均衡，当我们的客户端连接到HAProxy，然后由HAProxy负责将链接分配给其中一台rabbitmq，客户端需要需要负责断线重连，需要将获取的数据，分配消息给相应的处理方法，然后还需要回复给rabbitmq ACK，这其中客户端需要负责断线重连的逻辑是很重要的，因为有可能客户端和HAProxy的连接是正常的，但是HAProxy和rabbitmq的链接因为网络波动断开了，那么这个时候客户端其实是没有工作的，并且会在rabbitmq中不断积累消息。
下面的内容给出了一个比较完善的处理逻辑，以供参考。
实战
定义接口
从之前的说明来看，这是一个典型的观察者模式，由RabbitMQ对象负责维护连接，获取消息，然后定义若干个接收者注册到RabbitMQ对象中，这时候RabbitMQ对象一旦收到了由RabbitMQ发来的数据，就可以将该消息分发到相应的接收者去处理，当接收者处理完成后告诉RabbitMQ对象消息消费成功，然后由RabbitMQ对象回复RabbitMQ ACK，当然可以在其中加上重试机制，接收者有可能因为某种情况处理失败，那么每隔一定的时间RabbitMQ对象需要重新调用一次接收者重新处理，直至成功，然后再返回ACK。
先来看看基本的接口约定
// Receiver 观察者模式需要的接口
// 观察者用于接收指定的queue到来的数据
type Receiver interface {
QueueName() string // 获取接收者需要监听的队列
RouterKey() string // 这个队列绑定的路由
OnError(error) // 处理遇到的错误，当RabbitMQ对象发生了错误，他需要告诉接收者处理错误
OnReceive([]byte) bool // 处理收到的消息, 这里需要告知RabbitMQ对象消息是否处理成功
}这样就将接收者和RabbitMQ对象之间就解耦了，这样后期如果需要添加新的接收者那就很容易了。
下面来看一看RabbitMQ对象的定义:
这里用到的RabbitMQ client是RabbitMQ官方的 Github
// RabbitMQ 用于管理和维护rabbitmq的对象
type RabbitMQ struct {
wg sync.WaitGroup
channel *amqp.Channel
exchangeName string // exchange的名称
exchangeType string // exchange的类型
receivers []Receiver
}
// New 创建一个新的操作RabbitMQ的对象
func New() *RabbitMQ {
// 这里可以根据自己的需要去定义
return &RabbitMQ{
exchangeName: ExchangeName,
exchangeType: ExchangeType,
}
}
RabbitMQ对象的初始化操作
这里RabbitMQ对象需要初始化交换机，注册接收者并初始化接收者监听的Queue，以及断线重连的机制
// prepareExchange 准备rabbitmq的Exchange
func (mq *RabbitMQ) prepareExchange() error {
// 申明Exchange
err := mq.channel.ExchangeDeclare(
mq.exchangeName, // exchange
mq.exchangeType, // type
true, // durable
false, // autoDelete
false, // internal
false, // noWait
nil, // args
)
if nil != err {
return err
}
return nil
}
// run 开始获取连接并初始化相关操作
func (mq *RabbitMQ) run() {
if !config.Global.RabbitMQ.Refresh() {
log.Errorf("rabbit刷新连接失败，将要重连: %s", config.Global.RabbitMQ.URL)
return
}
// 获取新的channel对象
mq.channel = config.Global.RabbitMQ.Channel()
// 初始化Exchange
mq.prepareExchange()
for _, receiver := range mq.receivers {
mq.wg.Add(1)
go mq.listen(receiver) // 每个接收者单独启动一个goroutine用来初始化queue并接收消息
}
mq.wg.Wait()
log.Errorf("所有处理queue的任务都意外退出了")
// 理论上mq.run()在程序的执行过程中是不会结束的
// 一旦结束就说明所有的接收者都退出了，那么意味着程序与rabbitmq的连接断开
// 那么则需要重新连接，这里尝试销毁当前连接
config.Global.RabbitMQ.Distory()
}
// Start 启动Rabbitmq的客户端
func (mq *RabbitMQ) Start() {
for {
mq.run()
// 一旦连接断开，那么需要隔一段时间去重连
// 这里最好有一个时间间隔
time.Sleep(3 * time.Second)
}
}注册接收者
// RegisterReceiver 注册一个用于接收指定队列指定路由的数据接收者
func (mq *RabbitMQ) RegisterReceiver(receiver Receiver) {
mq.receivers = append(mq.receivers, receiver)
}
// Listen 监听指定路由发来的消息
// 这里需要针对每一个接收者启动一个goroutine来执行listen
// 该方法负责从每一个接收者监听的队列中获取数据，并负责重试
func (mq *RabbitMQ) listen(receiver Receiver) {
defer mq.wg.Done()
// 这里获取每个接收者需要监听的队列和路由
queueName := receiver.QueueName()
routerKey := receiver.RouterKey()
// 申明Queue
_, err := mq.channel.QueueDeclare(
queueName, // name
true, // durable
false, // delete when usused
false, // exclusive(排他性队列)
false, // no-wait
nil, // arguments
)
if nil != err {
// 当队列初始化失败的时候，需要告诉这个接收者相应的错误
receiver.OnError(fmt.Errorf("初始化队列 %s 失败: %s", queueName, err.Error()))
}
// 将Queue绑定到Exchange上去
err = mq.channel.QueueBind(
queueName, // queue name
routerKey, // routing key
mq.exchangeName, // exchange
false, // no-wait
nil,
)
if nil != err {
receiver.OnError(fmt.Errorf("绑定队列 [%s - %s] 到交换机失败: %s", queueName, routerKey, err.Error()))
}
// 获取消费通道
mq.channel.Qos(1, 0, true) // 确保rabbitmq会一个一个发消息
msgs, err := mq.channel.Consume(
queueName, // queue
"", // consumer
false, // auto-ack
false, // exclusive
false, // no-local
false, // no-wait
nil, // args
)
if nil != err {
receiver.OnError(fmt.Errorf("获取队列 %s 的消费通道失败: %s", queueName, err.Error()))
}
// 使用callback消费数据
for msg := range msgs {
// 当接收者消息处理失败的时候，
// 比如网络问题导致的数据库连接失败，redis连接失败等等这种
// 通过重试可以成功的操作，那么这个时候是需要重试的
// 直到数据处理成功后再返回，然后才会回复rabbitmq ack
for !receiver.OnReceive(msg.Body) {
log.Warnf("receiver 数据处理失败，将要重试")
time.Sleep(1 * time.Second)
}
// 确认收到本条消息, multiple必须为false
msg.Ack(false)
}
}
整合到一起
接收者的逻辑这里就不写的，只要根据实际的业务逻辑并实现了接口就可以了，这个比较容易。
获取RabbitMQ的连接
rabbitmqConn, err = amqp.Dial(url)
if err != nil {
panic("RabbitMQ 初始化失败: " + err.Error())
}
rabbitmqChannel, err = rabbitmqConn.Channel()
if err != nil {
panic("打开Channel失败: " + err.Error())
}// 启动并开始处理数据
func main() {
// 假设这里有一个AReceiver和BReceiver
aReceiver := NewAReceiver()
bReceiver := NewBReceiver()
mq := rabbitmq.New()
// 将这个接收者注册到
mq.RegisterReceiver(aReceiver)
mq.RegisterReceiver(bReceiver)
mq.Start()
}
应用场景
举一个我们自己用于生产环境的例子：
我们主要是用于接收Mysql的变更，并增量更新Elasticsearch的索引，负责数据库变更监听的服务用的是Canel，它伪装成一个mysql slave，用于接收mysql binlog的变更通知，然后将变更的数据格式化后写入RabbitMQ，然后由go实现的消费者去订阅数据库的变更通知。
由于客户端并不关心表中哪些字段发生了变化，只需要知道数据库指定的表有变更，那么就将此次变更写入Elasticsearch，这个逻辑对于每一张监听的表都是一样的，那么这样我们就可以将需要监听表变更的操作完全配置化，我只要再配置文件中指定一个接收者并指定待消费的队列，然后就可以由程序自动生成若干的接收者并且依次注册进RabbitMQ对象中，这样我们只需要针对一些特殊的操作写相应地代码即可，这样大大简化了我们地工作量，来看一看配置文件：
[[autoReceivers]]
receiverName = "article_receiver"
database = "blog"
tableName = "articles"
primaryKey = "articleId"
queueName = "articles_queue"
routerKey = "blog.articles.*"
esIndex = "articles_idx"
[[autoReceivers]]
receiverName = "comment_receiver"
database = "blog"
tableName = "comments"
primaryKey = "commentId"
queueName = "comments_queue"
routerKey = "blog.comments.*"
esIndex = "comments_idx"这个时候就需要调整一下接收者地注册函数了：
// WalkReceivers 使用callback遍历处理所有的接收者
// 这里地callback就是上面提到地 mq.RegisterReceiver
func WalkReceivers(callback func(rabbitmq.Receiver)) {
successCount := 0
// 遍历每一个配置项，依次生成需要自动创建接收者
// 这里的congfig是统一获取配置地对象，大家根据实际情况处理就可以了
for _, receiverCfg := range config.Global.AutoReceivers {
// 验证每一个接收者的合法性
err := receiverCfg.Validate()
if err != nil {
log.Criticalf("生成 %s 失败: %s, 使用该配置: %+v", receiverCfg.ReceiverName, err.Error(), receiverCfg)
continue
}
// 将接收者注册到监听rabbitmq的对象中
callback(NewAutoReceiver(receiverCfg))
log.Infof("生成 %s 成功使用该配置: %+v", receiverCfg.ReceiverName, receiverCfg)
successCount++
}
if successCount != len(config.Global.AutoReceivers) || successCount == 0 {
panic("无法启动所有的接收者，请检查配置")
}
// 如有必要，这里可以继续添加需要手工创建的接收者
}启动地流程也需要进行微调一下：
func registeAndStart() {
mq := rabbitmq.New()
// 遍历所有的receiver，将他们注册到rabbitmq中去
WalkReceivers(mq.RegisterReceiver)
log.Info("初始化所有的Receiver成功")
mq.Start()
}
这样就定义好了两个receiver，启动程序后，就可以接收到数据库地变更并更新elasticsearch中地索引了，非常地方便。
写在最后
这个是对平时工作地一点总结，希望可以给大家带来帮助，如果文中有纰漏之处，还望指正，这里完整地代码就不贴了，文章里已经搭起了一个完整地框架了，剩下地就是业务逻辑了，如果有必要地化，我会整理成一个完整地项目放到github上。
