【RabbitMQ】RPC模式（请求/回复）

本章目标

• 理解RabbitMQ RPC模式的工作原理和适用场景。
  
• 掌握回调队列（Callback Queue）和关联ID（Correlation Id）的使用。
  
• 实现基于RabbitMQ的异步RPC调用。
  
• 学习RPC模式下的错误处理和超时机制。
  
• 构建完整的微服务间同步通信解决方案。
  

一、理论部分

1. RPC模式简介

RPC（Remote Procedure Call）模式允许客户端应用程序调用远程服务器上的方法，就像调用本地方法一样。在RabbitMQ中，RPC是通过消息队列实现的异步RPC。
与传统HTTP RPC的区别：

• HTTP RPC：同步，直接连接，需要服务端在线
  
• 消息队列RPC：异步，通过消息代理，支持解耦和负载均衡
  

2. RabbitMQ RPC核心组件

• 请求队列（Request Queue）：客户端发送请求的队列
  
• 回复队列（Reply Queue）：服务器返回响应的队列
  
• 关联ID（Correlation Id）：匹配请求和响应的唯一标识
  
• 消息属性：使用IBasicProperties.ReplyTo和IBasicProperties.CorrelationId
  

3. RPC工作流程

1. Client端：
2. 1. 生成唯一CorrelationId
3. 2. 创建临时回复队列
4. 3. 发送请求到请求队列，设置ReplyTo和CorrelationId
5. 4. 监听回复队列，等待匹配的CorrelationId
7. Server端：
8. 1. 监听请求队列
9. 2. 处理请求
10. 3. 将响应发送到请求中的ReplyTo队列
11. 4. 设置相同的CorrelationId
13. Client端：
14. 5. 收到响应，根据CorrelationId匹配请求
15. 6. 处理响应

复制代码

4. 适用场景

• 需要同步响应的异步操作
  
• 微服务间的同步通信
  
• 计算密集型任务的分布式处理
  
• 需要负载均衡的同步调用
  

二、实操部分：构建分布式计算服务

我们将创建一个分布式斐波那契数列计算服务，演示完整的RPC模式实现。
第1步：创建项目结构

1. # 创建解决方案
2. dotnet new sln -n RpcSystem
4. # 创建项目
5. dotnet new webapi -n RpcClient.API
6. dotnet new classlib -n RpcClient.Core
7. dotnet new classlib -n RpcServer.Service
8. dotnet new classlib -n RpcShared
10. # 添加到解决方案
11. dotnet sln add RpcClient.API/RpcClient.API.csproj
12. dotnet sln add RpcClient.Core/RpcClient.Core.csproj
13. dotnet sln add RpcServer.Service/RpcServer.Service.csproj
14. dotnet sln add RpcShared/RpcShared.csproj
16. # 添加项目引用
17. dotnet add RpcClient.API reference RpcClient.Core
18. dotnet add RpcClient.API reference RpcShared
19. dotnet add RpcClient.Core reference RpcShared
20. dotnet add RpcServer.Service reference RpcShared
22. # 添加NuGet包
23. cd RpcClient.API
24. dotnet add package RabbitMQ.Client
26. cd ../RpcClient.Core
27. dotnet add package RabbitMQ.Client
29. cd ../RpcServer.Service
30. dotnet add package RabbitMQ.Client

复制代码

第2步：定义共享模型（RpcShared）

Models/RpcRequest.cs

1. using System.Text.Json.Serialization;
3. namespace RpcShared.Models
4. {
5.     public class RpcRequest
6.     {
7.         [JsonPropertyName("requestId")]
8.         public string RequestId { get; set; } = Guid.NewGuid().ToString();
10.         [JsonPropertyName("method")]
11.         public string Method { get; set; } = string.Empty;
13.         [JsonPropertyName("parameters")]
14.         public Dictionary<string, object> Parameters { get; set; } = new();
16.         [JsonPropertyName("timestamp")]
17.         public DateTime Timestamp { get; set; } = DateTime.UtcNow;
19.         public RpcRequest WithParameter(string key, object value)
20.         {
21.             Parameters[key] = value;
22.             return this;
23.         }
25.         public T? GetParameter<T>(string key)
26.         {
27.             if (Parameters.TryGetValue(key, out var value))
28.             {
29.                 try
30.                 {
31.                     return (T)Convert.ChangeType(value, typeof(T));
32.                 }
33.                 catch
34.                 {
35.                     return default;
36.                 }
37.             }
38.             return default;
39.         }
40.     }
41. }

复制代码

View CodeModels/RpcResponse.cs

