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

原创
2018/12/18 01:35
阅读数 9.7K

前言


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

展开阅读全文
加载中

作者的其它热门文章

打赏
0
30 收藏
分享
打赏
2 评论
30 收藏
0
分享
返回顶部
顶部
返回顶部
顶部