RxJS 是使用 Observables 的响应式编程的库，在异步编程中被广泛使用，在Node中也不例外，在TypeScript中使用RxJS编写相对复杂的响应式代码尤其便捷。

Redis 即Remote Dictionary Server（远程字典服务），属于内存高速缓存数据库，通过发布、订阅机制实现高效数据交换，被广泛应用于：数据缓存，消息事件和高速队列等场景。

那么我们如何与Redis进行交互呢？在网上查了一些代码，大都是对Redis的直接调用，而我们在真实应用场景中肯定是需要通过RxJS进行封装的，一方面是使得我们的应用代码与Redis解耦（以后若换成其他的消息中间件，也无需更改调用方代码），另一方面也可以在封装后的代码中增加自己想要的功能（比如日志、权限或队列处理等）。

代码功能说明

使用rxjs与redis进行交互，通过对producer的重载，实现对Redis的封装，在最外层调用封装后的代码，以实现调用方与Redis的解耦，最外层代码只需要使用标准的Observable对象即可。

示例代码

完整的示例代码如下：
https://gitee.com/JasonLiSz/typescript-rxjs-redis.git

核心代码如下：

subscribe(): Observable<string> {
    const self = this
    const producer = (subscriber: Subscriber<string>) => {
        self.client.subscribe(self.channel)

        self.client.on("message", (channel: string, message: string) => {
            const messageSentToClient = `received message from channel: ${channel}, message: ${message}`
            subscriber.next(messageSentToClient)
        })

        self.client.on("unsubscribe", (channe: string, count: number) => {
            subscriber.complete()
        })
    }

    return new Observable<string>(producer)
}

通过 npm run test 运行测试，得到如下结果。

:> npm run test
> npx mocha -r ts-node/register -r tsconfig-paths/register --timeout 100000 ./spec/**/*.spec.ts
  test suite - redis rxjs communication
    ✔ test 1
publishing message: message:0, by channel:internal-channel-1
published  message: message:0, by channel:internal-channel-1
publishing message: message:1, by channel:internal-channel-1
published  message: message:1, by channel:internal-channel-1
publishing message: message:2, by channel:internal-channel-1
published  message: message:2, by channel:internal-channel-1
publishing message: message:3, by channel:internal-channel-1
published  message: message:3, by channel:internal-channel-1
publishing message: message:4, by channel:internal-channel-1
published  message: message:4, by channel:internal-channel-1
received message from channel: internal-channel-1, message: message:0
received message from channel: internal-channel-1, message: message:1
received message from channel: internal-channel-1, message: message:2
received message from channel: internal-channel-1, message: message:3
received message from channel: internal-channel-1, message: message:4
publishing message: message:5, by channel:internal-channel-1
published  message: message:5, by channel:internal-channel-1
received message from channel: internal-channel-1, message: message:5
subscribe completed from channel: internal-channel-1

RxJS库

https://www.npmjs.com/package/rxjs

Redis环境准备

Linux环境

https://redis.io/

Windows环境

https://github.com/tporadowski/redis/releases