redis频道收到发布者信息后可以只推信息给一个订阅者吗

Posted 2023-05-16

Redis 的 SUBSCRIBE 命令可以让客户端订阅任意数量的频道， 每当有新信息发送到被订阅的频道时， 信息就会被发送给所有订阅指定频道的客户端。
作为例子， 下图展示了频道 channel1 ， 以及订阅这个频道的三个客户端 —— client2 、 client5 和 client1 之间的关系：

参考技术A 那可以采用生产者消费者模式，而不是发布订阅模式

Redis 发布/订阅模式

一.命令简介

1.PSUBSCRIBE 订阅一个或多个符合给定模式的频道。
2.PUBLISH 将信息 message 发送到指定的频道 channel 。
3.PUBSUB 是一个查看订阅与发布系统状态的内省命令， 它由数个不同格式的子命令组成
4.PUNSUBSCRIBE 指示客户端退订所有给定模式。
5.SUBSCRIBE 订阅给定的一个或多个频道的信息。
6.UNSUBSCRIBE 指示客户端退订给定的频道。

二.例子

1.订阅msg

2.发送信息

三.代码实现

using Newtonsoft.Json;
using StackExchange.Redis;
using System;

namespace ResdisPubSub.PubSub
{
    /// <summary>
    /// 通过redis实现的订阅-发布机制
    /// </summary>
    public class RedisSubscribe : ISubscribeService
    {
        //链接
        static ConnectionMultiplexer redis;

        static RedisSubscribe()
        {
            ConfigurationOptions config = new ConfigurationOptions()
            {
                AbortOnConnectFail = false,
                ConnectRetry = 10,
                ConnectTimeout = 5000,
                ResolveDns = true,
                SyncTimeout = 5000,
                EndPoints = { { "127.0.0.1:6379" } },
                Password = "111111",
                AllowAdmin = true,
                KeepAlive = 180
            };
            redis = ConnectionMultiplexer.Connect(config);
        }

        /// <summary>
        /// 发布消息
        /// </summary>
        /// <typeparam name="T">消息类型</typeparam>
        /// <param name="channel">频道：消息的名称</param>
        /// <param name="msg">消息内容</param>
        /// <returns></returns>
        public void Publish<T>(string channel, T msg)
        {
            try
            {
                if (redis != null && redis.IsConnected)
                {
                    redis.GetSubscriber().Publish(channel, JsonConvert.SerializeObject(msg));
                }
            }
            catch (InvalidOperationException ex)
            {
                Console.WriteLine("redis服务错误，详细信息：" + ex.Message + "，来源：" + ex.Source);
            }
   
        }

        /// <summary>
        /// 订阅消息
        /// </summary>
        /// <param name="subChannel">频道：消息的名称</param>
        /// <param name="action">收到消息后的处理</param>
        public void Subscribe(string subChannel, Action<string> action)
        {
            try
            {
                if (redis != null && redis.IsConnected)
                {
                    redis.GetSubscriber().Subscribe(subChannel, (channel, message) => { action(message); });
                }
            }
            catch (InvalidOperationException ex)
            {
                Console.WriteLine("redis服务错误，详细信息：" + ex.Message + "，来源：" + ex.Source);
            }

        }

        /// <summary>
        /// 取消订阅
        /// </summary>
        /// <param name="channel">频道：消息的名称</param>
        public void Unsubscribe(string channel)
        {
            try
            {
                if (redis != null && redis.IsConnected)
                {
                    redis.GetSubscriber().Unsubscribe(channel);
                }
            }
            catch (InvalidOperationException ex)
            {
                Console.WriteLine("redis服务错误，详细信息：" + ex.Message + "，来源：" + ex.Source);
            }

        }

        /// <summary>
        /// 取消全部订阅
        /// </summary>
        public void UnsubscribeAll()
        {
            try
            {
                if (redis != null && redis.IsConnected)
                {
                    redis.GetSubscriber().UnsubscribeAll();
                }
            }
            catch (InvalidOperationException ex)
            {
                Console.WriteLine("redis服务错误，详细信息：" + ex.Message + "，来源：" + ex.Source);
            }
   
        }
    }
}

 

    class Program
    {
        static ISubscribeService client = new RedisSubscribe();
        static void Main(string[] args)
        {
            client.Subscribe("bigbigChannel", m => { Console.WriteLine($"我是bigbigChannel，接收到信息：{m}"); });
            Thread t = new Thread(Run);
            t.Start();
        }

        static void Run()
        {
            for (int i = 0; i < 100; i++)
            {
                Thread.Sleep(1000);
                client.Publish("bigbigChannel", i.ToString());
            }
       
        }
    }

 

 源码下载：https://github.com/lgxlsm/ResdisPubSub