基于RabbitMQ解决微信大量并发回调的问题(微信产品服务端工程师值得看)

前言

1. 大量粉丝的公众号已经有非常多了，数百万数千万的都很多，再加上很多微信开放平台，导致接收的微信回调非常多，如：关注/取消关注，获取地理位置，发消息，向用户推送模板消息，点击菜单，扫描带参二维码。这么多场景都会导致微信回调，若单纯使用Tomcat，Undertow，apache之类的Web容器去处理，使用资源成本比较大，而且微信要求在5s内进行回复直接异步处理也需要很多机器才处理得来。可以选择使用些高性能IO框架去解决 如果：smart-io, tio, netty, swoole之类去增加吞吐量，但是这样增加了学习成本和维护成本。我之前写过一篇文章是用Redis来解决的解决模板消息回调的困扰, 但是死循环会长期占用一条RUNNABLE线程，比较浪费cpu资源，可以优化成用分布式锁+rangeAll降低循环频率从而减轻负载，这次我要介绍的是使用RabbitMQ去处理（Nginx + Lua + RabbitMQ）。

准备工作

• 搭建RabbitMQ集群，我有个小伙伴写了篇博客，现在分享一下: RabbitMQ集群原理和部署

  SpringBoot整合RabbitMQ

• Lua连接RabbitMQ官方推介使用rabbitmqstomp

  # 开启RabbitMQ的rabbitmq_stomp插件(版本过低的可能会没有这个插件)
  rabbitmq-plugins enable rabbitmq_stomp
  

• 开启*rabbitmq_stomp插件后会监听多一个端口 61613 (很重要,没有测试过端口冲突的情况)，如果是单机RabbitMQ的话，在Lua脚本中应该连接这个端口。

• 都说是针对生产环境搭建的，所以HAProxy肯定少不了，修改haproxy的配置文件 /etc/haproxy/haproxy.cfg ，不一样的自行切换。下面贴上HAProxy的配置：

listen rabbitmqstomp 10.0.0.1:5670
mode tcp
balance roundrobin
server rabbit-master 10.0.0.2:61613 check inter 5s rise 2 fall 3
server rabbit-node1 10.0.0.3:61613 check inter 5s rise 2 fall 3

Lua脚本

-- 打印日志
function log(fileName, content)
        local  f = assert(io.open(fileName,'a'))
        f:write(content..'\n')
        f:close()
end

-- 读取请求Get参数
ngx.req.read_body()
local data, err = ngx.req.get_body_data()
if (nil ~= err) then
    log(log_file, "body is null: " + err)
    return
end
if(nil == data)
then
 params = ngx.req.get_uri_args()
 ngx.say(ngx.var.arg_echostr)
 ngx.exit(ngx.HTTP_OK);
 return
end


local log_file = '/var/log/nginx/logs/redis.log'

local rabbitmq = require "resty.rabbitmqstomp"
local cjson = require "cjson"

local opts = {
    host = '10.0.0.1',
    port = '5670',
    username = 'abcd',
    password = 'abcd',
    vhost = 'abc'
}

local mq, err = rabbitmq:new(opts)
if not mq then
    log(log_file, "can not new rabbitmq: " .. err)
    return
end

mq:set_timeout(10000)

local ok, err = mq:connect(opts['host'], opts['port'])
if not ok then
    log(log_file, "Connect Error: " .. err)
    mq:close()
    return
end
	
local args, err = ngx.req.get_uri_args()
if (nil ~= err) then
    log(log_file, "getURI: " + err)
    mq:close()
    return
end

local message = {}
message['signature'] = args['msg_signature']
message['timestamp'] = args['timestamp']
message['nonce'] = args['nonce']
message['data'] = data
local text = cjson.encode(message)
ngx.log(ngx.ERR, text)

local headers = {}
-- 消息发送到哪里 /exchange/交换机名称/routing_key名称
headers["destination"] = "/exchange/topic.wechatEvent.receive/wechatEvent.receive"
-- -- 是否持久化
headers["persistent"] = "true"
-- -- 消息格式
headers["content-type"] = "application/json"
local ok, err = mq:send(text, headers)
if not ok then
    ngx.log(ngx.ERR, "cannot send mq")
    mq:close()
    return
end


-- 消息保持长连接，第一个参数表示连接超时时间，第二个参数是表示连接池大小
-- -- 由于 rabbitmq 连接建立比较耗时，所以保持连接池是非常必要的
local ok, err = mq:set_keepalive(10000, 500)
if not ok then
    log(log_file, "set keepalive error: " .. err)
    mq:close()
    return
end
ngx.say('success')
ngx.exit(ngx.HTTP_OK)

Java代码

@RabbitListener(queues = MQConstants.Queue.WECHAT_EVENT_RECEIVE)
    public void handle(WechatReceiveMessage receiveMessage) throws Exception {
	//TODO
}

总结

对比之前之前用Redis处理，用RabbitMQ更加实时，更加稳定更高并发，应该算是初创团队的最优方案了。觉得好的可以收藏下点个赞支持一下。