推荐成都网站建设58同城的网站建设

推荐成都网站建设,58同城的网站建设,房产公司网站模板,如何建设网站兴田德润在那里本章目标 掌握在ASP.NET Core中配置和依赖注入RabbitMQ服务。 学习使用IHostedService/BackgroundService实现常驻消费者服务。 实现基于RabbitMQ的请求-响应模式。 构建完整的微服务间异步通信解决方案。 学习配置管理和健康检查。 一、理论部分 1. ASP.NET Core集成模式 将…本章目标掌握在ASP.NET Core中配置和依赖注入RabbitMQ服务。学习使用IHostedService/BackgroundService实现常驻消费者服务。实现基于RabbitMQ的请求-响应模式。构建完整的微服务间异步通信解决方案。学习配置管理和健康检查。一、理论部分1. ASP.NET Core集成模式将RabbitMQ集成到ASP.NET Core应用程序时我们需要考虑几个关键方面依赖注入正确管理连接和通道的生命周期。托管服务实现后台消息消费者。配置管理从配置文件读取RabbitMQ连接设置。健康检查监控RabbitMQ连接状态。日志记录使用ASP.NET Core的日志系统。2. 生命周期管理IConnection建议注册为单例因为创建TCP连接开销大。IModel建议注册为瞬态或作用域因为通道不是线程安全的。生产者服务可以注册为作用域或瞬态。消费者服务通常在托管服务中管理。3. 托管服务Hosted ServicesASP.NET Core提供了IHostedService接口和BackgroundService基类用于实现长时间运行的后台任务。这是实现RabbitMQ消费者的理想方式。4. 微服务架构中的消息模式异步命令发送指令但不期待立即响应。事件通知广播状态变化。请求-响应类似RPC但通过消息中间件。二、实操部分构建订单处理微服务我们将创建一个完整的订单处理系统包含Order.API接收HTTP订单请求发布消息OrderProcessor.BackgroundService后台处理订单订单状态查询API健康检查配置管理第1步创建项目结构复制代码# 创建解决方案dotnet new sln -n OrderSystem# 创建Web API项目dotnet new webapi -n Order.APIdotnet new classlib -n Order.Coredotnet new classlib -n Order.Infrastructuredotnet new classlib -n OrderProcessor.Service# 添加到解决方案dotnet sln add Order.API/Order.API.csprojdotnet sln add Order.Core/Order.Core.csprojdotnet sln add Order.Infrastructure/Order.Infrastructure.csprojdotnet sln add OrderProcessor.Service/OrderProcessor.Service.csproj# 添加项目引用dotnet add Order.API reference Order.Coredotnet add Order.API reference Order.Infrastructuredotnet add OrderProcessor.Service reference Order.Coredotnet add OrderProcessor.Service reference Order.Infrastructuredotnet add Order.Infrastructure reference Order.Core# 添加NuGet包cd Order.APIdotnet add package RabbitMQ.Clientdotnet add package Microsoft.Extensions.Diagnostics.HealthCheckscd ../Order.Infrastructuredotnet add package RabbitMQ.Clientdotnet add package Microsoft.Extensions.Configurationcd ../OrderProcessor.Servicedotnet add package RabbitMQ.Client复制代码第2步定义领域模型Order.CoreModels/Order.cs复制代码namespace Order.Core.Models{public class Order{public string Id { get; set; } Guid.NewGuid().ToString();public string CustomerId { get; set; }public string ProductId { get; set; }public int Quantity { get; set; }public decimal TotalAmount { get; set; }public OrderStatus Status { get; set; } OrderStatus.Pending;public DateTime CreatedAt { get; set; } DateTime.UtcNow;public DateTime? ProcessedAt { get; set; }}public enum OrderStatus{Pending,Processing,Completed,Failed,Cancelled}}复制代码Messages/OrderMessage.cs复制代码namespace Order.Core.Messages{public class OrderMessage{public string OrderId { get; set; }public string CustomerId { get; set; }public string ProductId { get; set; }public int Quantity { get; set; }public decimal TotalAmount { get; set; }public string Action { get; set; } // create, cancel}public class OrderStatusMessage{public string OrderId { get; set; }public OrderStatus Status { get; set; }public string Message { get; set; }public DateTime Timestamp { get; set; } DateTime.UtcNow;}}复制代码第3步基础设施层Order.InfrastructureServices/IRabbitMQConnection.cs复制代码using RabbitMQ.Client;namespace Order.Infrastructure.Services{public interface IRabbitMQConnection : IDisposable{bool IsConnected { get; }IModel CreateModel();bool TryConnect();}}复制代码Services/RabbitMQConnection.cs复制代码using System.Net.Sockets;using Microsoft.Extensions.Logging;using RabbitMQ.Client;using RabbitMQ.Client.Events;using RabbitMQ.Client.Exceptions;namespace Order.Infrastructure.Services{public class RabbitMQConnection : IRabbitMQConnection{private readonly IConnectionFactory _connectionFactory;private readonly ILoggerRabbitMQConnection _logger;private IConnection _connection;private bool _disposed;private readonly object _syncRoot new object();public RabbitMQConnection(IConnectionFactory connectionFactory, ILoggerRabbitMQConnection logger){_connectionFactory connectionFactory;_logger logger;}public bool IsConnected _connection ! null _connection.IsOpen !_disposed;public IModel CreateModel(){if (!IsConnected){throw new InvalidOperationException(No RabbitMQ connections are available to perform this action);}return _connection.CreateModel();}public bool TryConnect(){lock (_syncRoot){if (IsConnected) return true;_logger.LogInformation(RabbitMQ Client is trying to connect);try{_connection _connectionFactory.CreateConnection();_connection.ConnectionShutdown OnConnectionShutdown;_connection.CallbackException OnCallbackException;_connection.ConnectionBlocked OnConnectionBlocked;_logger.LogInformation(RabbitMQ Client acquired a persistent connection to {HostName} and is subscribed to failure events,_connectionFactory.HostName);return true;}catch (BrokerUnreachableException ex){_logger.LogError(ex, RabbitMQ connection failed: {Message}, ex.Message);return false;}catch (SocketException ex){_logger.LogError(ex, RabbitMQ connection failed: {Message}, ex.Message);return false;}}}private void OnConnectionBlocked(object sender, ConnectionBlockedEventArgs e){if (_disposed) return;_logger.LogWarning(A RabbitMQ connection is blocked. Reason: {Reason}, e.Reason);// 这里可以实现重连逻辑TryConnect();}private void OnCallbackException(object sender, CallbackExceptionEventArgs e){if (_disposed) return;_logger.LogWarning(e.Exception, A RabbitMQ connection throw exception. Trying to re-connect...);// 这里可以实现重连逻辑TryConnect();}private void OnConnectionShutdown(object sender, ShutdownEventArgs reason){if (_disposed) return;_logger.LogWarning(A RabbitMQ connection is on shutdown. Trying to re-connect...);// 这里可以实现重连逻辑TryConnect();}public void Dispose(){if (_disposed) return;_disposed true;try{_connection?.Dispose();}catch (IOException ex){_logger.LogCritical(ex, Error disposing RabbitMQ connection);}}}}复制代码Services/IOrderPublisher.cs复制代码using Order.Core.Messages;namespace Order.Infrastructure.Services{public interface IOrderPublisher{Task PublishOrderCreatedAsync(OrderMessage order);Task PublishOrderStatusAsync(OrderStatusMessage status);}}复制代码Services/OrderPublisher.cs复制代码using System.Text;using System.Text.Json;using Microsoft.Extensions.Logging;using Order.Core.Messages;using RabbitMQ.Client;namespace Order.Infrastructure.Services{public class OrderPublisher : IOrderPublisher{private readonly IRabbitMQConnection _connection;private readonly ILoggerOrderPublisher _logger;private const string ExchangeName order.events;private const string OrderCreatedRoutingKey order.created;private const string OrderStatusRoutingKey order.status;public OrderPublisher(IRabbitMQConnection connection, ILoggerOrderPublisher logger){_connection connection;_logger logger;// 确保交换机和队列存在InitializeInfrastructure();}private void InitializeInfrastructure(){using var channel _connection.CreateModel();// 声明主题交换机channel.ExchangeDeclare(ExchangeName, ExchangeType.Topic, durable: true);// 声明订单创建队列channel.QueueDeclare(order.created.queue, durable: true, exclusive: false, autoDelete: false);channel.QueueBind(order.created.queue, ExchangeName, OrderCreatedRoutingKey);// 声明订单状态队列channel.QueueDeclare(order.status.queue, durable: true, exclusive: false, autoDelete: false);channel.QueueBind(order.status.queue, ExchangeName, OrderStatusRoutingKey);_logger.LogInformation(RabbitMQ infrastructure initialized);}public async Task PublishOrderCreatedAsync(OrderMessage order){await PublishMessageAsync(order, OrderCreatedRoutingKey, OrderCreated);}public async Task PublishOrderStatusAsync(OrderStatusMessage status){await PublishMessageAsync(status, OrderStatusRoutingKey, OrderStatus);}private async Task PublishMessageAsyncT(T message, string routingKey, string messageType){if (!_connection.IsConnected){_connection.TryConnect();}using var channel _connection.CreateModel();var json JsonSerializer.Serialize(message);var body Encoding.UTF8.GetBytes(json);var properties channel.CreateBasicProperties();properties.Persistent true;properties.ContentType application/json;properties.Type messageType;try{channel.BasicPublish(exchange: ExchangeName,routingKey: routingKey,mandatory: true,basicProperties: properties,body: body);_logger.LogInformation(Published {MessageType} message for Order {OrderId},messageType, GetOrderId(message));}catch (Exception ex){_logger.LogError(ex, Error publishing {MessageType} message for Order {OrderId},messageType, GetOrderId(message));throw;}await Task.CompletedTask;}private static string GetOrderIdT(T message){return message switch{OrderMessage order order.OrderId,OrderStatusMessage status status.OrderId,_ unknown};}}}复制代码第4步Order.API项目配置appsettings.json复制代码{RabbitMQ: {HostName: localhost,UserName: myuser,Password: mypassword,Port: 5672,VirtualHost: /},Logging: {LogLevel: {Default: Information,Microsoft.AspNetCore: Warning}},AllowedHosts: *}复制代码Program.cs复制代码using Order.API.Controllers;using Order.API.Services;using Order.Core.Models;using Order.Infrastructure.Services;using RabbitMQ.Client;var builder WebApplication.CreateBuilder(args);// Add services to the container.builder.Services.AddControllers();builder.Services.AddEndpointsApiExplorer();builder.Services.AddSwaggerGen();// Configure RabbitMQbuilder.Services.AddSingletonIConnectionFactory(sp {var configuration sp.GetRequiredServiceIConfiguration();return new ConnectionFactory{HostName configuration[RabbitMQ:HostName],UserName configuration[RabbitMQ:UserName],Password configuration[RabbitMQ:Password],Port int.Parse(configuration[RabbitMQ:Port] ?? 5672),VirtualHost configuration[RabbitMQ:VirtualHost] ?? /,DispatchConsumersAsync true};});// Register RabbitMQ servicesbuilder.Services.AddSingletonIRabbitMQConnection, RabbitMQConnection();builder.Services.AddScopedIOrderPublisher, OrderPublisher();builder.Services.AddScopedIOrderService, OrderService();// Add Health Checksbuilder.Services.AddHealthChecks().AddRabbitMQ(provider {var factory provider.GetRequiredServiceIConnectionFactory();return factory.CreateConnection();}, name: rabbitmq);// Add hosted service for status updates consumerbuilder.Services.AddHostedServiceOrderStatusConsumerService();var app builder.Build();// Configure the HTTP request pipeline.if (app.Environment.IsDevelopment()){app.UseSwagger();app.UseSwaggerUI();}app.UseHttpsRedirection();app.UseAuthorization();app.MapControllers();// Add health check endpointapp.MapHealthChecks(/health);app.Run();复制代码Services/IOrderService.cs复制代码using Order.Core.Models;namespace Order.API.Services{public interface IOrderService{TaskOrder CreateOrderAsync(string customerId, string productId, int quantity, decimal unitPrice);TaskOrder? GetOrderAsync(string orderId);Task UpdateOrderStatusAsync(string orderId, OrderStatus status);}}复制代码Services/OrderService.csView CodeServices/OrderStatusConsumerService.csView CodeControllers/OrdersController.csView Code第5步订单处理器服务OrderProcessor.ServiceProgram.csView CodeServices/OrderProcessorService.csView Code第6步运行与测试启动服务复制代码# 终端1启动Order.APIcd Order.APIdotnet run# 终端2启动OrderProcessor.Servicecd OrderProcessor.Servicedotnet run复制代码测试API复制代码# 创建订单curl -X POST https://localhost:7000/api/orders \-H Content-Type: application/json \-d {customerId: customer-123,productId: product-456,quantity: 2,unitPrice: 29.99}# 查询订单状态curl https://localhost:7000/api/orders/{orderId}复制代码测试健康检查GET https://localhost:7000/health观察日志输出Order.API接收HTTP请求发布订单创建消息OrderProcessor.Service消费订单消息处理业务逻辑发布状态更新Order.API消费状态更新消息测试错误场景停止RabbitMQ服务观察重连机制停止OrderProcessor.Service观察消息堆积重启服务观察消息恢复处理第7步高级特性 - 配置重试和 resilience在Order.Infrastructure中添加Polly支持复制代码// 添加NuGet包dotnet add package Pollydotnet add package Microsoft.Extensions.Http.Polly// 在Program.cs中添加重试策略builder.Services.AddHttpClient(retry-client).AddTransientHttpErrorPolicy(policy policy.WaitAndRetryAsync(3, retryAttempt TimeSpan.FromSeconds(Math.Pow(2, retryAttempt))));