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