2个nodejs进程利用redis 实现订阅发布

news2024/11/20 9:47:32

1.新建文件 redis_db.js

'use strict';

const redis = require('redis');
const options = {
    host: "127.0.0.1",
    port: 6379,
    password: "123456", // CONFIG SET requirepass "123456"
}

var array = []
for(var i=0; i<3; i++){
    const client = redis.createClient(options)
    array.push(client)
}

function getDB(index){
    if(typeof index === "number"){
        return array[index]
    }
    return array
}

for(var key in array){
    const client = array[key]
    client.on('error', err => console.log('------ client Redis connection failed ------' + err))
        .on('connect', () => console.log('------ client Redis connection succeed ------')
    )
}

module.exports = {
    getDB,
}

备注1：安装reids啥的就不说了

备注2：新增3个redis的连接: [0] 是之前的; [1] 用于sub; [2] 用于pub

如果没有多个的话，sub和pub的时候会报错

ReplyError: ERR only (P)SUBSCRIBE / (P)UNSUBSCRIBE / QUIT allowed in this context

2.新建文件 redis_ipc.js

var redis_db = require("./redis_db.js");
const clients  = redis_db.getDB()
const client1  = clients[1]
const client2  = clients[2]

function mySub(channelName, handleMessage){
    // 订阅指定的频道
    client1.subscribe(channelName, (err, channels) => {
        if (err) {
            console.error('无法订阅频道：', err);
        } else {
            // 设置每次收到新消息时调用的处理函数
            client1.on('message', handleMessage);
        }
    })
}

function myPub(channelName, messageContent){
    client2.publish(channelName, messageContent, (err, reply) => {
        if (!err && reply === 0) {
            console.log(`消息 "${messageContent}" 未被任何人接收。`);
        } else if (!err && reply > 0) {
            console.log(`消息 "${messageContent}" 已成功发送给 ${reply} 个订阅者。`);
        } else {
            console.error('发送消息失败：', err);
        }
    });
}

module.exports = {
    mySub,
    myPub,
}

3.新增测试代码

3.1新建文件test01.js 测试同个进程的


const redis_ipc = require('./redis_ipc.js')
function testSubPub(){
    const channelName = 'test_channel_1';
    function handleMessage(channel, message) {
        // 在这里编写处理收到消息后的操作
        console.log(`收到来自 ${channel} 通道的消息：${message}`);
    }
    redis_ipc.mySub(channelName, handleMessage)

    const messageContent = '{"data":{"name":"xxx","age":"18"}}';
    redis_ipc.myPub(channelName, messageContent)
}

testSubPub()

目录如下:

执行指令：node .\test01.js

3.2新建文件test02.js test03.js 测试同个进程的

// test02.js
const redis_ipc = require('./redis_ipc.js')
function testSubPub(){
    const channelName = 'test_channel_1';
    function handleMessage(channel, message) {
        // 在这里编写处理收到消息后的操作
        console.log(`收到来自 ${channel} 通道的消息：${message}`);
    }
    redis_ipc.mySub(channelName, handleMessage)

    // const messageContent = '{"data":{"name":"xxx","age":"18"}}';
    // redis_ipc.myPub(channelName, messageContent)
}

testSubPub()

test03.js
const redis_ipc = require('./redis_ipc.js')
function testSubPub(){
    const channelName = 'test_channel_1';
    // function handleMessage(channel, message) {
    //     // 在这里编写处理收到消息后的操作
    //     console.log(`收到来自 ${channel} 通道的消息：${message}`);
    // }
    // redis_ipc.mySub(channelName, handleMessage)

    const messageContent = '{"data":{"name":"xxx","age":"18"}}';
    redis_ipc.myPub(channelName, messageContent)
}

testSubPub()

订阅者：先在一个终端执行 node .\test02.js