如何在Ruby中优化大量IP网络对象的IP地址匹配性能

问题背景 #

现有Fluentd IP过滤插件采用线性遍历（O(n)）匹配子网，面对2万+子网条目、每秒2000+日志的场景，性能瓶颈明显。需通过预加载的高效数据结构替代线性遍历，提升匹配速度。

优化方案：基于前缀树（Trie）的IP子网匹配 #

利用IPv4地址的32位二进制特性，构建前缀树存储子网信息。每个IP匹配仅需遍历32位（固定时间O(32)），大幅降低匹配耗时。

核心思路 #

1. 将子网转换为二进制前缀，按高位到低位构建树结构
2. 每个树节点仅在子网的最后一位位置存储对应元数据（国家、site_id）
3. IP匹配时，从高位到低位遍历树，记录最长匹配的子网元数据（确保最精确的匹配结果）

修改后的插件代码 #

require "fluent/plugin/filter"
require "ipaddr"
require "socket"

module Fluent
  module Plugin
    class IpaddressFilter < Fluent::Plugin::Filter
      Fluent::Plugin.register_filter("ipaddress", self)

      config_param :ipaddress_file_path, :string
      config_param :source_address_field, :string, :default => "src_ip"
      config_param :destination_address_field, :string, :default => "dest_ip"

      def initialize
        super
        # 前缀树根节点，子节点用0/1表示二进制位
        @trie_root = { children: {}, metadata: nil }
      end

      def configure(conf)
        super
        ipaddress_data = JSON.parse(File.read(@ipaddress_file_path))

        ipaddress_data.each do |entry|
          cidr = entry["network"]
          ip_addr = IPAddr.new(cidr, Socket::AF_INET)
          # 将子网转换为整数和前缀长度
          ip_int = ip_addr.to_i
          prefix_len = ip_addr.prefix

          current_node = @trie_root
          # 从高位到低位遍历前缀的每一位
          prefix_len.times do |i|
            # 取当前位的值（0或1）
            bit = (ip_int >> (31 - i)) & 1
            current_node[:children][bit] ||= { children: {}, metadata: nil }
            current_node = current_node[:children][bit]
          end
          # 在子网的最后一位节点存储元数据
          current_node[:metadata] = {
            country: entry["country"],
            site_id: entry["site_id"]
          }
        end
      end

      def filter(tag, time, record)
        # 处理源IP
        src_ip = record[@source_address_field]
        if src_ip
          metadata = find_matching_metadata(src_ip)
          if metadata
            record["src_country"] = metadata[:country]
            record["src_site_id"] = metadata[:site_id]
          end
        end

        # 处理目标IP
        dest_ip = record[@destination_address_field]
        if dest_ip
          metadata = find_matching_metadata(dest_ip)
          if metadata
            record["dest_country"] = metadata[:country]
            record["dest_site_id"] = metadata[:site_id]
          end
        end

        record
      end

      private

      def find_matching_metadata(ip_str)
        ip_int = IPAddr.new(ip_str, Socket::AF_INET).to_i
        current_node = @trie_root
        matched_metadata = nil

        32.times do |i|
          bit = (ip_int >> (31 - i)) & 1
          break unless current_node[:children].key?(bit)

          current_node = current_node[:children][bit]
          # 记录当前匹配到的元数据（最长前缀优先）
          matched_metadata = current_node[:metadata] if current_node[:metadata]
        end

        matched_metadata
      end
    end
  end
end

优化说明 #

• 预构建前缀树：在configure阶段一次性将所有子网构建为前缀树，避免每次匹配重复遍历子网列表
• 固定时间匹配：每个IP匹配仅需32次位运算和树节点查找，时间复杂度为O(1)（固定步数），相比原O(n)线性遍历，在2万+子网场景下性能提升显著
• 最长前缀匹配：匹配过程中记录所有命中的子网元数据，最终返回最长前缀的结果，确保最精确的子网匹配（解决子网包含的场景）

内容的提问来源于stack exchange，提问作者Daniel