@[TOC](云原生网关 APISIX 的核心流程以源码分析的方式剖析其工作原理-3)

✨博主介绍

🌊 作者主页：苏州程序大白

🌊 作者简介：🏆CSDN人工智能域优质创作者🥇，苏州市凯捷智能科技有限公司创始之一，目前合作公司富士康、歌尔等几家新能源公司

💬如果文章对你有帮助，欢迎关注、点赞、收藏

💅 有任何问题欢迎私信，看到会及时回复
💅关注苏州程序大白，分享粉丝福利

主流程

以 Nginx HTTP Subsystem 为例分析主要执行逻辑，其中一些核心逻辑已在上述小节中流程分析过。

init_by_lua

function _M.http_init(args)
    require("resty.core")

    if require("ffi").os == "Linux" then
        require("ngx.re").opt("jit_stack_size", 200 * 1024)
    end

    require("jit.opt").start("minstitch=2", "maxtrace=4000",
                             "maxrecord=8000", "sizemcode=64",
                             "maxmcode=4000", "maxirconst=1000")

    core.resolver.init_resolver(args)
    -- 生成节点 ID
    core.id.init()

    -- 启用 openresty 的特权进程
    local process = require("ngx.process")
    local ok, err = process.enable_privileged_agent()
    if not ok then
        core.log.error("failed to enable privileged_agent: ", err)
    end

    -- 从 etcd / yaml 本地配置文件获取配置, etcd 有 init 函数
    if core.config.init then
        local ok, err = core.config.init()
        if not ok then
            core.log.error("failed to load the configuration: ", err)
        end
    end
end

init_worker_by_lua

function _M.http_init_worker()
    local seed, err = core.utils.get_seed_from_urandom()
    if not seed then
        core.log.warn('failed to get seed from urandom: ', err)
        seed = ngx_now() * 1000 + ngx.worker.pid()
    end
    math.randomseed(seed)
    -- for testing only
    core.log.info("random test in [1, 10000]: ", math.random(1, 10000))

    -- 进程间事件通信
    local we = require("resty.worker.events")
    local ok, err = we.configure({shm = "worker-events", interval = 0.1})
    if not ok then
        error("failed to init worker event: " .. err)
    end
    -- 服务发现 lib
    local discovery = require("apisix.discovery.init").discovery
    -- 默认没有开启服务发现
    if discovery and discovery.init_worker then
        discovery.init_worker()
    end
    -- 初始化负载均衡器, 方法为空
    require("apisix.balancer").init_worker()
    -- 负载均衡器
    load_balancer = require("apisix.balancer")
    -- TODO admin 流程分析
    require("apisix.admin.init").init_worker()
    -- 注册全局 timer
    require("apisix.timers").init_worker()

    -- 加载所有插件并执行插件 init
    plugin.init_worker()
    -- 初始化 router, 并加载 routes
    router.http_init_worker()

    -- 初始化 services, 加载 services
    require("apisix.http.service").init_worker()
    -- 加载插件配置
    plugin_config.init_worker()
    -- consumer 加载
    require("apisix.consumer").init_worker()

    if core.config == require("apisix.core.config_yaml") then
        core.config.init_worker()
    end

    require("apisix.debug").init_worker()
    -- upstreams 加载
    apisix_upstream.init_worker()
    require("apisix.plugins.ext-plugin.init").init_worker()

    local_conf = core.config.local_conf()

    if local_conf.apisix and local_conf.apisix.enable_server_tokens == false then
        ver_header = "APISIX"
    end
end

access_by_lua

