实时输出最近一个小时内访问频率最高的10个IP

来源：go-interview / Case

要实时输出最近一个小时内访问频率最高的10个IP，可以使用以下技术与工具：

数据收集与处理：利用日志系统（如Nginx、Apache）或网络监听工具（如tcpdump）来收集IP访问日志。
窗口化数据处理：使用窗口化的数据流处理框架来实时计算和统计访问频率。
输出频率最高的IP：持续更新和输出最近一个小时的访问频率最高的10个IP。

方案1：使用Flink进行实时流处理

Apache Flink 是一个流处理框架，适用于实时计算任务。下面是一个简单的架构思路：

1. 配置数据源

首先，收集HTTP访问日志，可以从Nginx日志、Kafka队列等中读取数据。

2. 定义滑动窗口

使用Flink定义一个滑动窗口，每个窗口长度为1小时，滑动间隔为1分钟（或其他较小的间隔）。

3. 处理数据流

统计每个IP在滑动窗口中的出现次数。

4. 输出结果

在每个滑动窗口结束时，输出访问频率最高的10个IP。

import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.java.tuple.Tuple2
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.api.windowing.triggers.ProcessingTimeTrigger
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
 
fun main(args: Array<String>) {
    val env = StreamExecutionEnvironment.getExecutionEnvironment()
 
    // 1. 从数据源读取访问日志（如Kafka、文件等）
    val accessLogStream: DataStream<String> = env.socketTextStream("localhost", 9000)
 
    // 2. 提取出IP地址
    val ipStream: DataStream<String> = accessLogStream.map { line -> extractIpFromLog(line) }
 
    // 3. 统计1小时滑动窗口内每个IP的访问次数
    val ipCountStream: DataStream<Tuple2<String, Long>> = ipStream
        .map { ip -> Tuple2(ip, 1L) }
        .keyBy(0)
        .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(1)))
        .aggregate(object : AggregateFunction<Tuple2<String, Long>, Long, Long> {
            override fun createAccumulator(): Long = 0
            override fun add(value: Tuple2<String, Long>, accumulator: Long): Long = accumulator + 1
            override fun getResult(accumulator: Long): Long = accumulator
            override fun merge(a: Long, b: Long): Long = a + b
        })
 
    // 4. 输出访问频率最高的10个IP
    ipCountStream
        .keyBy(1)
        .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(1)))
        .maxBy(1)
        .print()
 
    env.execute("Real-Time IP Access Counter")
}
 
fun extractIpFromLog(logLine: String): String {
    // 根据实际日志格式提取IP
    return logLine.split(" ")[0]
}

方案2：使用Redis进行实时计数

可以将每个IP的访问记录到Redis，并设置时间窗口进行定期清理。

1. 记录IP访问

在每次访问时，使用ZADD命令将IP记录到Redis的有序集合中，得分为当前时间戳。

2. 计算访问频率

每分钟执行一个Redis脚本，删除超过1小时的访问记录，并计算剩余记录中出现次数最多的前10个IP。

3. 输出结果

输出计算结果到日志或监控系统。

import redis
import time
 
r = redis.Redis()
 
def record_ip(ip):
    now = time.time()
    r.zadd('ip_access', {ip: now})
 
def get_top_ips():
    one_hour_ago = time.time() - 3600
    r.zremrangebyscore('ip_access', 0, one_hour_ago)
    return r.zrevrange('ip_access', 0, 9, withscores=True)
 
while True:
    top_ips = get_top_ips()
    print("Top 10 IPs: ", top_ips)
    time.sleep(60)

方案3：使用ELK Stack（Elasticsearch, Logstash, Kibana）

使用ELK堆栈对日志进行收集、分析和展示。通过设置Kibana的时间窗口和聚合查询，可以实时展示过去1小时内访问频率最高的10个IP。

Logstash：收集并解析日志，将IP和时间戳等信息导入Elasticsearch。
Elasticsearch：存储日志并实时计算IP的访问频率。
Kibana：通过设置仪表盘，实时展示最近一小时内访问最多的IP。

方案4：自定义脚本（适合轻量级场景）

可以编写一个自定义脚本，读取日志文件，统计每个IP在1小时内的访问次数，并实时输出结果。

tail -f access.log | awk '{print $1}' | while read ip; do
    now=$(date +%s)
    echo "$ip $now" >> ip_access.log
    # 删除超过1小时的记录
    awk -v limit=$(date -d '-1 hour' +%s) '$2 >= limit' ip_access.log > temp.log && mv temp.log ip_access.log
    # 统计每个IP的访问次数
    awk '{print $1}' ip_access.log | sort | uniq -c | sort -nr | head -n 10
done

总结

不同方案适合不同的场景，Flink 适合处理高并发大数据流，Redis 适合中等规模的实时计数，ELK 是成熟的日志分析解决方案，而自定义脚本则适合小规模、轻量级的需求。根据实际场景选择合适的实现方案。

探索

Akiyama's blog