1. using System.Text.Json.Serialization;
3. namespace RpcShared.Models
4. {
5.     public class RpcResponse
6.     {
7.         [JsonPropertyName("requestId")]
8.         public string RequestId { get; set; } = string.Empty;
10.         [JsonPropertyName("success")]
11.         public bool Success { get; set; }
13.         [JsonPropertyName("data")]
14.         public object? Data { get; set; }
16.         [JsonPropertyName("error")]
17.         public string? Error { get; set; }
19.         [JsonPropertyName("timestamp")]
20.         public DateTime Timestamp { get; set; } = DateTime.UtcNow;
22.         [JsonPropertyName("processingTimeMs")]
23.         public long ProcessingTimeMs { get; set; }
25.         public static RpcResponse SuccessResponse(string requestId, object data, long processingTimeMs = 0)
26.         {
27.             return new RpcResponse
28.             {
29.                 RequestId = requestId,
30.                 Success = true,
31.                 Data = data,
32.                 ProcessingTimeMs = processingTimeMs
33.             };
34.         }
36.         public static RpcResponse ErrorResponse(string requestId, string error, long processingTimeMs = 0)
37.         {
38.             return new RpcResponse
39.             {
40.                 RequestId = requestId,
41.                 Success = false,
42.                 Error = error,
43.                 ProcessingTimeMs = processingTimeMs
44.             };
45.         }
47.         public T? GetData<T>()
48.         {
49.             if (Data is JsonElement jsonElement)
50.             {
51.                 return jsonElement.Deserialize<T>();
52.             }
53.             return Data is T typedData ? typedData : default;
54.         }
55.     }
56. }

复制代码

View CodeMessages/FibonacciRequest.cs

1. namespace RpcShared.Messages
2. {
3.     public class FibonacciRequest
4.     {
5.         public int Number { get; set; }
6.         public bool UseOptimizedAlgorithm { get; set; } = true;
7.     }
9.     public class FibonacciResponse
10.     {
11.         public long Result { get; set; }
12.         public long CalculationTimeMs { get; set; }
13.         public int InputNumber { get; set; }
14.     }
15. }

复制代码

View Code第3步：RPC客户端核心库（RpcClient.Core）

Services/IRpcClient.cs

1. using RpcShared.Models;
3. namespace RpcClient.Core.Services
4. {
5.     public interface IRpcClient : IDisposable
6.     {
7.         Task<RpcResponse> CallAsync(RpcRequest request, TimeSpan timeout);
8.         Task<TResponse?> CallAsync<TResponse>(RpcRequest request, TimeSpan timeout) where TResponse : class;
9.     }
10. }

复制代码

View CodeServices/RpcClient.cs