-- access_by_lua 阶段, apisix 没有 rewrite_by_lua
-- ref: https://github.com/apache/apisix/issues/1120
-- ref: https://github.com/apache/apisix/issues/1120#issuecomment-584949073
function _M.http_access_phase()
    local ngx_ctx = ngx.ctx
	...

    -- 从 table 缓存池中获取 table
    -- always fetch table from the table pool, we don't need a reused api_ctx
    local api_ctx = core.tablepool.fetch("api_ctx", 0, 32)
    -- 将 table 储存在 ngx.ctx 中, 下一个阶段共享
    ngx_ctx.api_ctx = api_ctx

    -- 绑定 metatable 
    core.ctx.set_vars_meta(api_ctx)
	...

    -- router 路由匹配
    router.router_http.match(api_ctx)

    -- run global rule
    plugin.run_global_rules(api_ctx, router.global_rules, nil)

    ...
    local enable_websocket = route.value.enable_websocket

    -- route 插件配置绑定
    if route.value.plugin_config_id then
       	...
        route = plugin_config.merge(route, conf)
    end

    -- 获取对应的 service
    if route.value.service_id then
        local service = service_fetch(route.value.service_id)
        ...
        if enable_websocket == nil then
            enable_websocket = service.value.enable_websocket
        end

    else
        ...
    end
    api_ctx.route_id = route.value.id
    api_ctx.route_name = route.value.name

    -- 执行 script
    if route.value.script then
        script.load(route, api_ctx)
        script.run("access", api_ctx)
    else
        -- 插件过滤, 遍历插件列表, 匹配开启的插件, O(n)
        local plugins = plugin.filter(route)
        api_ctx.plugins = plugins

        -- fake 执行 rewrite 阶段
        plugin.run_plugin("rewrite", plugins, api_ctx)
        if api_ctx.consumer then
            local changed
            route, changed = plugin.merge_consumer_route(
                route,
                api_ctx.consumer,
                api_ctx
            )

            core.log.info("find consumer ", api_ctx.consumer.username,
                          ", config changed: ", changed)

            if changed then
                core.table.clear(api_ctx.plugins)
                api_ctx.plugins = plugin.filter(route, api_ctx.plugins)
            end
        end
        -- 执行 access 阶段
        plugin.run_plugin("access", plugins, api_ctx)
    end

    local up_id = route.value.upstream_id

    -- used for the traffic-split plugin
    if api_ctx.upstream_id then
        up_id = api_ctx.upstream_id
    end
	...

    -- websocket 特殊处理
    if enable_websocket then
        api_ctx.var.upstream_upgrade    = api_ctx.var.http_upgrade
        api_ctx.var.upstream_connection = api_ctx.var.http_connection
        core.log.info("enabled websocket for route: ", route.value.id)
    end

    if route.value.service_protocol == "grpc" then
        api_ctx.upstream_scheme = "grpc"
    end

    -- 获取 upstream 节点
    local code, err = set_upstream(route, api_ctx)
    if code then
        core.log.error("failed to set upstream: ", err)
        core.response.exit(code)
    end

    -- 负载均衡
    local server, err = load_balancer.pick_server(route, api_ctx)
    if not server then
        core.log.error("failed to pick server: ", err)
        return core.response.exit(502)
    end

    api_ctx.picked_server = server

    set_upstream_headers(api_ctx, server)

    -- stash ngx ctx 这部分与 Kong 一致, 怀疑是抄来的（95% 置信区间）
    ngx_var.ctx_ref = ctxdump.stash_ngx_ctx()
    local up_scheme = api_ctx.upstream_scheme
    if up_scheme == "grpcs" or up_scheme == "grpc" then
        return ngx.exec("@grpc_pass")
    end

    if api_ctx.dubbo_proxy_enabled then
        return ngx.exec("@dubbo_pass")
    end
end

一些思考

边缘计算

对于互联网设备，网络边缘是设备或包含设备的本地网络与互联网通信的位置。边缘是个比较模糊的术语。例如，可以将用户的计算机或 IoT 摄像头内部的处理器视为网络边缘，但也可以将用户的路由器、ISP 或本地边缘服务器视为边缘。重要的是，网络边缘在地理位置上靠近设备，与源站和云服务器不同，后者可能与它们相互通信的设备相距很远。

完全减轻额外硬件需求的一种方法是利用边缘服务器。例如，借助 Cloudflare 分散在全球各地的 194 个边缘服务器网络，Cloudflare 的客户可以使用 Cloudflare Workers 在全球范围内运行边缘代码。10

Cloudflare 的边缘计算是基于 Edge Gateway（边缘网关、边缘集群）的 Serverless 代码执行，提供了 JS 代码执行，以及 WASM 二进制。11

一些相关的 Issue:

• Support wasm in openresty?
• feature: support WebAssembly in apisix.

Serverless

APISIX 的 Serverless 插件功能支持注入任何 Lua 脚本，而 Kong 网关也有类似的插件功能。

Serverless 插件支持执行简单的函数方法。

WebAssembly

APISIX 自 2019 年发起提案，试图通过 WebAssembly 来扩展 Lua 贫乏的生态。 2021 年，在 WebAssembly 运行时的技术选型上，APISIX 的技术团队更偏向使用由 Fastly 团队 支撑13的 wasmtime 项目。

开源的 WebAssembly 除了 wasmtime 还有14：

• WasmEdge（前身 SSVM），由 Second State 开源的 CNCF 沙箱项目。
• Wasmer，Dart 语言使用的 Wasm 运行时。
• Lucet，由 Fastly 开源的 Bytecode Alliance 的 项目，将会与 wasmtime 合并。

在 Issue #157 的讨论中，Wasmer 的 CEO 也来插了一嘴， 希望 APISIX 能够选型 Wasmer 运行时，APISIX 成员给了 Wasmer 一个大大的赞， 最终在 api7/wasm-nginx-module 插件中， 还是使用 wasmtime 运行时实现了对 WebAssembly 的支持。

Service Mesh

APISIX 的 Service Mesh 项目 api7/apisix-mesh-agent，将 APISIX Proxy 作为 Sidecar 运作在数据平面。通过实现控制平面的接口，接入类似 Istio 或 Kuma（由 Kong 创建捐赠给 CNCF） 的控制平面，形成一套完整的 Service Mesh 方案。 该项目本质上是使用 APISIX 替换了 Istio 中的 Envoy。

值得一提的是 Kong 类似的 Service Mesh 项目，叫做 Kong Mesh，目前只提供企业版本。