[大数据] Redis 之 subscribe 订阅模式封装

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 大数据 -> Redis 之 subscribe 订阅模式封装 -> 正文阅读

[大数据]Redis 之 subscribe 订阅模式封装

subscribe 订阅模式封装

Redis 里的订阅/发布命令

命令	用例和描述
subscribe	subscribe channel [channel …] 订阅一个或多个频道
unsubscribe	unsubscribe [channel [channel …]] 退订频道，如果没有指定频道，则退订所有的频道
publish	publish channel message 给指定的频道发消息
psubscribe	psubscribe pattern [pattern …] 订阅给定模式相匹配的所有频道
punsubscribe	punsubscribe [pattern [pattern …]] 退订给定的模式，如果没有指定模式，则退订所有模式

由于 Redis 在有 subscribe操作时，就会进入订阅模式，此时就不会有publish 功能，因此这个封装进行单一化，只做事件的监听功能。也不做数据处理操作。

在取消所有订阅操作后，则会进入正常模式。

废话不多说直接上代码

// RedisSub.js

let EventEmitter = require('events');
let util = require('util');
const ioredis = require("ioredis");

/**
 * Redis 订阅端
 * @constructor
 */
let RedisSub = function (redisConfig ){
    EventEmitter.call(this);

    // log开关 true 开启 false 关闭
    this.LOGGER = false;
    
    this.redisC = new ioredis({
        port: redisConfig.port,
        host: redisConfig.host,
        password: redisConfig.password,
        db: redisConfig.db || 0,
        family: 4, // 4 (IPv4) or 6 (IPv6)
    });

    // 监听订阅成功事件
    this.redisC.on("subscribe", (channel, count) => {
        this.LOGGER && console.info("client subscribed to " + channel + "," + count + " total subscriptions");
    });
    // 收到消息后执行回调，message是redis发布的消息
    this.redisC.on("message", (channel, message) => {
        this.LOGGER && console.log(`Received ${message} from ${channel}`);
        this.emit(channel, message);
    });
    // 监听取消订阅事件
    this.redisC.on("unsubscribe", (channel, count) => {
        this.LOGGER && console.info(`client subscribed unsubscribed from ${channel}, ${count} subscriptions`)
    });
    
}
util.inherits(RedisSub, EventEmitter);

RedisSub.prototype.name = '__redisSubscribeClient__';

/**
 * 订阅频道
 */
RedisSub.prototype.subscribe = function (...channels) {
    let self = this;
    channels.forEach( function (channel) {
        self.redisC.subscribe(`${channel}`, function (err, count) {
            if (err) {
                console.error("Failed to channel: %s subscribe: %s", channel, err.message);
            } else {
                self.LOGGER && console.log(`Subscribed channel ${channel} successfully!`);
                self.LOGGER && console.log(`This client is currently subscribed to ${count} total channels.`);
            }
        });
    });
}


/**
 * 取消订阅频道
 */
RedisSub.prototype.unsubscribe = function (...channels) {
    let self = this;
    _.forEach(channels, function (channel){
        self.LOGGER && console.info("subscribe channel: %j", channel);
        self.redisC.unsubscribe(`${channel}`);
    });
}

/**
 * 创建 RedisSub
 * @param redisConfig 
 * @returns {RedisSub}
 */
module.exports.init = function (redisConfig ) {
    return new RedisSub(redisConfig );
};

使用方式

测试用代码

// test.js

let ioredis = require('ioredis');
let RedisSub = require('./redisSub');
let opts = {
    host:'127.0.0.1',
    port: 6379,
    password: ''
}

// 创建订阅功能Redis
let subC = RedisSub.init(opts);
// 开启日志
subC.LOGGER=true;
subC.subscribe("test");
subC.on("test", function (msg) {
    console.log("==> ", msg)
});

let redisC = new ioredis({
    port: opts.port,
    host: opts.host,
    password: opts.password,
    db: opts.db || 0,
    family: 4, // 4 (IPv4) or 6 (IPv6)
});

// 每秒发送一次
setInterval(()=>{
    redisC.publish("test", "aaaa")
}, 1000);