1. using System.Collections.Concurrent;
2. using System.Text;
3. using System.Text.Json;
4. using Microsoft.Extensions.Logging;
5. using RabbitMQ.Client;
6. using RabbitMQ.Client.Events;
7. using RpcShared.Models;
9. namespace RpcClient.Core.Services
10. {
11.     public class RpcClient : IRpcClient
12.     {
13.         private readonly IConnection _connection;
14.         private readonly IModel _channel;
15.         private readonly ILogger<RpcClient> _logger;
16.         private readonly string _replyQueueName;
17.         private readonly ConcurrentDictionary<string, TaskCompletionSource<RpcResponse>> _pendingRequests;
18.         private readonly AsyncEventingBasicConsumer _consumer;
19.         private bool _disposed = false;
21.         public RpcClient(
22.             IConnectionFactory connectionFactory,
23.             ILogger<RpcClient> logger)
24.         {
25.             _logger = logger;
26.             _pendingRequests = new ConcurrentDictionary<string, TaskCompletionSource<RpcResponse>>();
28.             // 建立连接和通道
29.             _connection = connectionFactory.CreateConnection();
30.             _channel = _connection.CreateModel();
32.             // 声明临时回复队列（排他性，连接关闭时自动删除）
33.             _replyQueueName = _channel.QueueDeclare(
34.                 queue: "",
35.                 durable: false,
36.                 exclusive: true,
37.                 autoDelete: true,
38.                 arguments: null).QueueName;
40.             // 创建消费者监听回复队列
41.             _consumer = new AsyncEventingBasicConsumer(_channel);
42.             _consumer.Received += OnResponseReceived;
44.             // 开始消费回复队列
45.             _channel.BasicConsume(
46.                 queue: _replyQueueName,
47.                 autoAck: false,
48.                 consumer: _consumer);
50.             _logger.LogInformation("RPC Client initialized with reply queue: {ReplyQueue}", _replyQueueName);
51.         }
53.         public async Task<RpcResponse> CallAsync(RpcRequest request, TimeSpan timeout)
54.         {
55.             if (_disposed)
56.                 throw new ObjectDisposedException(nameof(RpcClient));
58.             var tcs = new TaskCompletionSource<RpcResponse>();
59.             var cancellationTokenSource = new CancellationTokenSource(timeout);
61.             // 注册超时取消
62.             cancellationTokenSource.Token.Register(() =>
63.             {
64.                 if (_pendingRequests.TryRemove(request.RequestId, out var removedTcs))
65.                 {
66.                     removedTcs.TrySetException(new TimeoutException($"RPC call timed out after {timeout.TotalSeconds} seconds"));
67.                     _logger.LogWarning("RPC request {RequestId} timed out", request.RequestId);
68.                 }
69.             });
71.             // 将请求添加到待处理字典
72.             if (!_pendingRequests.TryAdd(request.RequestId, tcs))
73.             {
74.                 throw new InvalidOperationException($"Request with ID {request.RequestId} is already pending");
75.             }
77.             try
78.             {
79.                 // 序列化请求
80.                 var requestJson = JsonSerializer.Serialize(request);
81.                 var requestBody = Encoding.UTF8.GetBytes(requestJson);
83.                 // 设置消息属性
84.                 var properties = _channel.CreateBasicProperties();
85.                 properties.ReplyTo = _replyQueueName;
86.                 properties.CorrelationId = request.RequestId;
87.                 properties.Persistent = true;
89.                 _logger.LogDebug("Sending RPC request {RequestId} to queue: rpc_queue", request.RequestId);
91.                 // 发布请求到RPC队列
92.                 _channel.BasicPublish(
93.                     exchange: "",
94.                     routingKey: "rpc_queue",
95.                     basicProperties: properties,
96.                     body: requestBody);
98.                 _logger.LogInformation("RPC request {RequestId} sent successfully", request.RequestId);
100.                 // 等待响应
101.                 return await tcs.Task;
102.             }
103.             catch (Exception ex)
104.             {
105.                 // 发生异常时移除待处理请求
106.                 _pendingRequests.TryRemove(request.RequestId, out _);
107.                 _logger.LogError(ex, "Error sending RPC request {RequestId}", request.RequestId);
108.                 throw;
109.             }
110.         }
112.         public async Task<TResponse?> CallAsync<TResponse>(RpcRequest request, TimeSpan timeout) where TResponse : class
113.         {
114.             var response = await CallAsync(request, timeout);
115.             
116.             if (!response.Success)
117.             {
118.                 throw new InvalidOperationException($"RPC call failed: {response.Error}");
119.             }
121.             return response.GetData<TResponse>();
122.         }
124.         private async Task OnResponseReceived(object sender, BasicDeliverEventArgs ea)
125.         {
126.             var responseBody = ea.Body.ToArray();
127.             var responseJson = Encoding.UTF8.GetString(responseBody);
128.             var correlationId = ea.BasicProperties.CorrelationId;
130.             _logger.LogDebug("Received RPC response for correlation ID: {CorrelationId}", correlationId);
132.             try
133.             {
134.                 var response = JsonSerializer.Deserialize<RpcResponse>(responseJson);
135.                 if (response == null)
136.                 {
137.                     _logger.LogError("Failed to deserialize RPC response for correlation ID: {CorrelationId}", correlationId);
138.                     return;
139.                 }
141.                 // 查找匹配的待处理请求
142.                 if (_pendingRequests.TryRemove(correlationId, out var tcs))
143.                 {
144.                     tcs.TrySetResult(response);
145.                     _logger.LogDebug("RPC response for {CorrelationId} delivered to waiting task", correlationId);
146.                 }
147.                 else
148.                 {
149.                     _logger.LogWarning("Received response for unknown correlation ID: {CorrelationId}", correlationId);
150.                 }
152.                 // 手动确认消息
153.                 _channel.BasicAck(ea.DeliveryTag, false);
154.             }
155.             catch (Exception ex)
156.             {
157.                 _logger.LogError(ex, "Error processing RPC response for correlation ID: {CorrelationId}", correlationId);
158.                
159.                 // 处理失败时拒绝消息（不重新入队）
160.                 _channel.BasicNack(ea.DeliveryTag, false, false);
161.                
162.                 // 如果反序列化失败，仍然通知等待的任务
163.                 if (_pendingRequests.TryRemove(correlationId, out var tcs))
164.                 {
165.                     tcs.TrySetException(new InvalidOperationException("Failed to process RPC response"));
166.                 }
167.             }
169.             await Task.CompletedTask;
170.         }
172.         public void Dispose()
173.         {
174.             if (!_disposed)
175.             {
176.                 _disposed = true;
178.                 // 取消所有待处理的请求
179.                 foreach (var (requestId, tcs) in _pendingRequests)
180.                 {
181.                     tcs.TrySetCanceled();
182.                 }
183.                 _pendingRequests.Clear();
185.                 _channel?.Close();
186.                 _channel?.Dispose();
187.                 _connection?.Close();
188.                 _connection?.Dispose();
190.                 _logger.LogInformation("RPC Client disposed");
191.             }
192.         }
193.     }
194. }

复制代码

Services/FibonacciRpcClient.cs

