任务分配模式

export function* generateTasks(searchHash, alphabet,
                               maxWordLength, batchSize) {
    let nVariations = 0
    for (let n = 1; n <= maxWordLength; n++) {
        nVariations += Math.pow(alphabet.length, n)
    }
    console.log('Finding the hashsum source string over ' +
        `${nVariations} possible variations`)
    let batchStart = 1
    while (batchStart <= nVariations) {
        const batchEnd = Math.min(
            batchStart + batchSize - 1, nVariations)
        yield {
            searchHash,
            alphabet: alphabet,
            batchStart,
            batchEnd
        }
        batchStart = batchEnd + 1
    }
}

import zmq from 'zeromq'
import delay from 'delay'
import {generateTasks} from './generateTasks.js'

const ALPHABET = 'abcdefghijklmnopqrstuvwxyz'
const BATCH_SIZE = 10000
const [, , maxLength, searchHash] = process.argv

async function main() {
    const ventilator = new zmq.Push() // (1)
    await ventilator.bind('tcp://*:5016')
    await delay(1000) // wait for all the workers to connect
    const generatorObj = generateTasks(searchHash, ALPHABET,
        maxLength, BATCH_SIZE)
    for (const task of generatorObj) {
        await ventilator.send(JSON.stringify(task)) // (2)
    }
}

main().catch(err => console.error(err))

import isv from 'indexed-string-variation'
import {createHash} from 'crypto'

export function processTask(task) {
    const variationGen = isv.generator(task.alphabet)
    console.log('Processing from ' +
        `${variationGen(task.batchStart)} (${task.batchStart}) ` +
        `to ${variationGen(task.batchEnd)} (${task.batchEnd})`)
    for (let idx = task.batchStart; idx <= task.batchEnd; idx++) {
        const word = variationGen(idx)
        const shasum = createHash('sha1')
        shasum.update(word)
        const digest = shasum.digest('hex')
        if (digest === task.searchHash) {
            return word
        }
    }
}

import zmq from 'zeromq'
import {processTask} from './processTask.js'

async function main() {
    const fromVentilator = new zmq.Pull()
    const toSink = new zmq.Push()
    fromVentilator.connect('tcp://localhost:5016')
    toSink.connect('tcp://localhost:5017')
    for await (const rawMessage of fromVentilator) {
        const found = processTask(JSON.parse(rawMessage.toString()))
        if (found) {
            console.log(`Found! => ${found}`)
            await toSink.send(`Found: ${found}`)
        }
    }
}

main().catch(err => console.error(err))

import zmq from 'zeromq'

async function main() {
    const sink = new zmq.Pull()
    await sink.bind('tcp://*:5017')
    for await (const rawMessage of sink) {
        console.log('Message from worker: ', rawMessage.toString())
    }
}

main().catch(err => console.error(err))

node worker.js
node worker.js
node collector.js

node producer.js 4 f8e966d1e207d02c44511a58dccff2f5429e9a3b

import amqp from 'amqplib'
import {generateTasks} from './generateTasks.js'

const ALPHABET = 'abcdefghijklmnopqrstuvwxyz'
const BATCH_SIZE = 10000
const [, , maxLength, searchHash] = process.argv

async function main() {
    const connection = await amqp.connect('amqp://localhost')
    const channel = await connection.createConfirmChannel() // (1)
    await channel.assertQueue('tasks_queue')
    const generatorObj = generateTasks(searchHash, ALPHABET,
        maxLength, BATCH_SIZE)
    for (const task of generatorObj) {
        channel.sendToQueue('tasks_queue', // (2)
            Buffer.from(JSON.stringify(task)))
    }
    await channel.waitForConfirms()
    channel.close()
    connection.close()
}

main().catch(err => console.error(err))

import amqp from 'amqplib'
import {processTask} from './processTask.js'

async function main() {
    const connection = await amqp.connect('amqp://localhost')
    const channel = await connection.createChannel()
    const {queue} = await channel.assertQueue('tasks_queue')
    channel.consume(queue, async (rawMessage) => {
        const found = processTask(
            JSON.parse(rawMessage.content.toString()))
        if (found) {
            console.log(`Found! => ${found}`)
            await channel.sendToQueue('results_queue',
                Buffer.from(`Found: ${found}`))
        }
        await channel.ack(rawMessage)
    })
}

main().catch(err => console.error(err))

import amqp from 'amqplib'

async function main() {
    const connection = await amqp.connect('amqp://localhost')
    const channel = await connection.createChannel()
    const {queue} = await channel.assertQueue('results_queue')
    channel.consume(queue, msg => {
        console.log(`Message from worker: ${msg.content.toString()}`)
    })
}

main().catch(err => console.error(err))

node worker.js
node worker.js

node collector.js
node producer.js 4 f8e966d1e207d02c44511a58dccff2f5429e9a3b

import Redis from 'ioredis'
import {generateTasks} from './generateTasks.js'

const ALPHABET = 'abcdefghijklmnopqrstuvwxyz'
const BATCH_SIZE = 10000
const redisClient = new Redis()
const [, , maxLength, searchHash] = process.argv

async function main() {
    const generatorObj = generateTasks(searchHash, ALPHABET,
        maxLength, BATCH_SIZE)
    for (const task of generatorObj) {
        await redisClient.xadd('tasks_stream', '*',
            'task', JSON.stringify(task))
    }
    redisClient.disconnect()
}

main().catch(err => console.error(err))

import Redis from 'ioredis'
import {processTask} from './processTask.js'

const redisClient = new Redis()
const [, , consumerName] = process.argv

async function main() {
    await redisClient.xgroup('CREATE', 'tasks_stream', // (1)
        'workers_group', '$', 'MKSTREAM')
        .catch(() => console.log('Consumer group already exists'))
    const [[, records]] = await redisClient.xreadgroup( // (2)
        'GROUP', 'workers_group', consumerName, 'STREAMS',
        'tasks_stream', '0')
    for (const [recordId, [, rawTask]] of records) {
        await processAndAck(recordId, rawTask)
    }
    while (true) {
        const [[, records]] = await redisClient.xreadgroup( // (3)
            'GROUP', 'workers_group', consumerName, 'BLOCK', '0',
            'COUNT', '1', 'STREAMS', 'tasks_stream', '>')
        for (const [recordId, [, rawTask]] of records) {
            await processAndAck(recordId, rawTask)
        }
    }
}

async function processAndAck(recordId, rawTask) { // (4)
    const found = processTask(JSON.parse(rawTask))
    if (found) {
        console.log(`Found! => ${found}`)
        await redisClient.xadd('results_stream', '*', 'result',
            `Found: ${found}`)
    }
    await redisClient.xack('tasks_stream', 'workers_group', recordId)
}

main().catch(err => console.error(err))

node worker.js workerA
node worker.js workerB

node collector.js
node producer.js 4 f8e966d1e207d02c44511a58dccff2f5429e9a3b