Publish/Subscribe

Question

我正在尝试在没有 Redis 的 Node 中实现 publish/subscribe 模式。功能应该是一样的；您可以发布到频道，订阅频道并收听数据（如果您已订阅）；这是 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
});

下面的方法我都试过了，可能根本不是什么好方法（我只是展示一下我的尝试！）；我有自己的通道对象，其中保存了我们将新数据发布到特定通道时需要触发的通道和功能；如您所见，我正在为频道订阅功能；但是，我不知道有没有什么办法可以像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

您可以为每个频道使用 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);
  }
}

Publish/Subscribe

Publish/Subscribe

javascript

publish-subscribe

node.js

eventemitter