如何使用线程逐行读取文件？

Question

我一直在尝试优化使用 node 读取非常大的文件并开始遇到瓶颈，所以我想我会尝试使用 node 中的线程。

我正在尝试读取 2 个文件

包含 1000 - 100,000 个 uuid 的针我正在尝试查找我正在读取同步并拆分文件。

事先不知道行数。

包含数亿行 uuid 大小为 3.4gb+ 的 haystack 我想按读取量拆分它并拆分文件的读取。

// example needle.txt/haystack.txt
a57a258d-7e56-40e1-962e-d683a17d7d3b
4380b26b-36d3-4cfe-ae3b-9121bf1f0c12
ebda3a08-857f-41d2-99bf-a288f0470af8

process.argv[3] 是针，process.argv[4] 是草垛，我也可以接受来自process.stdin 的草垛

我正在努力解决的是如何在保留文件行的同时处理拆分文件，这可能吗？

#!/usr/bin/env node
const fs = require('fs');

const cluster = require('cluster');
const { createInterface } = require('readline');

if (cluster.isMaster) {
    const fileSize = fs.statSync(process.argv[4]).size;

    const threads = 4;

    const n = new Set(
        fs
            .readFileSync(process.argv[3])
            .toString()
            .split('\n')
    );

    for (var i = 0; i < threads; i++) {
        const buffer = {};
        const worker = cluster.fork();

        worker.on('message', function(lines) {
          // main thread received buffer from thread
          /*
             how to combine the data so i can read the lines

             if (n.has(line)) {
               console.log(line)   
             }
          */
        });

        worker.on('exit', threadNum => {
          // exited thread all parts read from that thread
        })

        worker.send({start: i * (fileSize / threads), end: (i + 1) * (fileSize / threads), i});
    }
} else {
    process.on('message', function({ start, end, i }) {
        /*
           worker code here we start streaming from one part of the file 
           to another
        */
        createInterface({
            input: fs.createReadStream(process.argv[4], {
                start,
                end
            })
        })
        .on('line', data => {
          // sending data back to main thread with thread number
          process.send({data, thread: i});
        })
        .on('close', () => process.exit(i))
    });
}

是否可以告诉 readline.createInterface 继续阅读直到它遇到一个字符？我可以通过传递线程号并在主线程中拆分来组合缓冲区吗？

任何帮助将不胜感激。

如果有人想知道我正在运行 cmd

node index.js --needles needles.txt haystack.txt

Answer 1

在下面试试这个，基本上是加载needles文件，然后将所有搜索词发送到大海捞针。

const cluster = require('cluster');

if (cluster.isMaster) {
    console.log("Master started.");
    // Only need filesystem access in the master node and readline in the master node
    const fs = require('fs');
    const rl = require('readline');

    // Number of cpus is the number of threads, we'll read 1000 lines before
    // sending the work to a worker to process and well set up a roundRobin style
    // counter to send data to each thread as equally as we can
    const numCPUs = require('os').cpus().length;
    const numOfLines = 1000;
    let lineNum = 0;
    let lines = [];
    let roundRobin = 1;
    let foundItems = {};

    let needles = [];

    // Hold all our workers and teh number of calls to the worker
    let workers = [];

    // Handle your messages this is the worker handler
    function messageHandler(message) {
        // set the roundRobin only when the thread is complete
        roundRobin = (roundRobin+1)%(numCPUs);

        // a found message
        for (let items in message.found) {
            // add the array of lineno to the collective found items
            foundItems[items] = foundItems[items].concat(message.found[items]);
        }
    };

    // Log to console all the found items at the end
    function logFoundItems(){
        console.log(foundItems);
    };

    // Start workers
    for (let i = 0; i < numCPUs; i++) {
        workers.push(cluster.fork());
    }

    // When worker sends back a message handle it
    for (const id in workers) {
        workers[id].on('message', messageHandler);
    }

    // First file of uuids to search for
    const searchFile = rl.createInterface({
        input: fs.createReadStream(process.argv[3]),
        crlfDelay: Infinity
    });

    let readLine = null; 

    // Load the search lines into an array and set up the collective search item arrays
    searchFile.on('line', (ln)=>{
        let item = ln.trim();
        needles.push(item);
        foundItems[item.toLowerCase()] = [];
    });

    // Whole search file is loaded start streaming the haystack file
    searchFile.on('close', ()=>{
        console.log('Search File Loaded... Starting processing.');
        // Start reading lines from the file stream
        readLine = rl.createInterface({
            input: fs.createReadStream(process.argv[4]),
            crlfDelay: Infinity
        });

        // When a line is recieved from realLine push it to an array and update the count
        // when the number of lines reaches a defined chunk of data to process pause the
        // readline send the chunk with it's line number and the search term then resume
        // the roundRobin handles the worker assignments
        readLine.on('line', (line) => {
            ++lineNum;
            lines.push(line);
            if (lineNum%numOfLines === 0) {
                readLine.pause();
                workers[roundRobin].send({
                    start: lineNum, 
                    lines: lines.slice(0), 
                    search: needles
                });
                lines = [];
                readLine.resume();
            }
        });

        // When the end of the file is reached this is where you can exit the program if you want
        readLine.on('close', () => {
            console.log('Done processing the file');
            setTimeout(()=>{
                logFoundItems();
                process.exit(0);
            }, 1000);

        });

    });

} else if (cluster.isWorker) {
    console.log("Worker", cluster.worker.id, "Started.");

    // When the worker recieves a message process the message
    process.on('message', (message)=>{
        let start = Number(message.start);
        let lines = message.lines;
        let search = message.search;
        let foundItems = {};

        // Iterate over the search terms and check the lines send back a 
        // message for each term found and which line
        search.forEach((uuid)=>{
            lines.forEach((line, ind)=>{
                let lineText = line.toLowerCase();
                let searchTerm = uuid.toLowerCase();
                if (lineText.search(searchTerm) > -1) {
                    if (!foundItems.hasOwnProperty(searchTerm)) {
                        foundItems[searchTerm] = [];
                    }
                    foundItems[searchTerm].push(start+ind);
                }
            });
        });

        // Send the message back to the master node
        process.send({
            type: 'found', 
            found: foundItems, 
            workerID: cluster.worker.id
        });

    });
}

