Efficient Microservices Communication: Redis Pub/Sub Message Broker Implementation Guide
In a microservices architecture, a message broker is an essential component that enables seamless data transfer between different services. Popular options like RabbitMQ and Kafka are commonly used to facilitate this communication. Another robust tool that can be utilized as a message broker is Redis. Apart from being a widely used distributed cache, Redis can function as a primary database or a message broker, as discussed in a previous article (1). This article explores how Redis can serve as a powerful message broker using its Pub/Sub pattern, enhancing the efficiency of inter-service communication in a microservices architecture. While Redis offers multiple patterns like Redis Stream and Redis Message Queue for data transfer between services, our focus here will be on the Pub/Sub pattern.
Redis Pub/Sub simplifies real-time communication between microservices, providing a straightforward yet robust mechanism for publishing and subscribing to messages. Future articles will delve into the other Redis patterns, but for now, let’s delve into how Redis Pub/Sub can streamline microservices communication.
What Pub/Sub pattern?is?
Based on Microsoft’s explanation:
Enable an application to announce events to multiple interested consumers asynchronously, without coupling the senders to the receivers. (2)
So, the process of sending messages between services is like the below diagram.
Based on the above explanation, Redis pub/sub is a messaging pattern where senders(publishers) send messages to the channels and subscribers receive messages from these channels. In Redis, this pattern is implemented using the PUBLISH, SUBSCRIBE, UNSUBSCRIBE, and PSUBSCRIBE commands.
Pros and Cons
Redis Pub/Sub, like any other architectural pattern, comes with its own set of advantages and disadvantages. Let's discuss them to better understand their suitability for various use cases:
?Advantages
Disadvantages
Prerequisites
Scenario
In this scenario, two services will be implemented. One is responsible for passing a message (Publish) and another service will receive (Subscribe) that message.
Step 1
In the first step, create 2?.NET API projects using the below command.
dotnet new webapi ProducerService
dotnet new webapi ConsumerService
Then create a class library that plays a message broker role using the below command.
dotnet new classlib -o RedisBroker
Now, add the above class library to both services.
dotnet add ProducerService reference RedisBroker/RedisBroker.csproj
dotnet add ConsumerService reference RedisBroker/RedisBroker.csproj
Now, add Redis packages inside the above class library using the Nuget package manager console or using?.NET CLI.
Install-Package StackExchange.Redis
dotnet add package StackExchange.Redis
Step 2
In this step, within the RedisBroker class library, create a new class named RedisMessageBroker.cs. Within this class, implement methods to serve as a publisher and subscriber, facilitating communication between our services. Below are the methods you should add to establish this functionality.
using StackExchange.Redis;
public class RedisMessageBroker
{
private readonly ConnectionMultiplexer _redis;
private readonly ISubscriber _subscriber;
public RedisMessageBroker(string connectionString)
{
_redis = ConnectionMultiplexer.Connect(connectionString);
_subscriber = _redis.GetSubscriber();
}
// Publish message
public async Task<bool> Publish(string channel, string message)
{
try
{
await _subscriber.PublishAsync(RedisChannel.Literal(channel),
message);
return true;
}
catch
{
return false;
}
}
// Subscribe message
public async Task<bool> Subscribe(string channel, Action<RedisChannel, RedisValue> handle)
{
try
{
await subscriber.SubscribeAsync(RedisChannel.Literal(channel), handle);
return true;
}
catch
{
return false;
}
}
}
As shown in the above code, 2 methods (PublishAsync/SubscribeAsync) have 2 parameters.
Channel: The channel name that passes data through it between services.
Messages (Publisher method): Contains the message which is going to pass
Handle (Subscriber method): Contains the value that comes through the channel which is the message.
Step 3
Create a Models directory in both ProducerService and ConsumerService projects, then add a new Message.cs class file inside each model's directory. Include the necessary properties in both Message.cs files for consistent communication between the projects.
public class Message
{
public int Id { get; set; }
public string Body { get; set; }
}
Step 4
Add a DBConfig directory to both projects’ root, then create a new RedisConfig.cs class file within each DBConfig directory. Include the necessary Redis configuration properties in both RedisConfig.cs files for consistent database configuration across the projects.
public class RedisConfig
{
public string RedisCacheURL { get; set; }
}
Now, inside the appsettings.json file of both projects, add the below connection string that is related to our Redis instance which is running through the Docker on the local machine.
领英推荐
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.AspNetCore": "Warning"
}
},
"ConnectionStrings": {
"RedisCacheURL": "127.0.0.1:6379"
},
"AllowedHosts": "*"
}
Now, register Redis insde the program.cs file in both projects.
using Microsoft.Extensions.Options;
using ProducerService.DBConfig; //inside ProducerService
using ConsumerService.DBConfig; //inside ConsumerService
using RedisBroker;
// Register Redis as a message broker
// Best practice: Use DI in the RedisBroker class library to have a clean- // code section inside the Program.cs file
builder.Services.Configure<RedisConfig>(builder.Configuration.GetSection("ConnectionStrings"));
builder.Services.AddSingleton<RedisMessageBroker>(sb =>
{
var config = sb.GetRequiredService<IOptions<RedisConfig>>().Value;
return new RedisMessageBroker(config.RedisCacheURL);
});
Step 5
It is time to generate a service to pass data to the Redis Pub/Sub. So, in the root directory of the ProducerService project, create a new directory called MessageService, and inside it, create a new interface called IMessageService.cs.
using ProducerService.Models;
public interface IMessageService
{
Task<Message> SendMessage(Message message);
}
Now, inside the same directory, create a new class called MessageService.cs.
using ProducerService.Models;
using RedisBroker;
using System.Text.Json;
public class MessageService : IMessageService
{
private readonly RedisMessageBroker _redisMessageBroker;
public MessageService(RedisMessageBroker redisMessageBroker)
{
_redisMessageBroker = redisMessageBroker;
}
public async Task<Message> SendMessage(Message message)
{
var res = await _redisMessageBroker.Publish("MessageChannel", JsonSerializer.Serialize(message));
if(res is true)
{
return message;
}
return null;
}
}
Note: This is the implementation of IMessageService and the interface will be injected into the controller to pass data between services using RedisMessageBroker that has been implemented inside the class library. So, to have more cleaner structure, it is better to use clean code architecture to maintain and manipulate everything easily.
Now, register this service, to program.cs file in the ProducerService project.
using ProducerService.MessageService;
// Register Message service
builder.Services.AddScoped<IMessageService, MessageService>();
Step 6
Here, a controller will be required to get data from the client and pass it to the Redis Publisher. So, in the root directory of the ProducerService project, inside the Controller directory, create a new controller called MessageController.cs and add the following action.
using Microsoft.AspNetCore.Mvc;
using ProducerService.MessageService;
using ProducerService.Models;
[Route("api/[controller]")]
[ApiController]
public class MessageController : ControllerBase
{
private readonly IMessageService _messageService;
private readonly ILogger<MessageController> _logger;
public MessageController(IMessageService messageService, ILogger<MessageController> logger)
{
_messageService = messageService;
_logger = logger;
}
[HttpPost]
public async Task<IActionResult> Post(Message message)
{
var res = await _messageService.SendMessage(message);
_logger.LogInformation($"{DateTime.Now.ToShortDateString()} : {res.Body}");
return Ok(res);
}
}
As shown in the above code, to keep everything straightforward, the base model which is Message.cs passes directly as an argument of the Post action. It is better to use a DTO object and a mapping function or a mapping library like AutoMapper, Mapster, etc. (3)
After passing data, it is time to receive the message inside the ConsumerService project.
Step 7
To receive incoming messages, create a new directory in the root directory of the ConsumerService project called Services, create a new class called ConsuerService.cs, and add the following method.
using ConsumerService.Models;
using StackExchange.Redis;
using System.Text.Json;
public class ConsumerServices
{
private readonly ILogger<ConsumerServices> _logger;
public ConsumerServices(ILogger<ConsumerServices> logger)
{
_logger = logger;
}
public void HandleMessage(RedisChannel redisChannel, RedisValue message)
{
var messageRes = JsonSerializer.Deserialize<Message>(message);
_logger.Log(LogLevel.Information, messageRes.Body);
}
}
Now, register ConsumerService,cs inside the Program.cs file.
using ConsumerService.Services;
builder.Services.AddSingleton<ConsumerServices>();
Step 8
Now, the channel should be subscribed by the ConsumerService project. So, inside the program.cs file, add the following code to listen to the channel to receive incoming messages.
var app = builder.Build();
var messageBroker = app.Services.GetRequiredService<RedisMessageBroker>();
var consumerService = app.Services.GetRequiredService<ConsumerServices>();
await messageBroker.Subscribe("MessageChannel", (channel, message) =>
{
consumerService.HandleMessage(channel, message);
});
Finally, it is time to test microservices and Redis pub/sub-broker. So, run the docker desktop and start the Redis instance.
Then, send a request to the API endpoint using the Postman.
Now, a message has been sent to the pub/sub-Redis broker.
As shown, in the above image, the message has existed inside the Redis channel.
As indicated, the message has been received and the log has been demonstrated on the API console.
Conclusion
Utilize this implementation as an effective means of implementing a broker for services that broadcast data, such as notification or alarm services. To ensure robustness and data consistency, consider integrating the SAGA pattern(4) and EventSourcing(5) in your services.
Github
YouTube
References
(4) https://blog.stackademic.com/implementation-of-saga-orchestration-using-masstransit-dd238530f0d7
Exciting possibilities ahead with Redis Pub/Sub. vahid alizadeh