阅读本文需要掌握的知识：
1、rabbit mq
2、golang
如果项目使用消息队列进行异步处理，大家应该都会碰上一个问题：如果某个消息正在处理中，这时候中断程序（或者更新），那么将会出现数据异常。这种情况可以通过两种方式解决该问题：
1、使用自带的ack机制，将在消息处理完之后，发送ack消息给mq，这样如果处理到一半，程序中断，mq会将该消息发送给其他消费者进行处理。但是该解决方案会产生另一个问题，假如消息处理到一半，一部分数据已经保存到数据库了，那么重新处理可能导致数据异常。这时候需要配合第二种方式进行处理。
2、让程序优雅的中断，也就是graceful shutdown。在接收到中断信号（类似control + c）的时候，不直接退出，而是等所有的子进程处理完闭后再退出。
下面用golang和rabbitmq实现了一个demo，具体代码查看https://github.com/bin-x/rabbitmq-graceful-demo。
/lib
该目录放了rabbit mq和server的封装，其中rabbit mq很重要的一点是在消费者代码中将auto-ack设为false，这样就可以在我们处理完消息后手动发送ack消息给rabbitmq的服务器，lib/rabbit.go：
consume, err := ch.Consume(
queueName, // queue
consumerName, // consumer
false, // auto-ack， 这里必须设为false
false, // exclusive
false, // no-local
true, // no-wait
nil, // args
)
其他关于rabbitmq需要的配置不在本文讨论范围。这里不展开介绍了。
lib/server.go 中相对重要的方法介绍:
// 开启graceful shutdown。默认不开启
func (srv *server) SetGraceful(duration time.Duration) {
srv.graceful = true
srv.gracefulTimeout = duration
}
// server的启动方法，Delivery chan作为参数，可以通过channel不断获取新消息。
func (srv *server) Run(deliveries <-chan amqp.Delivery) {
go func() {
// 处理消息
for d := range deliveries {
// 开启一个独立协程处理消息，如果不想并行处理多条消息，可以将go关键字去掉
go srv.startHandler(d)
// 关闭后不再处理新消息，且通知消息队列重新投递当前接收到的消息
if srv.close {
d.Reject(true)
break
}
}
}()
// graceful shutdown需要手动开启，否则只是简单的阻塞。
if srv.graceful {
srv.gracefulShutdown()
} else {
forever := make(chan bool)
<-forever
}
}
// 使用*sync.WaitGroup 对handler进行计数，保证所有handler都done之后关闭。
func (srv *server) startHandler(delivery amqp.Delivery) {
srv.wg.Add(1)
defer srv.wg.Done()
srv.handler(delivery)
}
// 核心逻辑
func (srv *server) gracefulShutdown() {
// 阻塞，直到接收到shutdown的信号
ch := make(chan os.Signal)
signal.Notify(ch, syscall.SIGTERM, syscall.SIGQUIT, syscall.SIGINT, syscall.SIGKILL)
_ = <-ch
log.Println("receive a shutdown signal")
//关闭后，Run方法中处理消息的协程将会关闭，不再处理新消息。
srv.close = true
//最大关闭时间，超过设定时间后，将会强制关闭。
cxt, cancel := context.WithTimeout(context.Background(), srv.gracefulTimeout)
defer cancel()
// graceful关闭
if err := srv.shutdown(cxt); err != nil {
log.Fatalln(err)
}
log.Println("close the consumer.")
}
// 阻塞，直到触发以下两种情况中的一种：
// 1、所有handler执行完毕，
// 2、到达设置的最大关闭时间，然后抛出一个错误。
func (srv *server) shutdown(ctx context.Context) error {
go srv.waitGroup()
select {
case <-srv.wait:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
// 监听handler是否完成，如果完成，通知系统可以关闭。
func (srv *server) waitGroup() {
srv.wg.Wait()
srv.wait <- true
}
接着我们来看看最重要的consumer.go:
// 创建一个新的server
server := lib.NewServer(work)
// 启用graceful，并设置最大20秒的超时时间，超过该时间强制关闭，防止代码死循环。
server.SetGraceful(time.Second * 20)
// 启动服务
server.Run(consume)
// 消息处理方法
func work(delivery amqp.Delivery) {
// 模拟长时间执行的任务
for i := 0; i < 10; i++ {
log.Println("i:", i)
time.Sleep(time.Second)
}
log.Println("mq's data:", string(delivery.Body[:]))
// 执行完毕后，需要通知mq。如果没有这条，那这条消息永远不会被确认，在中断rabbtmq的channel或者connection的时候，这条消息将会被重新投递。
delivery.Ack(false)
}
测试：
将consumer.go和producer.go中的rabbitmq信息改为你自己的：
err := mq.Connect("amqp://user:password@ip:port/yourhost")
go get 获取对应的包。
首先启动消费者：
$ go run consumer.go
2020/09/26 22:04:02 start consumer...
接着在一个新的终端页面，通过生产者发送条消息到队列中：
$ go run producer.go
2020/09/26 22:22:54 send message: this is message
这时候赶紧切回消费者的终端，按control+c，输出如下：
2020/09/26 22:04:02 start consumer...
2020/09/26 22:22:54 i: 0
2020/09/26 22:22:55 i: 1
2020/09/26 22:22:56 i: 2
2020/09/26 22:22:57 i: 3
^C2020/09/26 22:22:57 receive a shutdown signal
2020/09/26 22:22:58 i: 4
2020/09/26 22:22:59 i: 5
2020/09/26 22:23:00 i: 6
2020/09/26 22:23:01 i: 7
2020/09/26 22:23:02 i: 8
2020/09/26 22:23:03 i: 9
2020/09/26 22:23:04 mq's data: this is message
2020/09/26 22:23:04 close the consumer.
上面的演示中，我在第三秒的时候，按下了control + c, 然后程序接着执行，到整个任务执行完毕后，整个程序才真正关闭。
注意点：在control + c到真正关闭这段时间，你也可以通过生产者程序再发一条消息到队列中，看看新的消息是否会被处理（如果会被处理，在消息量大的情况下可能会触发强制关闭，导致最后几条消息执行异常。），这里不做演示了，读者可自行测试。
网友评论