最新要闻

广告

手机

iphone11大小尺寸是多少？苹果iPhone11和iPhone13的区别是什么？

iphone11大小尺寸是多少？苹果iPhone11和iPhone13的区别是什么？

 警方通报辅警执法直播中被撞飞：犯罪嫌疑人已投案

警方通报辅警执法直播中被撞飞：犯罪嫌疑人已投案

家电

  1. 首页
  2. 移动通信
  3. 内容

天天观速讯丨【网关开发】5.Openresty 自定义负载均衡与流量转发

来源：博客园
目录
  • 背景
  • 应用架构
  • 实现
    • 插件
    • 配置文件
    • 流量转发
    • 负载均衡器
  • 测试
  • 总结
  • 扩展
    • ip_hash
    • url_hash

背景

静态的nginx配置需要将负载均衡的服务节点信息都配置在配置文件中。现在微服务或云服务都会接入一些服务发现或者云控平台场景，经常需要更换节点，如果每次都要更新配置并且重启服务是无法接受的，所以需要网关提供动态扩展，实时更新自己负载均衡节点的能力，使用openresty网关需要使用lua扩展来实现自定义负载均衡的能力。


【资料图】

应用架构

实现

本文展示openresty如何开发，有一些结构体与细节可以直接去git上查看https://github.com/zhaoshoucheng/openresty/tree/main/pkg/lua_script使用插件地址：https://github.com/openresty/lua-resty-balancer"整体实现摸快配置文件摸快：nginx的conf配置文件lua-resty-balancer 插件：负载均衡器upstream_context : 从etcd中获取相应的服务节点，通过负载均衡类型，获取负载均衡器

插件

编译lua-resty-balancer插件简单，进入文件夹make就可以，然后会生成librestychash.so文件，将其放入到我们可以引入到的文件夹中例如

lua_package_cpath "/data/openresty/pkg/lua_script/?.so;;";

配置文件

upstream server_test1 {server 0.0.0.0:1234;balancer_by_lua_block {    if etcd_source_module then require "upstream.balance".do_balance("server_test1") else ngx.exit(502) end    }}

etcd_source_module etcd数据模块，负责从etcd中获取服务server_test1的IP，这里知识检查作用。do_balance 实现流量转发的模块函数

流量转发

local transform_data_simple = utils.transform_data_simplelocal module_name = (...):match("(.-)[^%.]+$")local upstream_context = require(module_name.."upstream_context")local conf = require(module_name.."config")local utils = require(module_name.."utils")local ngx_balancer = require "ngx.balancer"local cjson = require "cjson.safe"-- 从etcd中获取数据，通过transform_data_simple 转化数据local function get_upstream_context(name)    local ctx = upstream_contexts[name]    if not ctx then        local ups = etcd_source_module:get_value(name)        if not ups then            return nil, "upstream configure not found: "..name        end        ups = transform_data_simple(ups)        if ups then            ctx = upstream_context.new(name, ups)            upstream_contexts[name] = ctx        end    end    return ctxendfunction _M.do_balance(ups_name)    local ctx = ngx.ctx    local uctx, err = get_upstream_context(ups_name)    if not uctx then        ngx.log(ngx.ERR, ups_name..", "..tostring(err))        ngx.exit(502)        return    end    -- 获取负载均衡器    local b, err = uctx:get_prefered_balancer()    if not b then        ngx.log(ngx.ERR, "failed to get balancer: "..tostring(err))        ngx.exit(502)        return    end    -- 上次失败的节点要在下次节点中避免选择    local key, idx    local sn, sc = ngx_balancer.get_last_failure()    if not sn then        -- first call        local ok, err = ngx_balancer.set_more_tries(3)        if err and #err > 0 then            ngx.log(ngx.WARN, err)        end        key, idx = b:find(ctx.balance_key)        if not key then            ngx.log(ngx.ERR, "failed to get upstream endpoint")            ngx.exit(502)            return        end    else        -- 根据上次的负载情况选择下次的节点        key, idx = b:next(ctx.latest_idx)        ngx.log(ngx.WARN, "rebalancing: "..sn..", "..tostring(sc))    end    ngx.log(ngx.ERR,"do_balance b"..cjson.encode(b))  -- 打印负载均衡器    -- 从所有节点中进行选择    local peer = uctx._all_nodes[key]    ngx.log(ngx.ERR,"do_balance uctx._all_nodes"..cjson.encode(uctx._all_nodes))    -- 打印_all_node 信息    if not peer then        ngx.log(ngx.ERR, "failed to get upstream endpoint: "..tostring(key))        ngx.exit(502)        return    end    ctx.latest_peer = peer    ctx.latest_key = key    ctx.latest_idx = idx    -- 流量转发函数    local ok, err = ngx_balancer.set_current_peer(peer.ip, peer.port)    ngx.log(ngx.INFO, string.format(" do balance ip :%s point: %s", peer.ip, peer.port))  -- 打印目标节点    if not ok then        ngx.log(ngx.ERR, string.format("error while setting current upstream peer %s: %s", peer.ip, err))        ngx.exit(500)        return    endend

看下_all_nodes的结构，map结构key值是传入负载均衡器的值，后面会有用到

