javascript - 发布/订阅

Question

我正在尝试在 Node 中实现发布/订阅模式，但不使用 Redis。功能应该是相同的；您可以发布到 channel ，订阅 channel 并收听数据(如果您已订阅)；以下是 Redis 功能:

pub.publish('channel1', myData1, function (err, data) {     
});


sub.subscribe('channel1', function (err, data) {
});


sub.on("message", function (channel, data) {
    //Now you are easily listening to data
});

我尝试了以下方法，这可能根本不是一个好方法(我只是展示我的尝试!)；我有自己的 channel 对象，其中保存了当我们将新数据发布到特定 channel 时需要触发的 channel 和函数；正如你所看到的，我正在订阅 channel 的功能；但是，我不知道是否有任何方法可以像redis提供的那样监听数据 sub.on("message",function(){...!所以请你帮忙如果我能像 Redis 那样做，但使用 eventEmitter 或...

var events = require('events');
var emitter = new events.EventEmitter;
var channels = {};
//channels ={"channelA": ["func1", "func2"], "channelB": []}

//publish data to a channel and emits all of the functions which are registered for listening to the channel
function publish(channel, data, callback) {
    if (channels[channel] && channels[channel].length > 0) {
        for (var i = 0; i < channels[channel].length; i++) {
            emitter.emit(channels[channel][i], data);
        }
    }
    callback();

}

//subscribe a function to a channel
function subscribe(channel, FuncName, next) {
    if (!channels[channel]) {
        channels[channel] = [];
    }
    channels[channel].push(FuncName);
    next();
}

//Subscribe a callback function from a channel
function unsubscribe(channel, FuncName) {
    if (channels[channel]) {
        var indx = channels[channel].indexOf(FuncName);
        if (indx != -1) {
            channels[channel].splice(indx, 1);
        }
    }
}

Answer 1

您可以为每个 channel 使用一个EventEmitter。另外，您可能希望保存对回调的引用，而不是对函数名称字符串的引用。

var events = require('events');
var channels = {};

//publish data to a channel and emits all of the functions which are registered for listening to the channel
function publish(channel, data, callback) {
  if (channels[channel]) {
    channels[channel].emit('message', channel, data);
  }
  callback();
}

//subscribe a function to a channel
function subscribe(channel, fn, next) {
  if (!channels[channel]) {
    channels[channel] = new events.EventEmitter();
  }
  channels[channel].addListener('message', fn);
  next();
}

//unsubscribe a callback function from a channel
function unsubscribe(channel, fn) {
  if (channels[channel]) {
    channels[channel].removeListener('message', fn);
  }
}