您现在的位置是:首页 > 文章详情

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

日期:2018-12-17点击:416

前言


  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更加实时,更加稳定更高并发,应该算是初创团队的最优方案了。觉得好的可以收藏下点个赞支持一下。

原文链接:https://my.oschina.net/mHUMTSy9311/blog/2989710
关注公众号

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。

持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。

文章评论

共有0条评论来说两句吧...

文章二维码

扫描即可查看该文章

点击排行

推荐阅读

最新文章