bobolink

___.         ___.          .__  .__        __
\_ |__   ____\_ |__   ____ |  | |__| ____ |  | __
 | __ \ /  _ \| __ \ /  _ \|  | |  |/    \|  |/ /
 | \_\ (  <_> ) \_\ (  <_> )  |_|  |   |  \    <
 |___  /\____/|___  /\____/|____/__|___|  /__|_ \
     \/           \/                    \/     \/

Introduction

Lightweight JS task scheduling tool that allows for concurrency control, timeout, retry, error fetching, running status statistics, etc. It also supports multiple scheduling modes, including immediately scheduling and frequency scheduling.

点击查看中文文档

Application scenario

1. You want to control the number of concurrent tasks, such as large-scale interface requests, control the frequency will reduce the impact on the interface provider, but also reduce the load on the system, the CPU performance is smoother.

2. Hope that the task can be retried when it fails. Compared to self-implementation, bobolink can provide a more diverse and easy-to-use retry strategy.

3. I hope that the task can provide the ability to cancel the timeout, and also provide the execution indicators of each task, including the waiting time in the queue, the running time, the execution status, and so on.

4. I want to control the total number of asynchronous tasks. After reaching a threshold, bobolink will reject or discard the old tasks, so that the task will not be uncontrollable.

Summary: Bobolink can be understood as a thread pool, all asynchronous tasks can be scheduled through bobolink, with bobolink, asynchronous tasks are no longer barbaric, but completely controllable.

Install

npm i bobolink

Quick Start

Basic Usage

const Bobolink = require('bobolink');

// create a new queue instance
const queue = new Bobolink();

// submit a functional task
queue.put(() => {
    // The task must return a Promise
    return Promise.resolve(true);
}).then(ts => {
    console.log('task execution completed');
    // undefined
    console.log('whether succeed: ' + ts.err === undefined);
    // true
    console.log('return value: ' + ts.res);
    console.log('waiting time: ' + ts.waitingTime);
    console.log('run time: ' + ts.runTime);
    // 0
    console.log('number of retries: ' + ts.retry);
});

// submit a task that failed to execute
queue.put(() => {
    return Promise.reject('error');
}).then(ts => {
    console.log('task execution completed');
    // error
    console.log('error: ' + ts.err);
});

When initializing an instance, you are allowed to configure the properties of the queue:

// set the number of concurrency to 1 to get a serial queue
const queue = new Bobolink({
    concurrency: 1
});

If the task is detected to have a problem, an exception will be thrown immediately:

// submit a task group with no elements
// or submit a task group with existing elements, but none of the elements meet the current task model
queue.put([]).catch(err => {
    assert.equal(err, Bobolink.EMPTY_ELEMENTS);
});

Scheduled by frequency

const Bobolink = require('bobolink');
const assert = require('assert');

// create a new queue instance to enable the frequency scheduling mode
const queue = new Bobolink({
    // constants are directly mounted under Bobolink
    scheduleMode: Bobolink.SCHEDULE_MODE_FREQUENCY,
    // set two tasks per second, bobolink calculates a task every 500ms
    countPerSecond: 2
});

// calculate the number of task schedules by this variable
let scheduleCount = 0;

// one task is scheduled every 500ms, and only two tasks should be scheduled after 1200ms.
let t1 = new Promise(resolve => {
    setTimeout(() => {
        assert.equal(scheduleCount, 2);
        resolve();
    }, 1200);
});

// generate task function
function task() {
    return () => {
        return new Promise(resolve => {
            setTimeout(() => {
                scheduleCount++;
                resolve();
            }, 5);
        });
    }
}

// put three tasks into the queue
queue.push([task(), task()， task()]).then(ts => {
    console.log('The total time spent on this group of tasks:' + ts.runTime);
    // the return value of this group of tasks, res is an array, each element contains the execution status of the corresponding task
    assert.equal(ts.res.length, 3);
    // the first two tasks are executed normally, but the third task is cancelled because the queue is destroyed.
    assert.equal(ts.res[2].err, Bobolink.DISCARD);
});

t1.then(() => {
    // after the execution is completed, the resources occupied by the queue can be destroyed if necessary.
    queue.destory();
});

Data mode

const Bobolink = require('bobolink');
const assert = require('assert');

// create a new queue instance, providing a handler function
const queue = new Bobolink({
    handler: data => {
        assert.equal(data, true);
        return !data;
    }
});

// putting a non-functional task for the first time will automatically be inferred as a data mode
queue.push(true).then(ts => {
    assert.equal(ts.res, false);
});

Combination mode

const Bobolink = require('bobolink');
const assert = require('assert');

