云原生网关进阶：APISIX 插件开发与流量治理全攻略（附 3 个实战插件）

你用 APISIX 做了反向代理、做了限流、做了 JWT 认证，但遇到业务需求时发现官方插件不够用？这篇文章手把手教你写 APISIX 自定义插件——从”Hello World”到生产级插件全流程，附带 3 个可直接复用的实战案例。

---

一、为什么需要自定义插件？

APISIX 内置了 80+ 官方插件，覆盖了大部分通用场景。但真实业务中你总会遇到这些需求：

需求场景	有现成插件吗	解决方案
按用户等级限流	⚠️ 只能按 IP/Key	✅ 自定义分级限流
请求参数签名校验	❌ 没有	✅ 自定义签名验证
AI 服务路由（模型版本切换）	❌ 没有	✅ 自定义智能路由
响应数据脱敏（手机号/身份证）	❌ 没有	✅ 自定义响应过滤
多租户请求隔离	⚠️ 不够灵活	✅ 自定义租户鉴权

好消息：APISIX 的插件开发基于 LuaJIT + OpenResty，性能极高，写起来也不复杂。一个熟练的开发者半天就能搞定一个生产级插件。

---

二、APISIX 插件架构速懂

请求进来
    ↓
[rewrite] → 路径改写 / 重定向
    ↓
[access] → 认证、限流、黑白名单 ← 大部分逻辑在这里
    ↓
[proxy] → 代理转发前后的处理
    ↓
[header_filter] → 修改响应头
    ↓
[body_filter] → 修改响应体
    ↓
响应出去

每个阶段都可以插入你的代码，最常用的是 rewrite 和 access。

插件文件结构

apisix/plugins/
└── my_plugin.lua        # 主文件（必选）
└── my_plugin/
    └── utils.lua         # 工具函数（可选）
└── t/
    └── plugin/my_plugin.t # 测试文件（可选）

---

三、Hello World 插件

-- apisix/plugins/hello-world.lua
local core = require("apisix.core")
local schema = {
    type = "object",
    properties = {
        message = {
            type = "string",
            default = "Hello from custom plugin!",
        },
    },
}

local _M = {
    version = 0.1,
    priority = 10000,
    name = "hello-world",
    schema = schema,
}

function _M.rewrite(conf, ctx)
    -- 在 rewrite 阶段直接返回自定义消息
    local msg = conf.message or "Hello!"

    -- 写入日志（开发调试用）
    core.log.info("hello-world plugin triggered: ", msg)

    -- 返回 JSON 响应，不再继续代理
    return 200, {
        ["Content-Type"] = "application/json",
    }, core.json.encode({
        code = 200,
        message = msg,
        from = "custom-plugin",
        timestamp = ngx.now(),
    })
end

return _M

在 APISIX 配置中启用：

curl -X PUT http://localhost:9180/apisix/routes/1 \
  -H 'X-API-KEY: your-api-key' \
  -d '{
    "uri": "/hello",
    "plugins": {
      "hello-world": {
        "message": "欢迎访问我的 API 网关！"
      }
    }
  }'

测试：

$ curl http://localhost:9080/hello
{"code":200,"message":"欢迎访问我的 API 网关！","from":"custom-plugin","timestamp":1742544000}

---

四、实战插件 1 — 分级限流

场景：普通用户每秒最多 10 个请求，VIP 用户每秒最多 50 个请求，超级 VIP 不限流。

-- apisix/plugins/tiered-rate-limit.lua
local core = require("apisix.core")
local limit_req = require("resty.limit.req")

local schema = {
    type = "object",
    properties = {
        tiers = {
            type = "object",
            properties = {
                normal = { type = "integer", minimum = 1, default = 10 },
                vip = { type = "integer", minimum = 1, default = 50 },
                super_vip = { type = "integer", minimum = -1, default = -1 },  -- -1 表示不限
            },
        },
        user_header = { type = "string", default = "X-User-Tier" },
        burst = { type = "integer", default = 5 },
    },
}

local _M = {
    version = 0.2,
    priority = 9990,
    name = "tiered-rate-limit",
    schema = schema,
}

-- 为每个级别维护独立的限流器
local limiters = {}

function get_limiter(tier_name, rate)
    if not limiters[tier_name] then
        limiters[tier_name], _ = limit_req.new(rate, 20)   -- rate + burst
    end
    return limiters[tier_name]
