OpenResty与Kafka消息过滤实现方法

OpenResty 到底是什么？简单来说，它是一个基于 Nginx 和 Lua 的高性能 Web 平台，自带一整套模块和工具，能让你轻松实现各种复杂功能。比如，怎么用 OpenResty 给 Kafka 消息做过滤？这事儿听起来挺技术，但一步步拆开看，其实很清晰。

第一步：先把环境搭起来

前提是你已经装好了 OpenResty。接着，需要搞一个叫 LuaRocks 的工具，专门用来管理 Lua 库。命令并不复杂，一行一行跑就行：

wget https://luarocks.org/installers/luarocks-3.7.0-1.src.tar.gz
tar xzvf luarocks-3.7.0-1.src.tar.gz
cd luarocks-3.7.0-1
./configure
make
sudo make install

装好 LuaRocks 后，就可以用它来装 Kafka 模块了：

luarocks install kafka

第二步：写个过滤消息的 Lua 脚本

新建一个文件，比如叫 kafka_filter.lua，然后往里填代码。这里核心要做两件事：一是创建 Kafka 消费者，二是定义过滤规则。举个例子：

local kafka = require("resty.kafka")

-- Kafka 配置
local consumer_config = {
    "bootstrap.servers" = "localhost:9092",
    "group.id" = "my_group",
    "auto.offset.reset" = "earliest"
}

-- 创建 Kafka 消费者
local consumer, err = kafka.new_consumer(consumer_config)
if not consumer then
    ngx.log(ngx.ERR, "Failed to create Kafka consumer: ", err)
    return
end

-- 消息过滤函数
local function filter_message(message)
    -- 在这里添加你的过滤逻辑，例如只接受主题为 "my_topic" 的消息
    if message.topic == "my_topic" then
        return true
    else
        return false
    end
end

-- 消费消息
consumer:subscribe({"my_topic"})
local ok, err = consumer:consume(function(message)
    if not ok then
        ngx.log(ngx.ERR, "Failed to consume message: ", err)
        return
    end
    if filter_message(message) then
        -- 如果消息满足过滤条件，处理消息
        ngx.log(ngx.INFO, "Received and filtered message: ", message)
    else
        ngx.log(ngx.INFO, "Received but filtered message: ", message)
    end
end)

if not ok then
    ngx.log(ngx.ERR, "Failed to start consuming messages: ", err)
end

这段代码的思路很直接：先创建消费者，订阅 my_topic 主题，然后每收到一条消息就用 filter_message 函数判断一下。符合条件就记录“收到并处理”，不符合就记录“收到但过滤掉”。实际的过滤逻辑你完全可以按需调整——比如按消息内容、按时间戳或者按其他元数据。

第三步：把脚本挂到 OpenResty 上跑起来

需要修改 Nginx 配置文件，让 OpenResty 知道去哪找你的 Lua 文件。在 http 块里加上这样一行：

http {
    ...
    lua_package_path "/path/to/your/project/.lua;;";
    ...
}

把 /path/to/your/project/ 换成你实际存放脚本的目录。搞定之后，重启 OpenResty 或重新加载配置，它就会自动消费 Kafka 主题的消息，并按你写的过滤逻辑进行筛选。

整个过程没什么黑科技，关键就是把环境装对、脚本写清楚、路径配准确。以后想改过滤条件，直接在 filter_message 函数里动手就可以了，灵活得很。