添加版本的概念,版本增加才重新 up/down upstream

@@ -308,14 +308,28 @@ timer_handler(true,2,limit_ip_access_conf_to_worker)
 timer_handler(true,2,query_common_conf_to_worker)
 
 
+local  stream_ctx={}
 function updownstream()
   local keys=upstream_cache:get_keys()
+  
   for _,k in ipairs(keys) do
-   local value=upstream_cache:get(k)
-   local v=cjson.decode(value)
-   local ok,err=upstream.set_peer_down(v["upstream"],v["backup"],v["id"],v["value"])  
-   if not ok then
-     ngx.log(ngx.ERR,"up or down stream err:",ngx.worker.id(),value,err)
+   if string.sub(k,1,1)=="d" then 
+     
+     local vKey="v" .. string.sub(k,2)
+     local version=upstream_cache:get(vKey)
+ 
+     local value=upstream_cache:get(k)
+     local v=cjson.decode(value)
+ 
+     if ( not stream_ctx[vKey] ) or stream_ctx[vKey] < version then
+       local ok,err=upstream.set_peer_down(v["upstream"],v["backup"],v["id"],v["value"])  
+       if not ok then
+         ngx.log(ngx.ERR,"up or down stream err:",ngx.worker.id(),value,err)
+       else 
+          stream_ctx[vKey]=version
+       end
+     end
+ 
    end
   end
 end

@@ -48,17 +48,22 @@ end
 
 
 function upordown(upstream_name,is_backup,peer_id,down_value)
-   local ok, err = upstream.set_peer_down(upstream_name, is_backup, peer_id, down_value)
    local t={}
    t["upstream"]=upstream_name
    t["backup"]=is_backup
    t["id"]=peer_id
    t["value"]=down_value 
-   local suc=cache:set("d:" .. upstream_name .. ":" .. tostring(id)  .. ":" .. tostring(is_backup),json.encode(t),300)
+   local key="d:" .. upstream_name .. ":" .. tostring(id)  .. ":" .. tostring(is_backup)
+   local vKey="v:" .. upstream_name .. ":" .. tostring(id)  .. ":" .. tostring(is_backup)
+   cache:add(vKey,0) 
+   local v,err=cache:incr(vKey,1)
+   if not v then
+     return false
+   end
+   local suc=cache:set(key,json.encode(t))
    return suc
 end
 
-
 local args=ngx.req.get_uri_args()
 local method=args["method"]
 if method == "list" then