end

function _M.access(conf, ctx)
    -- 从请求头读取用户等级
    local user_tier = ngx.req.get_headers()[conf.user_header]
    if not user_tier then
        user_tier = "normal"  -- 默认普通用户
    end

    local tiers = conf.tiers or {}
    local rate_limit = tiers[user_tier] or tiers.normal or 10

    -- super_vip 直接放行
    if rate_limit == -1 then
        core.log.info("super_vip 用户放行: ", user_tier)
        return
    end

    -- 执行限流检查
    local limiter = get_limiter(user_tier, rate_limit)
    local delay, err = limiter:wait(1)  -- key 用 remote_addr

    if delay > 0 then
        -- 请求被排队等待
        core.log.info("请求延迟: ", delay, ", 用户等级: ", user_tier)
    elseif not delay then
        -- 被限流了
        core.log.warn("触发限流! 用户等级: ", user_tier)
        return 429, {
            ["Content-Type"] = "application/json",
            ["Retry-After"] = "1",
        }, core.json.encode({
            error = "rate_limited",
            tier = user_tier,
            limit = rate_limit,
            message = "请求过于频繁，请稍后再试",
        })
    end
end

return _M

使用示例：

# 给不同路由配置不同限流策略
curl -X PUT http://localhost:9180/apisix/routes/api-vip \
  -H 'X-API-KEY: your-key' \
  -d '{
    "uri": "/api/*",
    "plugins": {
      "tiered-rate-limit": {
        "tiers": {"normal": 10, "vip": 50, "super_vip": -1},
        "user_header": "X-User-Tier"
      }
    },
    "upstream": {
      "nodes": {"192.168.1.100:8080": 1}
    }
  }'

客户端请求时带上等级头：

# 普通：10 req/s
curl -H 'X-User-Tier: normal' http://localhost:9080/api/data

# VIP：50 req/s
curl -H 'X-User-Tier: vip' http://localhost:9080/api/data

# 超级VIP：不限
curl -H 'X-User-Tier: super_vip' http://localhost:9080/api/data

---

五、实战插件 2 — AI 服务智能路由

场景：你有两套 AI 推理服务——本地部署的 DeepSeek（便宜但慢）和调用云端 GPT API（贵但快），希望根据请求复杂度自动路由。

-- apisix/plugins/ai-smart-router.lua
local core = require("apisix.core")
local http = require("resty.http")
local cjson = require("cjson.safe")

local schema = {
    type = "object",
    properties = {
        routes = {
            type = "array",
            items = {
                type = "object",
                properties = {
                    name = { type = "string" },
                    upstream_url = { type = "string" },
                    max_tokens = { type = "integer", default = 4096 },
                    cost_per_1k = { type = "number" },
                    timeout_ms = { type = "integer", default = 30000 },
                },
                required = {"name", "upstream_url"},
            },
        },
        fallback_route = { type = "string" },
        token_header = { type = "string", default = "X-Estimated-Tokens" },
    },
}

local _M = {
    version = 0.1,
    priority = 9980,
    name = "ai-smart-router",
    schema = schema,
}