1. using RpcClient.Core.Services;
2. using RpcShared.Messages;
3. using RpcShared.Models;
5. namespace RpcClient.Core.Services
6. {
7.     public class FibonacciRpcClient
8.     {
9.         private readonly IRpcClient _rpcClient;
10.         private readonly ILogger<FibonacciRpcClient> _logger;
12.         public FibonacciRpcClient(IRpcClient rpcClient, ILogger<FibonacciRpcClient> logger)
13.         {
14.             _rpcClient = rpcClient;
15.             _logger = logger;
16.         }
18.         public async Task<long> CalculateFibonacciAsync(int number, bool useOptimized = true, TimeSpan? timeout = null)
19.         {
20.             var request = new RpcRequest
21.             {
22.                 Method = "fibonacci.calculate",
23.                 Timestamp = DateTime.UtcNow
24.             }
25.             .WithParameter("number", number)
26.             .WithParameter("useOptimized", useOptimized);
28.             timeout ??= TimeSpan.FromSeconds(30);
30.             try
31.             {
32.                 _logger.LogInformation("Calculating Fibonacci({Number}) with timeout {Timeout}s",
33.                     number, timeout.Value.TotalSeconds);
35.                 var response = await _rpcClient.CallAsync<FibonacciResponse>(request, timeout.Value);
36.                
37.                 if (response != null)
38.                 {
39.                     _logger.LogInformation(
40.                         "Fibonacci({Number}) = {Result} (calculated in {Time}ms)",
41.                         number, response.Result, response.CalculationTimeMs);
42.                     
43.                     return response.Result;
44.                 }
46.                 throw new InvalidOperationException("Received null response from RPC server");
47.             }
48.             catch (TimeoutException ex)
49.             {
50.                 _logger.LogError(ex, "Fibonacci calculation timed out for number {Number}", number);
51.                 throw;
52.             }
53.             catch (Exception ex)
54.             {
55.                 _logger.LogError(ex, "Error calculating Fibonacci for number {Number}", number);
56.                 throw;
57.             }
58.         }
60.         public async Task<FibonacciResponse> CalculateFibonacciDetailedAsync(int number, bool useOptimized = true, TimeSpan? timeout = null)
61.         {
62.             var request = new RpcRequest
63.             {
64.                 Method = "fibonacci.calculate",
65.                 Timestamp = DateTime.UtcNow
66.             }
67.             .WithParameter("number", number)
68.             .WithParameter("useOptimized", useOptimized);
70.             timeout ??= TimeSpan.FromSeconds(30);
72.             var response = await _rpcClient.CallAsync<FibonacciResponse>(request, timeout.Value);
73.             return response ?? throw new InvalidOperationException("Received null response from RPC server");
74.         }
75.     }
76. }

复制代码

第4步：RPC客户端API（RpcClient.API）

Program.cs

1. using RpcClient.API.Services;
2. using RpcClient.Core.Services;
3. using RpcShared.Models;
4. using RabbitMQ.Client;
6. var builder = WebApplication.CreateBuilder(args);
8. // Add services to the container.
9. builder.Services.AddControllers();
10. builder.Services.AddEndpointsApiExplorer();
11. builder.Services.AddSwaggerGen();
13. // Configure RabbitMQ
14. builder.Services.AddSingleton<IConnectionFactory>(sp =>
15. {
16.     var configuration = sp.GetRequiredService<IConfiguration>();
17.     return new ConnectionFactory
18.     {
19.         HostName = configuration["RabbitMQ:HostName"],
20.         UserName = configuration["RabbitMQ:UserName"],
21.         Password = configuration["RabbitMQ:Password"],
22.         Port = int.Parse(configuration["RabbitMQ:Port"] ?? "5672"),
23.         VirtualHost = configuration["RabbitMQ:VirtualHost"] ?? "/",
24.         DispatchConsumersAsync = true
25.     };
26. });
28. // Register RPC services
29. builder.Services.AddSingleton<IRpcClient, RpcClient>();
30. builder.Services.AddScoped<FibonacciRpcClient>();
31. builder.Services.AddScoped<IMathRpcService, MathRpcService>();
33. // Add health checks
34. builder.Services.AddHealthChecks()
35.     .AddRabbitMQ(provider =>
36.     {
37.         var factory = provider.GetRequiredService<IConfiguration>();
38.         return factory.CreateConnection();
39.     });
41. var app = builder.Build();
43. // Configure the HTTP request pipeline.
44. if (app.Environment.IsDevelopment())
45. {
46.     app.UseSwagger();
47.     app.UseSwaggerUI();
48. }
50. app.UseHttpsRedirection();
51. app.UseAuthorization();
52. app.MapControllers();
53. app.MapHealthChecks("/health");
55. app.Run();

复制代码

View CodeServices/IMathRpcService.cs

1. using RpcShared.Messages;
3. namespace RpcClient.API.Services
4. {
5.     public interface IMathRpcService
6.     {
7.         Task<long> CalculateFibonacciAsync(int number);
8.         Task<FibonacciResponse> CalculateFibonacciDetailedAsync(int number);
9.         Task<bool> HealthCheckAsync();
10.     }
11. }

复制代码

View CodeServices/MathRpcService.cs

