在實際開發(fā)中經(jīng)常涉及到項目的升級，而該升級不能簡單的上線就完事了，需要驗證該升級是否兼容老的上線，因此可能需要并行運行兩個項目一段時間進行數(shù)據(jù)比對和校驗，待沒問題后再進行上線。這其實就需要進行流量復制，把流量復制到其他服務器上，一種方式是使用如 tcpcopy 引流；另外我們還可以使用 nginx 的 HttpLuaModule 模塊中的 ngx.location.capture_multi 進行并發(fā)執(zhí)行來模擬復制。
構造兩個服務
Java 代碼
location /test1 {
keepalive_timeout 60s;
keepalive_requests 1000;
content_by_lua '
ngx.print("test1 : ", ngx.req.get_uri_args()["a"])
ngx.log(ngx.ERR, "request test1")
';
}
location /test2 {
keepalive_timeout 60s;
keepalive_requests 1000;
content_by_lua '
ngx.print("test2 : ", ngx.req.get_uri_args()["a"])
ngx.log(ngx.ERR, "request test2")
';
} 通過 ngx.location.capture_multi 調用
Java 代碼
location /test {
lua_socket_connect_timeout 3s;
lua_socket_send_timeout 3s;
lua_socket_read_timeout 3s;
lua_socket_pool_size 100;
lua_socket_keepalive_timeout 60s;
lua_socket_buffer_size 8k;
content_by_lua '
local res1, res2 = ngx.location.capture_multi{
{ "/test1", { args = ngx.req.get_uri_args() } },
{ "/test2", { args = ngx.req.get_uri_args()} },
}
if res1.status == ngx.HTTP_OK then
ngx.print(res1.body)
end
if res2.status ~= ngx.HTTP_OK then
--記錄錯誤
end
';
} 此處可以根據(jù)需求設置相應的超時時間和長連接連接池等；ngx.location.capture 底層通過cosocket 實現(xiàn)，而其支持 Lua 中的協(xié)程，通過它可以以同步的方式寫非阻塞的代碼實現(xiàn)。
此處要考慮記錄失敗的情況，對失敗的數(shù)據(jù)進行重放還是放棄根據(jù)自己業(yè)務做處理。
AB 測試即多版本測試，有時候我們開發(fā)了新版本需要灰度測試，即讓一部分人看到新版，一部分人看到老版，然后通過訪問數(shù)據(jù)決定是否切換到新版。比如可以通過根據(jù)區(qū)域、用戶等信息進行切版本。
比如京東商城有一個 cookie 叫做__jda，該 cookie 是在用戶訪問網(wǎng)站時種下的，因此我們可以拿到這個 cookie，根據(jù)這個 cookie 進行版本選擇。
比如兩次清空 cookie 訪問發(fā)現(xiàn)第二個數(shù)字串是變化的，即我們可以根據(jù)第二個數(shù)字串進行判斷。
__jda=122270672.1059377902.1425691107.1425691107.1425699059.1
__jda=122270672.556927616.1425699216.1425699216.1425699216.1。
判斷規(guī)則可以比較多的選擇，比如通過尾號；要切 30% 的流量到新版，可以通過選擇尾號為 1，3,5 的切到新版，其余的還停留在老版。
Java 代碼
map $cookie___jda $ab_key {
default "0";
~^\d+\.\d+(?P<k>(1|3|5))\. "1";
} 使用 map 映射規(guī)則，即如果是到新版則等于 "1"，到老版等于 “0”； 然后我們就可以通過 ngx.var.ab_key 獲取到該數(shù)據(jù)。
Java 代碼
location /abtest1 {
if ($ab_key = "1") {
echo_location /test1 ngx.var.args;
}
if ($ab_key = "0") {
echo_location /test2 ngx.var.args;
}
} 此處也可以使用 proxy_pass 到不同版本的服務器上
Java 代碼
location /abtest2 {
if ($ab_key = "1") {
rewrite ^ /test1 break;
proxy_pass http://backend1;
}
rewrite ^ /test2 break;
proxy_pass http://backend2;
} 首先下載 lua-resty-cookie
Java 代碼
cd /usr/example/lualib/resty/
wget https://raw.githubusercontent.com/cloudflare/lua-resty-cookie/master/lib/resty/cookie.lua Java 代碼
location /abtest3 {
content_by_lua '
local ck = require("resty.cookie")
local cookie = ck:new()
local ab_key = "0"
local jda = cookie:get("__jda")
if jda then
local v = ngx.re.match(jda, [[^\d+\.\d+(1|3|5)\.]])
if v then
ab_key = "1"
end
end
if ab_key == "1" then
ngx.exec("/test1", ngx.var.args)
else
ngx.print(ngx.location.capture("/test2", {args = ngx.req.get_uri_args()}).body)
end
';
} 首先使用 lua-resty-cookie 獲取 cookie，然后使用 ngx.re.match 進行規(guī)則的匹配，最后使用 ngx.exec 或者 ngx.location.capture 進行處理。此處同時使用 ngx.exec 和ngx.location.capture 目的是為了演示，此外沒有對 ngx.location.capture 進行異常處理。
Lua 中沒有線程和異步編程編程的概念，對于并發(fā)執(zhí)行提供了協(xié)程的概念，個人認為協(xié)程是在A運行中發(fā)現(xiàn)自己忙則把 CPU 使用權讓出來給B使用，最后 A 能從中斷位置繼續(xù)執(zhí)行，本地還是單線程，CPU 獨占的；因此如果寫網(wǎng)絡程序需要配合非阻塞 I/O 來實現(xiàn)。
ngx_lua 模塊對協(xié)程做了封裝，我們可以直接調用 ngx.thread API 使用，雖然稱其為“輕量級線程”，但其本質還是 Lua 協(xié)程。 該 API 必須配合該 ngx_lua 模塊提供的非阻塞 I/O API 一起使用，比如我們之前使用的 ngx.location.capture_multi 和 lua-resty-redis、lua-resty-mysql 等基于 cosocket 實現(xiàn)的都是支持的。
通過 Lua 協(xié)程我們可以并發(fā)的調用多個接口，然后誰先執(zhí)行成功誰先返回，類似于 BigPipe 模型。
Java 代碼
location /api1 {
echo_sleep 3;
echo api1 : $arg_a;
}
location /api2 {
echo_sleep 3;
echo api2 : $arg_a;
} 我們使用 echo_sleep 等待 3 秒。
Java 代碼
location /serial {
content_by_lua '
local t1 = ngx.now()
local res1 = ngx.location.capture("/api1", {args = ngx.req.get_uri_args()})
local res2 = ngx.location.capture("/api2", {args = ngx.req.get_uri_args()})
local t2 = ngx.now()
ngx.print(res1.body, "<br/>", res2.body, "<br/>", tostring(t2-t1))
';
} 即一個個的調用，總的執(zhí)行時間在6秒以上，比如訪問 http://192.168.1.2/serial?a=22
Java 代碼
api1 : 22
api2 : 22
6.0040001869202 Java 代碼
location /concurrency1 {
content_by_lua '
local t1 = ngx.now()
local res1,res2 = ngx.location.capture_multi({
{"/api1", {args = ngx.req.get_uri_args()}},
{"/api2", {args = ngx.req.get_uri_args()}}
})
local t2 = ngx.now()
ngx.print(res1.body, "<br/>", res2.body, "<br/>", tostring(t2-t1))
';
} 直接使用 ngx.location.capture_multi 來實現(xiàn)，比如訪問 http://192.168.1.2/concurrency1?a=22
Java 代碼
api1 : 22
api2 : 22
3.0020000934601 Java 代碼
location /concurrency2 {
content_by_lua '
local t1 = ngx.now()
local function capture(uri, args)
return ngx.location.capture(uri, args)
end
local thread1 = ngx.thread.spawn(capture, "/api1", {args = ngx.req.get_uri_args()})
local thread2 = ngx.thread.spawn(capture, "/api2", {args = ngx.req.get_uri_args()})
local ok1, res1 = ngx.thread.wait(thread1)
local ok2, res2 = ngx.thread.wait(thread2)
local t2 = ngx.now()
ngx.print(res1.body, "<br/>", res2.body, "<br/>", tostring(t2-t1))
';
} 使用 ngx.thread.spawn 創(chuàng)建一個輕量級線程，然后使用 ngx.thread.wait 等待該線程的執(zhí)行成功。比如訪問 http://192.168.1.2/concurrency2?a=22
Java 代碼
api1 : 22
api2 : 22
3.0030000209808 其有點類似于 Java 中的線程池執(zhí)行模型，但不同于線程池，其每次只執(zhí)行一個函數(shù)，遇到 IO 等待則讓出 CPU 讓下一個執(zhí)行。我們可以通過下面的方式實現(xiàn)任意一個成功即返回，之前的是等待所有執(zhí)行成功才返回。
Java 代碼
local ok, res = ngx.thread.wait(thread1, thread2) 
掃碼關注鍍金池微信號
手機訂課，優(yōu)惠多多