// I want to perform a set of cleanups every minute, each group will carry a number of IDs that need to be cleaned up.
// each ID cleanup takes 1 minute (mainly IO time), and at most 2 IDs are cleaned at the same time.
let clearMap = {
    group1: [1, 2, 3],
    group2: [5, 6, 7, 8]
}

// ID cleanup queue
let clearQueue = new Bobolink({
    // clean up 2 IDs at the same time
    concurrency: 2,
    // cleanup handler
    handler: id => {
        return new Promise(resolve => {
            setTimeout(resolve, 1000 * 60);
        });
    }
});

// process one group per minute
let intervalQueue = new Bobolink({
    timeScale: 60,
    countPerTimeScale: 1,
    handler: groupId => {
        // submit the list of IDs under this group to the cleanup queue
        return clearQueue.put(clearMap[groupId]);
    }
});

intervalQueue(['group1', 'group2']).then(ts => {
    console.log('all tasks are completed and the total time is spent:' + ts.runTime);
});

Supported configuration items

Configuration	Description	Value
concurrency	Concurrent number	Greater than 0, the default is 5
timeout	Task timeout (ms)	Greater than or equal to 0, set to 0 does not time out, the default is 15000
retry	Number of failed retries	Greater than or equal to 0, set to 0 will not retry, the default is 0
retryPrior	Whether to perform the retry task first	Default is false
newPrior	Whether to prioritize new tasks	The default is false, does not support modification
catch	Crawl the task's exception, retry multiple times and call multiple catches	Need to provide a function, the default is null
max	The maximum number of tasks that can be submitted to the queue and are awaiting execution	-1 is no upper limit, the default is 65536
scheduleMode	Scheduling mode, support for frequency scheduling and immediately scheduling	Bobolink.SCHEDULE_MODE_FREQUENCY and Bobolink.SCHEDULE_MODE_IMMEDIATELY (default), oes not support modification
countPerTimeScale	Number of tasks scheduled per time scale when scheduled by frequency	Greater than 0, default is 100, set to -1 to schedule all tasks at a time, does not support modification
timeScale	Time scale (s)	Greater than or equal to 1, default is 1s
taskMode	Task mode, ie put, the type of task belongs to function or data	Bobolink.TASK_MODE_DATA and Bobolink.TASK_MODE_FUNCTION, If it is not set, it will be automatically inferred and does not support modification.
handler	When the task is in data mode, you need to provide the handler function.	A function that returns a Promise
saturationPolicy	Policy for submitting tasks when the queue is saturated	Bobolink.SATURATION_POLICY_ABORT (abort, default) and Bobolink.SATURATION_POLICY_DISCARD_OLDEST (discard the oldest task)

Task execution status

Field	Description
err	Exception information of the task, when the task is executed successfully, err is equal to undefined
res	The return value of the task. When a single task is submitted, res is the return value of the task. When a group of tasks is submitted, res is an array, and each element corresponds to the execution status of each task.
runTime	Task execution time, when a group of tasks is submitted, the execution time of the entire group of tasks
waitingTime	The waiting time of the task in the queue
retry	Number of task retries

API brief

Bobolink

Bobolink is the class used to generate the execution queue and needs to borrow the new keyword.

grammar:

new Bobolink(options)

[option]

For configuration items, refer to Supported Configuration Items for detailed configuration.

Bobolink mounts some string constants, such as the identifier of the task cancellation: Bobolink.DISCARD, which is preferred.

Bobolink.prototype.put(tasks, prior)

Alias function: Bobolink.prototype.push(tasks, prior)。

Submit the task to the queue instance for scheduling.

[tasks]

The submitted tasks can be single or a group. The task type can be either data or a function.

[prior]

Whether the task submitted this time is prioritized.

Returning to Promise, when the task itself is detected to be abnormal or not accepted (not entering the execution flow), an error is thrown and a catch is required. Once the submission is successful, the execution status of the task can be obtained in the then process regardless of whether the task is executed successfully.

Bobolink.prototype.setOption(options)

Update the configuration of the queue instance. Except for the configuration items that do not support the change during the runtime, other configuration items can be updated at will, and take effect immediately.

Bobolink.prototype.runningTasksCount

Get the number of tasks in action.

Bobolink.prototype.queueTaskSize

Gets the number of tasks waiting in the queue.

Bobolink.prototype.options

Get the configuration of the current instance.

Bobolink.prototype.destory()

Clean up the current queue instance. This includes removing the timing scheduler (if it exists) and cleaning up the remaining unexecuted tasks (all return cancellations).

LICENSE

MIT