2023/01/19 14:41:32 [error] 17588#0: *234587 [lua] balance.lua:66: do_balance(): do_balance uctx._all_nodes{"10.218.22.246\u00008090":{"state":"up","fail_timeout":3000,"weight":1,"ip":"10.218.22.246","max_fail":3,"port":8090},"10.218.22.239\u00008090":{"state":"up","fail_timeout":3000,"weight":1,"ip":"10.218.22.239","max_fail":3,"port":8090}} while connecting to upstream, client: 10.99.4.169, server: server_test.com, request: "GET /ping HTTP/1.1", host: "server_test.com:9090"

打印每一次获得负载均衡器

2023/01/19 14:45:32 [error] 17656#0: *234610 [lua] balance.lua:67: do_balance(): do_balance b{"gcd":1,"cw":1,"nodes":{"10.218.22.239\u00008090":1,"10.218.22.246\u00008090":1},"last_id":"10.218.22.239\u00008090","max_weight":1} while connecting to upstream, client: 10.99.4.169, server: server_test.com, request: "GET /ping HTTP/1.1", host: "server_test.com:9090"

负载均衡器

-- 获取负载均衡器local function get_prefered_balancer(self)    local b = self._prefered_balancer    local err = nil    if not b then        local nodes, _ = process_upstream_nodes(self._ups.nodes)        local lb = self._ups.load_balance        local err        ngx.log(ngx.ERR,"get_prefered_balancer nodes"..cjson.encode(nodes))  -- 打印nodes数据        b, err = balancers.create(lb.type, nodes)        if not b then            return nil, err        end        self._prefered_balancer = b    end    return b, errend

不需要过度关注结构体的设计，用法场景不同，结构体自然不同，比较灵活，process_upstream_nodes目的就是选出目标所有可用的节点，可以结合健康检查结果，或者流量规则进行使用最终生成的nodes就是map的key,权重的结构。将其作为负载均衡插件的参数。

2023/01/19 14:29:14 [error] 17335#0: *234481 [lua] upstream_context.lua:78: get_prefered_balancer(): get_prefered_balancer nodes{"10.218.22.239\u00008090":1,"10.218.22.246\u00008090":1} while connecting to upstream, client: 10.99.4.169, server: server_test.com, request: "GET /ping HTTP/1.1", host: "server_test.com:9090"

选择负载均衡器,将提供的负载均衡器封装成函数

local roundrobin = require("resty.balancer.roundrobin")local chash = require("resty.balancer.chash")local iphash = require("resty.balancer.iphash")local urlhash = require("resty.balancer.urlhash")local _M = { }local function create(typ, ...)    if typ == "roundrobin" or typ == "round_robin" then        return roundrobin:new(...)    elseif typ == "chash" then        return chash:new(...)    elseif typ == "iphash" or typ == "ip_hash" then        return iphash:new(...)    elseif typ == "urlhash" or typ == "url_hash" then        return urlhash:new(...)    else        return nil, "unsupported balancer type: "..tostring(typ)    endend_M.create = createreturn _M

测试

round_robin 测试

ip_hash 测试

总结

当然企业级工程不可能这么简单，这只是核心的细节。我试图将复杂的整体工程拆分成一个一个的小知识点来进行分析，学习。完善自己的知识体系。所以单单是看单个知识点都比较简单。尽量避免了一些复杂的结构体和复杂的业务场景来讲解架构与知识点，这样更加的容易理解。能把一件事情讲明白其实是一件很不容易的事情，如果有不解或者有误的地方欢迎沟通交流,有些细节代码也上传了github,欢迎查看。https://github.com/zhaoshoucheng/openresty/tree/main/pkg/lua_script

扩展

因为github插件上提供的都是基础功能round_robin和chash，其实其他负载均衡算法也可以在这上面进行扩展。或者实现自己的负载均衡算法。现在提供ip_hash 和 url_hash的写法

ip_hash

local module_name = (...):match("(.-)[^%.]+$")local chash = require(module_name.."chash")local setmetatable = setmetatablelocal _M = { _VERSION = "0.1" }local _MT = { __index = _M }function _M.new(_, nodes)    return setmetatable({        _chb = chash:new(nodes)    }, _MT)endfunction _M:reinit(nodes)    self._chb:reinit(nodes)endfunction _M:set(...)    self._chb:set(...)endfunction _M:find()    return self._chb:find(ngx.var.remote_addr) -- TODO: use xff ?endfunction _M:next(...)    self._chb:next(...)endreturn _M

url_hash

local module_name = (...):match("(.-)[^%.]+$")local chash = require(module_name.."chash")local setmetatable = setmetatablelocal _M = { _VERSION = "0.1" }local _MT = { __index = _M }function _M.new(_, nodes)    return setmetatable({        _chb = chash:new(nodes)    }, _MT)endfunction _M:reinit(nodes)    self._chb:reinit(nodes)endfunction _M:set(...)    self._chb:set(...)endfunction _M:find()    return self._chb:find(ngx.var.uri)endfunction _M:next(...)    self._chb:next(...)endreturn _M

关键词： 负载均衡 配置文件