Answer 2

我评论了代码以便更好地解释正在发生的事情。主要思想是将您正在使用的任何数据传递给工作人员，因为工作人员看不到主服务器中初始化的任何内容。我可能会将 master 部分和 worker 部分放入他们自己的文件中以保持整洁。

const cluster = require('cluster');

if (cluster.isMaster) {
    console.log("Master started.");
    // Only need filesystem access in the master node and readline in the master node
    const fs = require('fs');
    const rl = require('readline');

    // Number of cpus is the number of threads, we'll read 1000 lines before
    // sending the work to a worker to process and well set up a roundRobin style
    // counter to send data to each thread as equally as we can
    const numCPUs = require('os').cpus().length;
    const numOfLines = 1000;
    let lineNum = 0;
    let lines = [];
    let roundRobin = 0;

    // Hold all our workers and teh number of calls to the worker
    let workers = [];
    let usage = {};

    // Handle your messages this is the worker handler
    function messageHandler(message) {
        // increment the number of calls to the specific worker
        if (usage.hasOwnProperty(message.worker)) {
            usage[message.worker]++;
        } else {
            usage[message.worker] = 1;
        }

        // set the next worker
        roundRobin = (roundRobin+1)%(numCPUs);

        // if the text was found in the line return the line numbers
        if (message.found) {
            console.log("Line Numbers:", message.lineNo);
        }

    }

    // Start workers
    for (let i = 0; i < numCPUs; i++) {
        workers.push(cluster.fork());
    }

    // When worker sends back a message handle it
    for (const id in workers) {
        workers[id].on('message', messageHandler);
    }

    // Start reading lines from the file stream
    const readLine = rl.createInterface({
        input: fs.createReadStream(process.argv[4]),
        crlfDelay: Infinity
    });

    // When a line is recieved from realLine push it to an array and update the count
    // when the number of lines reaches a defined chunk of data to process pause the
    // readline send the chunk with it's line number and the search term then resume
    // the roundRobin handles the worker assignments
    readLine.on('line', (line) => {
        //console.log('Line:', line);
        ++lineNum;
        lines.push(line);
        if (lineNum%numOfLines === 0) {
            readLine.pause();
            workers[roundRobin].send({
                start: lineNum, 
                lines: lines.slice(0), 
                search: process.argv[3]
            });
            lines = [];
            readLine.resume();
        }
    });

    // When the end of the file is reached thsi is where you can exit the program if you want
    readLine.on('close', () => {
        console.log('Done processing the file');
    });

} else if (cluster.isWorker) {
    console.log("Worker Started:", cluster.worker.id);

    // When the worker recieves a message process the message
    process.on('message', (message)=>{
        let start = Number(message.start);
        let lines = message.lines;
        let search = message.search.trim();

        // Initialize a swicth to determine if we found the data or not which
        // line numbers we found the text on
        let found = false;
        let lineNo = [];

        // Iterate over the lines passed in and search for the search term in the line
        // if it's found add the line number to the array and set found to true
        // the start variable hold the start of this chunks line number
        lines.forEach(function(val, ind) {
            let text = val.toLowerCase();
            if (text.search(search.toLowerCase()) !== -1) {
                lineNo.push(start + ind);
                found = true;
            }
        });

        // Send back the data to master
        process.send({found: found, lineNo: lineNo, search: search, start: start, worker: cluster.worker.id});
    });
}