1. using RpcClient.Core.Services;
2. using RpcShared.Messages;
4. namespace RpcClient.API.Services
5. {
6.     public class MathRpcService : IMathRpcService
7.     {
8.         private readonly FibonacciRpcClient _fibonacciClient;
9.         private readonly ILogger<MathRpcService> _logger;
11.         public MathRpcService(FibonacciRpcClient fibonacciClient, ILogger<MathRpcService> logger)
12.         {
13.             _fibonacciClient = fibonacciClient;
14.             _logger = logger;
15.         }
17.         public async Task<long> CalculateFibonacciAsync(int number)
18.         {
19.             if (number < 0)
20.                 throw new ArgumentException("Number must be non-negative", nameof(number));
22.             if (number > 50)
23.                 throw new ArgumentException("Number too large for demonstration", nameof(number));
25.             return await _fibonacciClient.CalculateFibonacciAsync(number);
26.         }
28.         public async Task<FibonacciResponse> CalculateFibonacciDetailedAsync(int number)
29.         {
30.             if (number < 0)
31.                 throw new ArgumentException("Number must be non-negative", nameof(number));
33.             if (number > 50)
34.                 throw new ArgumentException("Number too large for demonstration", nameof(number));
36.             return await _fibonacciClient.CalculateFibonacciDetailedAsync(number);
37.         }
39.         public async Task<bool> HealthCheckAsync()
40.         {
41.             try
42.             {
43.                 // 简单的健康检查：计算 Fibonacci(1)
44.                 var result = await _fibonacciClient.CalculateFibonacciAsync(1, timeout: TimeSpan.FromSeconds(5));
45.                 return result == 1;
46.             }
47.             catch (Exception ex)
48.             {
49.                 _logger.LogWarning(ex, "RPC health check failed");
50.                 return false;
51.             }
52.         }
53.     }
54. }

复制代码

View CodeControllers/MathController.cs

1. using Microsoft.AspNetCore.Mvc;
2. using RpcClient.API.Services;
3. using RpcShared.Messages;
5. namespace RpcClient.API.Controllers
6. {
7.     [ApiController]
8.     [Route("api/[controller]")]
9.     public class MathController : ControllerBase
10.     {
11.         private readonly IMathRpcService _mathService;
12.         private readonly ILogger<MathController> _logger;
14.         public MathController(IMathRpcService mathService, ILogger<MathController> logger)
15.         {
16.             _mathService = mathService;
17.             _logger = logger;
18.         }
20.         [HttpGet("fibonacci/{number}")]
21.         public async Task> CalculateFibonacci(int number)
22.         {
23.             try
24.             {
25.                 _logger.LogInformation("Calculating Fibonacci({Number}) via RPC", number);
26.                 var result = await _mathService.CalculateFibonacciAsync(number);
27.                 return Ok(result);
28.             }
29.             catch (ArgumentException ex)
30.             {
31.                 return BadRequest(ex.Message);
32.             }
33.             catch (TimeoutException ex)
34.             {
35.                 _logger.LogWarning(ex, "Fibonacci calculation timed out for number {Number}", number);
36.                 return StatusCode(408, "Calculation timed out");
37.             }
38.             catch (Exception ex)
39.             {
40.                 _logger.LogError(ex, "Error calculating Fibonacci for number {Number}", number);
41.                 return StatusCode(500, "Internal server error");
42.             }
43.         }
45.         [HttpGet("fibonacci/{number}/detailed")]
46.         public async Task> CalculateFibonacciDetailed(int number)
47.         {
48.             try
49.             {
50.                 _logger.LogInformation("Calculating Fibonacci({Number}) with details via RPC", number);
51.                 var result = await _mathService.CalculateFibonacciDetailedAsync(number);
52.                 return Ok(result);
53.             }
54.             catch (ArgumentException ex)
55.             {
56.                 return BadRequest(ex.Message);
57.             }
58.             catch (TimeoutException ex)
59.             {
60.                 _logger.LogWarning(ex, "Fibonacci calculation timed out for number {Number}", number);
61.                 return StatusCode(408, "Calculation timed out");
62.             }
63.             catch (Exception ex)
64.             {
65.                 _logger.LogError(ex, "Error calculating Fibonacci for number {Number}", number);
66.                 return StatusCode(500, "Internal server error");
67.             }
68.         }
70.         [HttpGet("health")]
71.         public async Task HealthCheck()
72.         {
73.             var isHealthy = await _mathService.HealthCheckAsync();
74.             return isHealthy ? Ok("RPC service is healthy") : StatusCode(503, "RPC service is unavailable");
75.         }
76.     }
77. }

复制代码

View Code第5步：RPC服务器（RpcServer.Service）

Program.cs

