پیاده سازی RabbitMQ
نویسنده: زمانی فرهاد
تاریخ: ۱۴۰۰/۰۲/۱۰ ۲۱:۰
آدرس: www.dntips.ir
| مطالب | ۳۶۹۴ |
| نویسندگان | ۲۷۶ |
| گروههای مطالب | ۱۰۲۴ |
| نقشههای راه | ۱۱۹ |
| دورهها | ۱۴ |
| اشتراکها | ۱۷۹۱۴ |
public class RabbitMqConfiguration
{
public string HostName { get; set; }
public string Username { get; set; }
public string Password { get; set; }
} public interface IRabbitMqService
{
IConnection CreateChannel();
}
public class RabbitMqService : IRabbitMqService
{
private readonly RabbitMqConfiguration _configuration;
public RabbitMqService(IOptions<RabbitMqConfiguration> options)
{
_configuration = options.Value;
}
public IConnection CreateChannel()
{
ConnectionFactory connection = new ConnectionFactory()
{
UserName = _configuration.Username,
Password = _configuration.Password,
HostName = _configuration.HostName
};
connection.DispatchConsumersAsync = true;
var channel = connection.CreateConnection();
return channel;
}
} public static class StartupExtension
{
public static void AddCommonService(this IServiceCollection services, IConfiguration configuration)
{
services.Configure<RabbitMqConfiguration>(a => configuration.GetSection(nameof(RabbitMqConfiguration)).Bind(a));
services.AddSingleton<IRabbitMqService, RabbitMqService>();
}
} <ItemGroup>
<PackageReference Include="Microsoft.Extensions.Configuration.Abstractions" Version="5.0.0" />
<PackageReference Include="Microsoft.Extensions.Configuration.Binder" Version="5.0.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="5.0.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="5.0.0" />
<PackageReference Include="Microsoft.Extensions.Options" Version="5.0.0" />
<PackageReference Include="RabbitMQ.Client" Version="6.2.1" />
</ItemGroup> public interface IConsumerService
{
Task ReadMessgaes();
}
public class ConsumerService : IConsumerService, IDisposable
{
private readonly IModel _model;
private readonly IConnection _connection;
public ConsumerService(IRabbitMqService rabbitMqService)
{
_connection = rabbitMqService.CreateChannel();
_model = _connection.CreateModel();
_model.QueueDeclare(_queueName, durable: true, exclusive: false, autoDelete: false);
_model.ExchangeDeclare("UserExchange", ExchangeType.Fanout, durable: true, autoDelete: false);
_model.QueueBind(_queueName, "UserExchange", string.Empty);
}
const string _queueName = "User";
public async Task ReadMessgaes()
{
var consumer = new AsyncEventingBasicConsumer(_model);
consumer.Received += async (ch, ea) =>
{
var body = ea.Body.ToArray();
var text = System.Text.Encoding.UTF8.GetString(body);
Console.WriteLine(text);
await Task.CompletedTask;
_model.BasicAck(ea.DeliveryTag, false);
};
_model.BasicConsume(_queueName, false, consumer);
await Task.CompletedTask;
}
public void Dispose()
{
if (_model.IsOpen)
_model.Close();
if (_connection.IsOpen)
_connection.Close();
}
} public class ConsumerHostedService : BackgroundService
{
private readonly IConsumerService _consumerService;
public ConsumerHostedService(IConsumerService consumerService)
{
_consumerService = consumerService;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
await _consumerService.ReadMessgaes();
}
} public class Startup
{
public Startup(IConfiguration configuration)
{
Configuration = configuration;
}
public IConfiguration Configuration { get; set; }
public void ConfigureServices(IServiceCollection services)
{
services.AddCommonService(Configuration);
services.AddSingleton<IConsumerService, ConsumerService>();
services.AddHostedService<ConsumerHostedService>();
}
public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
{
}
} [Route("api/[controller]/[action]")]
[ApiController]
public class RabbitController : ControllerBase
{
private readonly IRabbitMqService _rabbitMqService;
public HomeController(IRabbitMqService rabbitMqService)
{
_rabbitMqService = rabbitMqService;
}
[HttpPost]
public IActionResult SendMessage()
{
using var connection = _rabbitMqService.CreateChannel();
using var model = connection.CreateModel();
var body = Encoding.UTF8.GetBytes("Hi");
model.BasicPublish("UserExchange",
string.Empty,
basicProperties: null,
body: body);
return Ok();
}
} public class Startup
{
public Startup(IConfiguration configuration)
{
Configuration = configuration;
}
public IConfiguration Configuration { get; set; }
public void ConfigureServices(IServiceCollection services)
{
services.AddControllers();
services.AddCommonService(Configuration);
}
public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
{
app.UseRouting();
app.UseEndpoints(endpoints =>
{
endpoints.MapDefaultControllerRoute();
});
}
} {
"RabbitMqConfiguration": {
"HostName": "localhost",
"Username": "guest",
"Password": "guest"
}
} version: "3.2"
services:
rabbitmq:
image: rabbitmq:3-management-alpine
container_name: 'rabbitmq'
ports:
- 5672:5672
- 15672:15672 using var connection = _rabbitMqService.CreateChannel();
using var model = connection.CreateModel();
IBasicProperties props = model.CreateBasicProperties();
props.DeliveryMode = 2;
var body = Encoding.UTF8.GetBytes("Hi");
model.BasicPublish("UserExchange",
string.Empty,
basicProperties: props,
body: body); _model.BasicQos(0, 16, true);