function _M.rewrite(conf, ctx)
    -- 解析请求体中的 token 数量估算
    local body = core.request.get_body()
    local estimated_tokens = 0

    if body then
        local parsed = cjson.decode(body)
        if parsed and parsed.messages then
            for _, msg in ipairs(parsed.messages) do
                if msg.content then
                    -- 粗略估算：中文字符约 1.5 tokens/字
                    local content_len = string.len(msg.content)
                    estimated_tokens = estimated_tokens + math.ceil(content_len * 1.5)
                end
            end
        end
    end

    -- 如果请求头有显式指定，优先使用
    local header_tokens = tonumber(ngx.req.get_headers()[conf.token_header])
    if header_tokens then
        estimated_tokens = header_tokens
    end

    core.log.info("估算 Token 数: ", estimated_tokens)

    -- 选择最优路由
    local target_route = nil
    table.sort(conf.routes, function(a, b)
        return (a.max_tokens or 0) < (b.max_tokens or 0)
    end)

    for _, route in ipairs(conf.routes) do
        if estimated_tokens <= route.max_tokens then
            target_route = route
            break
        end
    end

    -- 如果没有匹配的路由，使用 fallback 或最大的那个
    if not target_route then
        target_route = conf.fallback_route and find_route_by_name(conf, conf.fallback_route)
            or conf.routes[#conf.routes]
    end

    if not target_route then
        return 500, {["Content-Type"] = "application/json"}, 
            cjson.encode({error = "no_available_route"})
    end

    -- 设置上游地址和超时
    local url_parsed = core.utils.parse_uri(target_route.upstream_url)
    ctx.var.upstream_host = url_parsed.host
    ctx.var.upstream_port = url_parsed.port or 80

    -- 记录路由决策（可接入监控）
    ngx.ctx.ai_route = {
        selected = target_route.name,
        estimated_tokens = estimated_tokens,
        cost_per_1k = target_route.cost_per_1k,
    }

    core.log.info("AI 路由决策: ", target_route.name, 
                  " (tokens:", estimated_tokens, ")")
end

function _M.header_filter(conf, ctx)
    -- 把路由信息注入响应头，方便客户端和监控系统追踪
    if ngx.ctx.ai_route then
        ngx.header["X-AI-Route"] = ngx.ctx.ai_route.selected
        ngx.header["X-AI-Cost"] = tostring(ngx.ctx.ai_route.cost_per_1k)
    end
end

function find_route_by_name(conf, name)
    for _, r in ipairs(conf.routes) do
        if r.name == name then return r end
    end
    return nil
end

return _M

配置路由规则：

curl -X PUT http://localhost:9180/apisix/routes/ai-gateway \
  -H 'X-API-KEY: your-key' \
  -d '{
    "uri": "/v1/chat/completions",
    "methods": ["POST"],
    "plugins": {
      "ai-smart-router": {
        "routes": [
          {
            "name": "deepseek-local",
            "upstream_url": "http://192.168.1.101:8000/v1/chat/completions",
            "max_tokens": 2048,
            "cost_per_1k": 0.001,
            "timeout_ms": 60000
          },
          {
            "name": "gpt-cloud",
            "upstream_url": "https://api.openai.com/v1/chat/completions",
            "max_tokens": 128000,
            "cost_per_1k": 0.02,
            "timeout_ms": 120000
          }
        ],
        "fallback_route": "gpt-cloud"
      }
    },
    "plugins": {
      "key-auth": {}  # 同时启用 API Key 鉴权
    }
  }'

效果：
– 小请求（<2048 tokens）走本地 DeepSeek，成本极低
– 大请求自动切到云端 GPT，保证质量
– 响应头 X-AI-Route 标记实际走的哪条路，方便计费统计

---

六、实战插件 3 — 敏感数据脱敏

场景：API 返回的数据中有手机号、身份证号、银行卡号等敏感字段，需要在网关层统一脱敏，避免每个服务单独处理。

-- apisix/plugins/response-masking.lua
local core = require("apisix.core")
local cjson = require("cjson.safe")
local re_match = ngx.re.match

local schema = {
    type = "object",
    properties = {
        fields = {
            type = "object",
            additionalProperties = {
                type = "string",
                enum = {"phone", "idcard", "bank_card", "email", "custom"},
            },
        },
        mask_char = { type = "string", default = "*" },
        keep_prefix = { type = "integer", default = 3 },
        keep_suffix = { type = "integer", default = 4 },
    },
}

local _M = {
    version = 0.1,
    priority = 9000,  -- 较低优先级，在其他插件之后执行
    name = "response-masking",
    schema = schema,
    response_type = "json",  -- 只处理 JSON 响应
}

