May i subscribe multiple hosts (and clusters) with one subscriber pattern in Redis?

Question

I am implementing a feature which integrating redis published messages to mongodb that i made the project and working perfect on testing environment.

But I'm concerning about the production environment, that i have 3 master server and they exist 12 slave cluster. if i publish message from them to a channel pattern may i subscribe all messages in one place

Answer 1

Yes it is possible by stackexchange redis settings, i had done my general structure like below

 public class RedisSubscriber: IRedisSubscriber
    {
        private readonly RedisConfigurationManager _config;
        private readonly IMongoDbRepository _mongoDbRepository;
        private readonly ILogger<RedisSubscriber> _logger;
        private readonly IConnectionMultiplexer _connectionMultiplexer;


        public RedisSubscriber(IServiceProvider serviceLocator, ILogger<RedisSubscriber> logger, IConnectionMultiplexer conn)
        {
            _config = (RedisConfigurationManager)serviceLocator.GetService(typeof(RedisConfigurationManager));
            _mongoDbRepository = (IMongoDbRepository)serviceLocator.GetService(typeof(IMongoDbRepository));
            _connectionMultiplexer = conn;
            _logger = logger;
        }

        public void SubScribeChannel()
        {
            _logger.LogInformation("!SubScribeChannel started!!");

            string channelName = _config.ActiveChannelName;
            var pubSub = _connectionMultiplexer.GetSubscriber();
            try
            {
                pubSub.Subscribe(channelName, async (channel, message) => await MessageActionAsync(message, channel));
            }
            catch(Exception ex)
            {
                _logger.LogInformation(String.Format("!error: {0}", ex.Message));
            }
            Debug.WriteLine("EOF");
        }

        private async Task MessageActionAsync(RedisValue message, string channel)
        {
            try
            {
                Transformer t = new Transformer(_logger);
                _logger.LogInformation(String.Format("!SubScribeChannel message received on message!! channel: {0}, message: {1}", channel, message));
                string transformedMessage = Transformer.TransformJsonStringData2Message(message);
                List<Document> documents = Transformer.Deserialize<List<Document>>(transformedMessage);

                await MergeToMongoDb(documents, channel);
                _logger.LogInformation("!Merged");
            }
            catch (Exception ex)
            {
                _logger.LogInformation(String.Format("!error: {0}", ex.Message));
            }
        }

        private async Task MergeToMongoDb(IList<Document> documents, string channelName)
        {
            try
            {
                foreach (Document doc in documents)
                {
                    TurSysPartitionedDocument td = JsonConvert.DeserializeObject<TurSysPartitionedDocument>(JsonConvert.SerializeObject(doc));
                    td.DepartureDate = td.DepartureDate.ToLocalTime();
                    td.PartitionKey = channelName;
                    TurSysPartitionedDocument isExist = await _mongoDbRepository.GetOneAsync<TurSysPartitionedDocument>(k =>
                                k.ProductCode == td.ProductCode &&
                                k.ProviderCode == td.ProviderCode &&
                                k.CabinClassName == td.CabinClassName &&
                                k.OriginAirport == td.OriginAirport &&
                                k.DestinationAirport == td.DestinationAirport &&
                                k.Adult >= td.Adult &&
                                k.DepartureDate == td.DepartureDate,
                                td.PartitionKey);

                    if (isExist != null)
                    {
                        //_logger.LogInformation(String.Format("!isExist departure date: {0}", isExist.DepartureDate));
                        isExist.SearchCount++;
                        await _mongoDbRepository.UpdateOneAsync(isExist, k => k.Adult, td.Adult);
                        await _mongoDbRepository.UpdateOneAsync(isExist, k => k.SearchCount, isExist.SearchCount);
                    }
                    else
                    {
                        //_logger.LogInformation(String.Format("!last ToLocalTime td departure date: {0}", td.DepartureDate));
                        td.SearchCount = 1;
                        await _mongoDbRepository.AddOneAsync(td);
                        //_logger.LogInformation(String.Format("!last ToLocalTime result td departure date: {0}", td.DepartureDate));
                    }
                }
            }
            catch(Exception ex)
            {
                _logger.LogInformation(String.Format("!error: {0}", ex.Message));
            }
        }


    }