Redis-Message

Install

npm install wm-redis-message --save

Usage

const RedisMessage = require('wm-redis-message').RedisMessage;

const cache = new RedisMessgae(options);

console.log(await cache.getMessages(10));

Opitons

| name | marks | default | | -------------------- | ------------------------------- | ------- | | topic | topic | | | messageType | messageType | | | redis | redis object | | | fetchMessage | fetchMessage 方法 | | | afterFetchMessage | fetchMessage 之后调用的方法 | | | dealfailedMessage | 处理超过 max 失败次数的 Message | | | maxAckTimeout | 最长 acks 时间，单位 ms | 60000 | | eachMessageCount | 每次获取 messageCount 的数量 | 200 | | minRedisMessageCount | redis 维持的最小数量的 message | 200 | | maxRetryTimes | 消息消费最多失败次数 | 5 | | lockExpireTime | 锁超时时间，单位 s | 60 |

const {
    topic, // topic
    messageType, // messageType
    redis, // redis
    keyHeader = 'msg_', // redis key header
    maxAckTimeout = 60 * 1000, // 消费超时时间 
    eachMessageCount = 200, // 每次 Message 获取数量
    minRedisMessageCount = 200, // Redis 最少的 count 数量
    maxRetryTimes = 5, // 消息消费失败重新消费次数
    lockExpireTime = 60, // Lock timeout

    fetchMessage = function (options) {
        // const { topic, messageType } = options;
        return async function () {
            return [];
        }
    },
    afterFetchMessage = function (options) {
        // const { topic, messageType } = options;
        return async function (data = {}) {
            const {
                offset
            } = data;
            debug(`message has been offset to :${offset}`);
        }
    },
    dealfailedMessage = function (options) {
        // const { topic, messageType } = options;
        return async function (messageId, detail) {
            debug(`messageId: ${messageId} has been error acks`);
            console.log(`message failed!! id: ${messageId} data: ${JSON.stringify(detail)}`);
            return;
        }
    }
} = options;