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