1. using RpcServer.Service.Services;
2. using RpcShared.Models;
3. using RabbitMQ.Client;
5. var builder = Host.CreateApplicationBuilder(args);
7. builder.Services.AddHostedService<FibonacciRpcServer>();
9. // Configure RabbitMQ
10. builder.Services.AddSingleton<IConnectionFactory>(sp =>
11. {
12.     var configuration = sp.GetRequiredService<IConfiguration>();
13.     return new ConnectionFactory
14.     {
15.         HostName = configuration["RabbitMQ:HostName"],
16.         UserName = configuration["RabbitMQ:UserName"],
17.         Password = configuration["RabbitMQ:Password"],
18.         Port = int.Parse(configuration["RabbitMQ:Port"] ?? "5672"),
19.         VirtualHost = configuration["RabbitMQ:VirtualHost"] ?? "/",
20.         DispatchConsumersAsync = true
21.     };
22. });
24. builder.Services.AddSingleton<FibonacciCalculator>();
26. var host = builder.Build();
27. host.Run();

复制代码

View CodeServices/FibonacciCalculator.cs

1. using RpcShared.Messages;
3. namespace RpcServer.Service.Services
4. {
5.     public class FibonacciCalculator
6.     {
7.         private readonly ILogger<FibonacciCalculator> _logger;
8.         private readonly Dictionary<int, long> _cache = new();
10.         public FibonacciCalculator(ILogger<FibonacciCalculator> logger)
11.         {
12.             _logger = logger;
13.         }
15.         public FibonacciResponse Calculate(int number, bool useOptimized = true)
16.         {
17.             var startTime = DateTime.UtcNow;
19.             try
20.             {
21.                 _logger.LogInformation("Calculating Fibonacci({Number}) with optimized: {Optimized}",
22.                     number, useOptimized);
24.                 long result;
25.                 if (useOptimized)
26.                 {
27.                     result = CalculateOptimized(number);
28.                 }
29.                 else
30.                 {
31.                     result = CalculateNaive(number);
32.                 }
34.                 var calculationTime = (DateTime.UtcNow - startTime).TotalMilliseconds;
36.                 _logger.LogInformation(
37.                     "Fibonacci({Number}) = {Result} (calculated in {Time}ms)",
38.                     number, result, calculationTime);
40.                 return new FibonacciResponse
41.                 {
42.                     Result = result,
43.                     CalculationTimeMs = (long)calculationTime,
44.                     InputNumber = number
45.                 };
46.             }
47.             catch (Exception ex)
48.             {
49.                 _logger.LogError(ex, "Error calculating Fibonacci({Number})", number);
50.                 throw;
51.             }
52.         }
54.         private long CalculateOptimized(int n)
55.         {
56.             if (n < 0) throw new ArgumentException("Number must be non-negative");
57.             if (n <= 1) return n;
59.             // 检查缓存
60.             if (_cache.TryGetValue(n, out var cachedResult))
61.             {
62.                 _logger.LogDebug("Cache hit for Fibonacci({Number})", n);
63.                 return cachedResult;
64.             }
66.             long a = 0, b = 1;
68.             for (int i = 2; i <= n; i++)
69.             {
70.                 var temp = a + b;
71.                 a = b;
72.                 b = temp;
74.                 // 缓存中间结果
75.                 if (i % 10 == 0) // 每10个数缓存一次以减少内存使用
76.                 {
77.                     _cache[i] = b;
78.                 }
79.             }
81.             // 缓存最终结果
82.             _cache[n] = b;
83.             return b;
84.         }
86.         private long CalculateNaive(int n)
87.         {
88.             if (n < 0) throw new ArgumentException("Number must be non-negative");
89.             if (n <= 1) return n;
91.             // 模拟计算密集型任务
92.             Thread.Sleep(100);
94.             return CalculateNaive(n - 1) + CalculateNaive(n - 2);
95.         }
97.         public void ClearCache()
98.         {
99.             _cache.Clear();
100.             _logger.LogInformation("Fibonacci cache cleared");
101.         }
102.     }
103. }

复制代码

View Code第6步：高级特性 - 带重试的RPC客户端

Services/ResilientRpcClient.cs

