本文将向您展示如何在 C# 应用程序中使用 Rabbit MQ。我将解释如何使用 WEB API、Rabbit MQ 如何广播以及客户端应用程序如何使用广播消息。
如果您喜欢此文章，请收藏、点赞、评论，谢谢，祝您快乐每一天。
什么是 Rabbit MQ 及其功能？
RabbitMQ 是一款开源消息代理软件，可促进分布式应用程序各个组件之间的通信和数据交换。RabbitMQ 充当消息队列，管理从发送者（生产者）到接收者（消费者）的消息路由、排队和传递。
解耦应用程序： RabbitMQ 可用于解耦应用程序，这意味着它们可以相互通信，而无需成为一个整体。 实现异步通信： RabbitMQ 可用于实现异步通信，这意味着消息的发送和接收不会阻塞发送者或接收者。 实时流： RabbitMQ 可用于实现实时流，这意味着消息是实时传递的。 负载平衡： RabbitMQ 可用于在不同的消费者之间平衡消息负载。 故障转移： RabbitMQ 可以为应用程序提供故障转移功能。这意味着，如果一个 RabbitMQ 节点发生故障，另一个节点可以接管并继续传递消息。 审计： RabbitMQ 可用于审计消息。这意味着您可以跟踪消息的发送者和接收者，以及消息的发送和接收时间。 监控：可以监控 RabbitMQ 以跟踪其性能和健康状况。
RabbitMQ 的优势
多平台通信意味着消息以 JSON 等常用语言进行序列化/反序列化。 对于这个非常庞大的社区来说，开源可以开发和改进这些问题。 灵活、可靠、可扩展。 支持多种协议（STOMP、MQTT、RPC、HTTPS、HTTP 等）。
RabbitMQ 安装
由于我使用的是 Windows 操作系统，因此请下载适用于 Windows 的 Rabbit MQ。
RabbitMQ下载链接及安装过程
默认配置
端点地址： http://localhost:15672/ 登录：访客 密码： guest Windows 服务名称： RabbitMQ 如果未启动，则启动服务的 Windows 菜单项： RabbitMQ 服务 – 启动
创建 Dotnet core webapi 并广播消息（使用最新的 .Net 8）。
工具： Visual Studio 2022 检查最新的 dotnet 版本 打开 CMD 并输入： dotnet –version 查看所有版本： dotnet –list-SDKs
步骤1.添加新的dotnet webapi项目。
步骤2.选择框架。
步骤3.创建项目，并请检查并确认该项目是.net 8框架。
开始编码
添加NewFolder调用模型，并添加名为Product.cs的产品模型类。
namespace RabbitMQProductApi.Models { public class Product { public int ProductID { get; set; } public string ProductName { get; set; }=string.Empty; public string ProductDescription { get; set; }= string.Empty; public int ProductPrice { get; set; } public int ProductStock { get; set; } } }
添加NewFolder调用RabbitMQ，添加Rabbitmq接口名称同IRabbitMQProducer.cs。
namespace RabbitMQProductApi.RabbitMQ { public interface IRabbitMQProducer { public void SendProductMessage<T>(T message); } }
添加一个新的 RabbitMQProducer 类并使用 IRabbitMQProducer 接口。请从nuget.org安装最新的RabbitMQ.Client。
using RabbitMQ.Client;
namespace RabbitMQProductApi.RabbitMQ { public class RabbitMQProducer : IRabbitMQProducer { public void SendProductMessage<T>(T message) { var factory=new ConnectionFactory() { HostName = "localhost" }; var connection = factory.CreateConnection(); var channel = connection.CreateModel(); channel.QueueDeclare(queue: "ProductQueue", durable: false, exclusive: false, autoDelete: false, arguments: null); var json= System.Text.Json.JsonSerializer.Serialize(message); var body = System.Text.Encoding.UTF8.GetBytes(json); channel.BasicPublish(exchange:"", routingKey: "ProductQueue", basicProperties: null, body: body); } } }
添加一个名为 Data 的新文件夹（用于数据库目的），请确保您已使用 nuget.org 为该项目安装了实体框架。
using Microsoft.EntityFrameworkCore; using RabbitMQProductApi.Models;
namespace RabbitMQProductApi.Data { public class DBContextClass:DbContext { protected readonly IConfiguration Configuration; public DBContextClass(IConfiguration configuration) { Configuration = configuration; } protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) { optionsBuilder.UseSqlServer("Server=Machine Name;Database=ProductDB;Trusted_Connection=True;"); optionsBuilder.UseSqlServer(Configuration.GetConnectionString("DefaultConnection")); }
public DbSet<Product> Products { get; set; } } }
添加 NewFolder 服务并使用 IProductService.cs 添加服务接口名称。
namespace RabbitMQProductApi.Services { public interface IProductServices { public IEnumerable<Product> GetProducts(); public Product GetProduct(int id); public Product AddProduct(Product product); public Product UpdateProduct(Product product); public bool DeleteProduct(int id); public bool DeleteProduct(Product product);
} }
添加 ProductService.cs 类使用 IProductServices。
namespace RabbitMQProductApi.Services { public class ProductServices : IProductServices { private readonly DBContextClass _dbContext; public ProductServices(DBContextClass dbContext) { _dbContext = dbContext; }
public IEnumerable<Product> GetProducts() { return _dbContext.Products.ToList(); }
public Product GetProduct(int id) { return _dbContext.Products.FirstOrDefault(x => x.ProductID == id); }
public Product AddProduct(Product product) { var result= _dbContext.Products.Add(product); _dbContext.SaveChanges(); return result.Entity; }
public Product UpdateProduct(Product product) { var result= _dbContext.Products.Update(product); _dbContext.SaveChanges(); return result.Entity; }
public bool DeleteProduct(int id) { var result=_dbContext.Products.FirstOrDefault(x => x.ProductID == id); if (result != null) { _dbContext.Products.Remove(result); _dbContext.SaveChanges(); return true; } return false; }
public bool DeleteProduct(Product product) { var result = _dbContext.Products.FirstOrDefault(x => x.ProductID == product.ProductID); if (result != null) { _dbContext.Products.Remove(result); _dbContext.SaveChanges(); return true; } else { return false; } } } }
添加迁移脚本，在包管理器控制台中执行以下命令。
– add-migration "ProductMigration"
然后通过传递以下命令来更新数据库。
– update-database
protected override void Up(MigrationBuilder migrationBuilder) { migrationBuilder.CreateTable( name: "Products", columns: table => new { ProductID = table.Column<int>(type: "int", nullable: false) .Annotation("SqlServer:Identity", "1, 1"), ProductName = table.Column<string>(type: "nvarchar(max)", nullable: false), ProductDescription = table.Column<string>(type: "nvarchar(max)", nullable: false), ProductPrice = table.Column<int>(type: "int", nullable: false), ProductStock = table.Column<int>(type: "int", nullable: false) }, constraints: table => { table.PrimaryKey("PK_Products", x => x.ProductID); }); }
/// <inheritdoc /> protected override void Down(MigrationBuilder migrationBuilder) { migrationBuilder.DropTable( name: "Products"); }
现在添加 NewFolder，以控制器命名，并添加 ProductController.cs。
namespace RabbitMQProductApi.Controllers { [Route("api/[controller]")] [ApiController] public class ProductController : ControllerBase { private readonly IRabbitMQProducer _rabbitMQProducer; private readonly ILogger<ProductController> _logger; private readonly IProductServices _productService; public ProductController(ILogger<ProductController> logger, IProductServices productService, IRabbitMQProducer rabbitMQProducer) { _logger = logger; _productService = productService; _rabbitMQProducer = rabbitMQProducer; } [HttpGet("productlist")] public IActionResult GetProducts() { var products = _productService.GetProducts(); return Ok(products); } [HttpGet("product/{id}")] public IActionResult GetProduct(int id) { var product = _productService.GetProduct(id); if (product == null) { return NotFound(); } return Ok(product); } [HttpPost("addproduct")] public IActionResult AddProduct([FromBody] Product product) { var newProduct = _productService.AddProduct(product); _rabbitMQProducer.SendProductMessage(newProduct); return Ok(newProduct); //return CreatedAtRoute("GetProduct", new { id = newProduct.ProductID }, newProduct); } [HttpPut("updateproduct")] public IActionResult UpdateProduct([FromBody] Product product) { var updatedProduct = _productService.UpdateProduct(product); return Ok(updatedProduct); } [HttpDelete("deleteproduct/{id}")] public IActionResult DeleteProduct(int id) { var result = _productService.DeleteProduct(id); if (result) { return Ok(); } return NotFound(); }
[HttpDelete("deleteproduct")] public IActionResult DeleteProduct([FromBody]Product product) { var result = _productService.DeleteProduct(product); if (result) { return Ok(); } return NotFound(); }
} 使用下面的代码更新 Program.cs 类。
namespace RabbitMQProductApi { public class Program { public static void Main(string[] args) { var builder = WebApplication.CreateBuilder(args);
// Add services to the container. builder.Services.AddScoped<IProductServices, ProductServices>(); builder.Services.AddDbContext<DBContextClass>(); builder.Services.AddScoped<IRabbitMQProducer, RabbitMQProducer>(); builder.Services.AddControllers(); // Learn more about configuring Swagger/OpenAPI at https://aka.ms/aspnetcore/swashbuckle builder.Services.AddEndpointsApiExplorer(); builder.Services.AddSwaggerGen();
var app = builder.Build();
// Configure the HTTP request pipeline. if (app.Environment.IsDevelopment()) { app.UseSwagger(); app.UseSwaggerUI(); }
app.UseHttpsRedirection();
app.UseAuthorization();
app.MapControllers();
app.Run(); } } }
现在构建项目并在本地机器上运行它。
将项目添加到产品数据库中。
请检查数据库。
完毕
下一部分将会看到如何在客户端应用程序中使用消息。
如果您喜欢此文章，请收藏、点赞、评论，谢谢，祝您快乐每一天。
