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 函数即可，灵活性极高。