1. using System.Text;
2. using System.Text.Json;
3. using Microsoft.Extensions.Options;
4. using RabbitMQ.Client;
5. using RabbitMQ.Client.Events;
6. using RpcShared.Messages;
7. using RpcShared.Models;
9. namespace RpcServer.Service.Services
10. {
11.     public class FibonacciRpcServer : BackgroundService
12.     {
13.         private readonly IConnection _connection;
14.         private readonly IModel _channel;
15.         private readonly FibonacciCalculator _calculator;
16.         private readonly ILogger<FibonacciRpcServer> _logger;
17.         private const string QueueName = "rpc_queue";
19.         public FibonacciRpcServer(
20.             IConnectionFactory connectionFactory,
21.             FibonacciCalculator calculator,
22.             ILogger<FibonacciRpcServer> logger)
23.         {
24.             _calculator = calculator;
25.             _logger = logger;
27.             // 建立连接和通道
28.             _connection = connectionFactory.CreateConnection();
29.             _channel = _connection.CreateModel();
31.             InitializeQueue();
32.         }
34.         private void InitializeQueue()
35.         {
36.             // 声明RPC请求队列（持久化）
37.             _channel.QueueDeclare(
38.                 queue: QueueName,
39.                 durable: true,
40.                 exclusive: false,
41.                 autoDelete: false,
42.                 arguments: null);
44.             // 设置公平分发，每次只处理一个请求
45.             _channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
47.             _logger.LogInformation("Fibonacci RPC Server initialized and listening on queue: {QueueName}", QueueName);
48.         }
50.         protected override Task ExecuteAsync(CancellationToken stoppingToken)
51.         {
52.             stoppingToken.ThrowIfCancellationRequested();
54.             var consumer = new AsyncEventingBasicConsumer(_channel);
55.             consumer.Received += async (model, ea) =>
56.             {
57.                 RpcResponse response = null;
58.                 string correlationId = ea.BasicProperties.CorrelationId;
59.                 string replyTo = ea.BasicProperties.ReplyTo;
61.                 _logger.LogDebug("Received RPC request with correlation ID: {CorrelationId}", correlationId);
63.                 try
64.                 {
65.                     // 处理请求
66.                     response = await ProcessRequestAsync(ea.Body.ToArray());
67.                     response.RequestId = correlationId;
68.                 }
69.                 catch (Exception ex)
70.                 {
71.                     _logger.LogError(ex, "Error processing RPC request {CorrelationId}", correlationId);
72.                     response = RpcResponse.ErrorResponse(correlationId, ex.Message);
73.                 }
74.                 finally
75.                 {
76.                     // 发送响应
77.                     if (!string.IsNullOrEmpty(replyTo))
78.                     {
79.                         SendResponse(replyTo, correlationId, response ??
80.                             RpcResponse.ErrorResponse(correlationId, "Unknown error occurred"));
81.                     }
83.                     // 确认消息处理完成
84.                     _channel.BasicAck(ea.DeliveryTag, false);
85.                 }
86.             };
88.             _channel.BasicConsume(
89.                 queue: QueueName,
90.                 autoAck: false, // 手动确认
91.                 consumer: consumer);
93.             _logger.LogInformation("Fibonacci RPC Server started successfully");
95.             return Task.CompletedTask;
96.         }
98.         private async Task<RpcResponse> ProcessRequestAsync(byte[] body)
99.         {
100.             var startTime = DateTime.UtcNow;
102.             try
103.             {
104.                 var requestJson = Encoding.UTF8.GetString(body);
105.                 var request = JsonSerializer.Deserialize<RpcRequest>(requestJson);
107.                 if (request == null)
108.                 {
109.                     return RpcResponse.ErrorResponse("unknown", "Invalid request format");
110.                 }
112.                 _logger.LogInformation("Processing RPC request {RequestId}, Method: {Method}",
113.                     request.RequestId, request.Method);
115.                 // 根据方法名路由到不同的处理逻辑
116.                 object result = request.Method.ToLowerInvariant() switch
117.                 {
118.                     "fibonacci.calculate" => ProcessFibonacciRequest(request),
119.                     "ping" => new { message = "pong", timestamp = DateTime.UtcNow },
120.                     _ => throw new NotSupportedException($"Method {request.Method} is not supported")
121.                 };
123.                 var processingTime = (DateTime.UtcNow - startTime).TotalMilliseconds;
125.                 return RpcResponse.SuccessResponse(
126.                     request.RequestId,
127.                     result,
128.                     (long)processingTime);
129.             }
130.             catch (Exception ex)
131.             {
132.                 var processingTime = (DateTime.UtcNow - startTime).TotalMilliseconds;
133.                 _logger.LogError(ex, "Error processing RPC request");
134.                
135.                 return RpcResponse.ErrorResponse(
136.                     "unknown",
137.                     ex.Message,
138.                     (long)processingTime);
139.             }
140.         }
142.         private FibonacciResponse ProcessFibonacciRequest(RpcRequest request)
143.         {
144.             var number = request.GetParameter<int>("number");
145.             var useOptimized = request.GetParameter<bool>("useOptimized") ?? true;
147.             if (number < 0)
148.             {
149.                 throw new ArgumentException("Fibonacci number must be non-negative");
150.             }
152.             // 防止过大的计算消耗资源
153.             if (number > 50)
154.             {
155.                 throw new ArgumentException("Number too large for calculation");
156.             }
158.             return _calculator.Calculate(number, useOptimized);
159.         }
161.         private void SendResponse(string replyTo, string correlationId, RpcResponse response)
162.         {
163.             try
164.             {
165.                 var responseJson = JsonSerializer.Serialize(response);
166.                 var responseBody = Encoding.UTF8.GetBytes(responseJson);
168.                 var properties = _channel.CreateBasicProperties();
169.                 properties.CorrelationId = correlationId;
170.                 properties.Persistent = true;
172.                 _channel.BasicPublish(
173.                     exchange: "",
174.                     routingKey: replyTo,
175.                     basicProperties: properties,
176.                     body: responseBody);
178.                 _logger.LogDebug("Sent RPC response for correlation ID: {CorrelationId}", correlationId);
179.             }
180.             catch (Exception ex)
181.             {
182.                 _logger.LogError(ex, "Error sending RPC response for correlation ID: {CorrelationId}", correlationId);
183.             }
184.         }
186.         public override void Dispose()
187.         {
188.             _channel?.Close();
189.             _channel?.Dispose();
190.             _connection?.Close();
191.             _connection?.Dispose();
192.             
193.             _logger.LogInformation("Fibonacci RPC Server disposed");
194.             
195.             base.Dispose();
196.         }
197.     }
198. }

