寻找出现次数最多的100个IP

问题及背景

首先，该问题出现在一场面试中，详细描述如下：有一个10G的IP日志，每行为一条IP地址，JVM堆内存为1G，现在要求统计出该日志中出现次数最多的100个IP地址。面试时想出了思路，并且也给出了相应的代码，但文件读取的方法用的不多，代码写的不完善，这里给出详细的思路和代码。

思路

要统计IP出现的次数，那么很容易想到用HashMap来计数，用ip地址作key，出现次数作为value。然而考虑到日志文件大小为10G，堆内存为1G，那么显然没办法把读取整个IP日志然后统计。所以必然是要做将这个大的日志文件进行切分的，比如可以切分成100个小文件，对每个小文件使用HashMap来统计出现次数最多的前100个ip，至此我们获取到了10000个ip，然后再对这10000个ip进行排序找频次最高的100个即可，这里可以使用优先队列PriorityQueue来存储。
而这里需要注意的是，为了防止某个ip被分散在多个小文件中，并且在该小文件中出现的次数不是前100，但最后总次数可能是前100而被抛弃的情况，我们可以通过计数ip的hash值后对文件数量取模的方式来得出该ip需要放入的具体文件。

具体实现

综合上述思路，该功能的实现需要三步，1、划分小文件，2、统计小文件中出现次数最多的ip，3、从小文件中出现次数最多的ip中寻找出现总次数最多的top100。
代码如下：

package Interview;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;
import java.util.stream.Collectors;

public class IPCount {

    private static final int NUMBER_OF_FILES = 100; // 划分的小文件数量
    private static final int topN = 100;			// 需要统计的次数最多的前多少个


    // 切分小文件
    public static void splitLargeFile(String inputFilePath) throws IOException {
        // 初始化写入器数组
        BufferedWriter[] writers = new BufferedWriter[NUMBER_OF_FILES];
        for (int i = 0; i < NUMBER_OF_FILES; i++) {
            writers[i] = Files.newBufferedWriter(Paths.get(inputFilePath + "_part_" + i + ".txt"));
        }

        // 读取大文件并分割
        try (BufferedReader reader = Files.newBufferedReader(Paths.get(inputFilePath))) {
            String line;
            while ((line = reader.readLine()) != null) {
                int fileIndex = Math.abs(line.hashCode() % NUMBER_OF_FILES);
                writers[fileIndex].write(line);
                writers[fileIndex].newLine();
            }
        }

        // 关闭所有写入器
        for (BufferedWriter writer : writers) {
            writer.close();
        }
    }

    // 统计小文件中的topN
    public static Map<String, Integer> countIPsInFile(String fileName) throws IOException {
        Map<String, Integer> ipCount = new HashMap<>();

        try (BufferedReader reader = Files.newBufferedReader(Paths.get(fileName))) {
            String line;
            while ((line = reader.readLine()) != null) {
                ipCount.put(line, ipCount.getOrDefault(line, 0) + 1);
            }

        }

        return ipCount;
    }

    
    public static List<Map.Entry<String, Integer>> findTopIPs(Map<String, Integer> ipMap) {

        PriorityQueue<Map.Entry<String, Integer>> ipQueue = new PriorityQueue<>((o1, o2) -> (int) o1.getValue() - (int) o2.getValue());

        for (Map.Entry<String, Integer> entry : ipMap.entrySet()) {
            ipQueue.offer(entry);
            if (ipQueue.size() > topN) ipQueue.poll();
        }

        return new ArrayList<>(ipQueue);
    }


    // 合并最后结果
    public static List<Map.Entry<String, Integer>> mergeTopIps(List<List<Map.Entry<String, Integer>>> topIpList) {
        PriorityQueue<Map.Entry<String, Integer>> ipQueue = new PriorityQueue<>((o1, o2) -> (int) o1.getValue() - (int) o2.getValue());

        for (List<Map.Entry<String, Integer>> list : topIpList) {
            for (Map.Entry<String, Integer> ipEntry : list) {
                ipQueue.offer(ipEntry);
                if (ipQueue.size() > topN) {
                    ipQueue.poll();
                }
            }
        }

        return new ArrayList<>(ipQueue);
    }

    // 将结果写入文件
    public static void saveTopIP(List<Map.Entry<String, Integer>> ipList) throws IOException {
        BufferedWriter writer = new BufferedWriter(Files.newBufferedWriter(Paths.get("top_100_ip.txt")));
        for (Map.Entry ip : ipList) {
            System.out.println(ip.getKey() + "\t" + ip.getValue());
            writer.write(ip.getKey() + "\t" + ip.getValue());
        }
    }

    public static void main(String[] args) throws IOException {
        String largeFile = "ip_log.txt";
        splitLargeFile(largeFile);

        List<List<Map.Entry<String, Integer>>> partialTopIPList = new ArrayList<>();

        for (int i = 0; i < NUMBER_OF_FILES; i++) {
            Map<String, Integer> counts = countIPsInFile(largeFile + "_part_" + i + ".txt");
            partialTopIPList.add(findTopIPs(counts));
            counts.clear();
        }

        List<Map.Entry<String, Integer>> topIpList = mergeTopIps(partialTopIPList);
        Collections.sort(topIpList, (o1, o2) -> {
            return (int) o2.getValue() - (int) o1.getValue();
        });
        saveTopIP(topIpList);
    }
}

测试

上述代码已经在本地进行过测试，可以正常运行。这里提供一段批量生成ip地址的Python代码，感兴趣的读者可以据此生成测试样例自行测试。

import random

def generate_random_ip():
    ip = ".".join(str(random.randint(0, 255)) for _ in range(4))
    return ip

if __name__ == "__main__":
    num_ips = 1000000000
    output_file = "random_ips.txt"

    with open(output_file, "w") as f:
        for _ in range(num_ips):
            random_ip = generate_random_ip()
            f.write(random_ip + "\n")

    print(f"{num_ips} random IP addresses generated and saved in {output_file}")