-- 脱敏函数库
local maskers = {
    phone = function(val, conf)
        -- 手机号：138****1234
        return string.sub(val, 1, 3) .. string.rep(conf.mask_char, 4) .. string.sub(val, -4)
    end,

    idcard = function(val, conf)
        -- 身份证：410***********1234
        return string.sub(val, 1, 6) .. string.rep(conf.mask_char, 8) .. string.sub(val, -4)
    end,

    bank_card = function(val, conf)
        -- 银行卡：6222****8888
        return string.sub(val, 1, 4) .. string.rep(conf.mask_char, #val - 8) .. string.sub(val, -4)
    end,

    email = function(val, conf)
        -- 邮箱：a***@example.com
        local at_pos = string.find(val, "@")
        if not at_pos then return val end
        return string.sub(val, 1, 1) .. string.rep(conf.mask_char, 3) .. string.sub(val, at_pos)
    end,

    custom = function(val, conf)
        -- 通用：保留前N后M位
        local len = string.len(val)
        if len <= conf.keep_prefix + conf.keep_suffix then return val end
        return string.sub(val, 1, conf.keep_prefix) 
            .. string.rep(conf.mask_char, len - conf.keep_prefix - conf.keep_suffix)
            .. string.sub(val, -(conf.keep_suffix))
    end,
}

-- 递归遍历 JSON 对象进行脱敏
function mask_json(obj, field_rules, conf)
    if type(obj) ~= "table" then return obj end

    for k, v in pairs(obj) do
        local mask_type = field_rules[k]

        if mask_type and type(v) == "string" and maskers[mask_type] then
            obj[k] = maskers[mask_type](v, conf)
        elseif type(v) == "table" then
            -- 数组或嵌套对象递归处理
            if #v > 0 then
                for i, item in ipairs(v) do
                    v[i] = mask_json(item, field_rules, conf)
                end
            else
                mask_json(v, field_rules, conf)
            end
        end
    end
    return obj
end

function _M.body_filter(conf, ctx)
    -- body_filter 可能被多次调用（chunked 编码），只在最后一块处理
    if not ctx.body_filtered then
        local body = core.response.hold_body_chunk(true, true)
        if body then
            local data = cjson.decode(body)
            if data then
                data = mask_json(data, conf.fields, conf)
                local masked_body = cjson.encode(data)
                ngx.arg[1] = masked_body
            else
                ngx.arg[1] = body
            end
            ctx.body_filtered = true
        end
    else
        ngx.arg[1] = nil  -- 清除后续 chunk
    end
end

return _M

使用方式：

# 对所有返回用户信息的接口开启脱敏
curl -X PUT http://localhost:9180/apisix/routes/user-api \
  -d '{
    "uri": "/user/*",
    "plugins": {
      "response-masking": {
        "fields": {
          "mobile": "phone",
          "id_number": "idcard",
          "bank_account": "bank_card",
          "email": "email",
          "real_name": "custom"   -- 用 custom 规则
        },
        "mask_char": "*"
      }
    }
  }'

原始响应 vs 脱敏后：

// 原始
{"name": "张三", "mobile": "13800138000", "id_number": "410102199001011234"}

// 脱敏后
{"name": "张三", "mobile": "138****8000", "id_number": "410***********1234"}

---

七、插件开发最佳实践

性能注意事项

要点	说明
避免同步 I/O	所有网络操作必须用 cosocket（非阻塞）
共享内存	用 ngx.shared.DICT 缓存热点数据
正则表达式预编译	用 ngx.re.compile() 预编译正则
JSON 库选择	生产环境用 cjson（比标准库快 10 倍+）
错误处理	不要让插件异常导致整个请求失败

测试你的插件

-- t/plugin/tiered-rate-limit.t
local t = require("lib.test_admin").test
local json = require("cjson")

t("tiered_rate_limit plugin", function()
    -- 测试正常请求通过
    local code, body = t("/test", "GET", {
        headers = {["X-User-Tier"] = "normal"}
    })
    assert(code == 200)

    -- TODO: 模拟高频请求测试限流触发
end)

make test  # 运行全部插件测试

---

总结

APISIX 插件开发的门槛比你想象的要低得多。从今天介绍的三个实战插件来看：

插件	适用场景	开发难度	核心价值
分级限流	SaaS 多租户	⭐⭐ 中等	精细化流量控制
AI 智能路由	多模型推理	⭐⭐⭐ 较高	成本优化 90%+
数据脱敏	金融/政务合规	⭐⭐ 中等	安全合规必备

掌握了这三个插件的模式，其他 80% 的业务需求都能举一反三。下次遇到”网关层要加个 XX 功能”，别再到处找现成的了，自己写一个可能更快。

---

👤 作者简介

一枚在大中原腹地（河南）卖公有云的从业者，主营腾讯云/阿里云/华为云，曾踩坑无数，现专注AI大模型应用落地。关注公众号「AI热点日报」，围观AI前沿动态~

博客：y1cloud.cn

---