复制代码

View Code第7步：运行与测试

• 启动服务
  1. using Microsoft.Extensions.Logging;
  2. using Polly;
  3. using Polly.Retry;
  4. using RpcShared.Models;
  6. namespace RpcClient.Core.Services
  7. {
  8.     public class ResilientRpcClient : IRpcClient
  9.     {
  10.         private readonly IRpcClient _innerClient;
  11.         private readonly ILogger<ResilientRpcClient> _logger;
  12.         private readonly AsyncRetryPolicy _retryPolicy;
  14.         public ResilientRpcClient(IRpcClient innerClient, ILogger<ResilientRpcClient> logger)
  15.         {
  16.             _innerClient = innerClient;
  17.             _logger = logger;
  19.             // 配置重试策略
  20.             _retryPolicy = Policy
  21.                 .Handle<TimeoutException>()
  22.                 .Or<InvalidOperationException>()
  23.                 .WaitAndRetryAsync(
  24.                     retryCount: 3,
  25.                     sleepDurationProvider: retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)),
  26.                     onRetry: (exception, delay, retryCount, context) =>
  27.                     {
  28.                         _logger.LogWarning(
  29.                             "RPC call failed. Retry {RetryCount} after {Delay}ms. Error: {Error}",
  30.                             retryCount, delay.TotalMilliseconds, exception.Message);
  31.                     });
  32.         }
  34.         public async Task<RpcResponse> CallAsync(RpcRequest request, TimeSpan timeout)
  35.         {
  36.             return await _retryPolicy.ExecuteAsync(async () =>
  37.             {
  38.                 return await _innerClient.CallAsync(request, timeout);
  39.             });
  40.         }
  42.         public async Task<TResponse?> CallAsync<TResponse>(RpcRequest request, TimeSpan timeout) where TResponse : class
  43.         {
  44.             return await _retryPolicy.ExecuteAsync(async () =>
  45.             {
  46.                 return await _innerClient.CallAsync<TResponse>(request, timeout);
  47.             });
  48.         }
  50.         public void Dispose()
  51.         {
  52.             _innerClient?.Dispose();
  53.         }
  54.     }
  55. }

  复制代码
• 测试API
  1. # 终端1：启动RPC服务器
  2. cd RpcServer.Service
  3. dotnet run
  5. # 终端2：启动RPC客户端API
  6. cd RpcClient.API
  7. dotnet run

  复制代码
• 测试错误场景
  1. # 计算斐波那契数列
  2. curl -X GET "https://localhost:7000/api/math/fibonacci/10"
  3. curl -X GET "https://localhost:7000/api/math/fibonacci/20/detailed"
  5. # 健康检查
  6. curl -X GET "https://localhost:7000/api/math/health"

  复制代码
• 观察日志输出
  
  
  
  • 客户端发送请求，生成CorrelationId
    
  • 服务器接收请求，处理计算
    
  • 服务器发送响应，使用相同的CorrelationId
    
  • 客户端接收响应，匹配CorrelationId
    
  
  

第8步：性能测试和监控

创建性能测试控制器

1. # 测试超时（设置很小的超时时间）
2. # 测试无效输入
3. curl -X GET "https://localhost:7000/api/math/fibonacci/-5"
4. curl -X GET "https://localhost:7000/api/math/fibonacci/100"

复制代码

本章总结

在这一章中，我们完整实现了RabbitMQ的RPC模式：

• RPC核心概念：理解了回调队列、关联ID、请求-响应模式。
  
• 客户端实现：创建了能够发送请求并异步等待响应的RPC客户端。
  
• 服务器实现：构建了处理请求并返回响应的RPC服务器。
  
• 错误处理：实现了超时控制、异常处理和重试机制。
  
• 性能优化：使用缓存和优化算法提高计算效率。
  
• ** resilience**：通过Polly实现了弹性重试策略。
  

RPC模式为微服务架构提供了强大的同步通信能力，结合消息队列的异步特性，既保持了系统的解耦性，又提供了同步调用的便利性。这种模式特别适合需要等待计算结果的分布式任务。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！