一个好的 API 架构对于有效处理微服务之间的通信很重要。不要害怕创建新的微服务,并尽可能地尝试解耦功能。例如,与其创建一个通知服务,不如尝试为电子邮件通知、SMS 通知和移动推送通知创建单独的微服务。
在这里,我假设您有一个 API 网关来管理请求、处理到负载平衡服务器的路由并限制未经授权的访问。
通讯类型
- 同步协议:HTTP 是一种同步协议。客户端发送请求并等待服务的响应。这与客户端代码执行无关,它可以是同步的(线程被阻塞)或异步的(线程未被阻塞,并且响应最终会到达回调)。这里的重点是协议(HTTP/HTTPS)是同步的,客户端代码只有在收到 HTTP 服务器响应后才能继续其任务。
- 异步协议:其他协议如 AMQP(许多操作系统和云环境支持的协议)使用异步消息。客户端代码或消息发送者通常不等待响应。它只是将消息发送到消息代理服务,例如 RabbitMQ 或 Kafka(如果我们使用的是事件驱动架构)。
为什么你应该避免同步协议
- 如果您不断添加相互通信的新微服务,那么在代码中使用端点会造成混乱,尤其是当您必须在端点中传递额外信息时。例如,身份验证令牌。
- 您必须等待耗时的调用才能获得响应。
- 如果响应失败并且您有重试策略,那么它可能会造成瓶颈。
- 如果接收器服务关闭或无法处理请求,那么我们要等到服务启动。例如,在电子商务网站中,用户下订单并请求发送到发货服务以发货,但发货服务关闭,我们丢失了订单。一旦完成,如何将相同的订单发送到运输服务?
- 接收方可能无法一次处理大量请求,因此应该有一个地方让请求必须等待,直到接收方准备好处理下一个请求。
为了应对这些挑战,我们可以使用一个中间服务来处理两个微服务之间的通信,也称为“消息代理”。
RabbitMQ 被广泛用作消息代理服务,如果您将 Azure 云作为托管服务提供商,您也可以使用 Azure 服务总线。
如何使用RabbitMQ来处理微服务之间的通信
可能存在发件人想要向多个服务发送消息的情况。 让我们看看 RabbitMQ 如何处理的下图。
当发布者发送消息时,它被 Exchange 接收,然后 Exchange 将其发送到目标队列。消息保持在队列中,直到接收方接收并处理它。
交换类型
- 直接交换根据消息路由键将消息传递到队列。这是默认的交换类型。
- 扇出交换将消息传递到所有队列。
- Header Exchange 根据消息头标识目标队列。
- 主题交换类似于直接交换,但路由是根据路由模式完成的。它不使用固定的路由键,而是使用通配符。
例如,假设我们有以下路由模式。
- order.logs.customer
- order.logs.international
- order.logs.customer.electronics
- order.logs.international.electronics
“order.*.*.electronics” 的路由模式只匹配第一个词是“order”,第四个词是“electronics”的路由键。
“order.logs.customer.#”的路由模式匹配任何以“order.logs.customer”开头的路由键。
实现RabbitMQ
安装
按照此链接在 Windows 上安装 RabbitMQ。安装后 RabbitMQ 服务将在 http://localhost:15672/ 上启动并运行。在用户名和密码中输入“guest”登录,您将能够看到所有静态信息。
创建发件人服务
RabbitMQ 启动并运行后,创建两个控制台应用程序
Sender:向RabbitMQ发送消息
Receiver:从RabbitMQ接收消息
向两个应用程序添加包“RabbitMQ.Client”。
using System; using RabbitMQ.Client; using System.Text; class Send { public static void Main() { var factory = new ConnectionFactory() { HostName = "localhost" }; using(var connection = factory.CreateConnection()) using(var channel = connection.CreateModel()) { channel.QueueDeclare(queue: "hello", durable: false, exclusive: false, autoDelete: false, arguments: null); string message = "Hello World!"; var body = Encoding.UTF8.GetBytes(message); channel.BasicPublish(exchange: "", routingKey: "hello", basicProperties: null, body: body); Console.WriteLine(" [x] Sent {0}", message); } Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } }
上面的代码将创建一个到 RabbitMQ 的连接,创建一个队列“hello”并向队列发布一条消息。
using RabbitMQ.Client; using RabbitMQ.Client.Events; using System; using System.Text; class Receive { public static void Main() { var factory = new ConnectionFactory() { HostName = "localhost" }; using(var connection = factory.CreateConnection()) using(var channel = connection.CreateModel()) { channel.QueueDeclare(queue: "hello", durable: false, exclusive: false, autoDelete: false, arguments: null); Console.WriteLine(" [*] Waiting for messages."); var consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { var body = ea.Body.ToArray(); var message = Encoding.UTF8.GetString(body); Console.WriteLine(" [x] Received {0}", message); }; channel.BasicConsume(queue: "hello", autoAck: true, consumer: consumer); Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } } }
上面的代码将创建一个到 RabbitMQ 的连接,创建一个队列(如果它还没有创建),并注册一个将接收和处理消息的处理程序。
在运行发送方和接收方应用程序时,您将能够看到在 RabbitMQ 门户上创建的队列,以及表示收到新消息的图形上的尖峰。从门户中,您将能够看到哪个服务有待处理的消息,您可以添加该服务的另一个实例以进行负载平衡。
一开始你可以使用rabbitMQ,事情会很顺利。但是当复杂性增加并且您有很多端点调用其他服务时,它就会造成混乱。很快,您会发现自己围绕驱动程序创建了一个包装器,这样您就可以减少需要编写的代码量。例如,每次您调用另一个服务的端点时,您都必须提供身份验证令牌。然后你会发现自己需要处理 ack 与 nack,你将为此创建一个简单的 API。最终,您将需要处理有害消息——格式错误并导致异常的消息。
要处理所有这些工作流,您可以使用 NserviceBus。让我们讨论一个项目结构:
考虑到这种架构,ClientUI 端点将 PlaceOrder 命令发送到 Sales 端点。 因此,Sales 端点将使用发布/订阅模式发布 OrderPlaced 事件,该事件将由 Billing 端点接收。
NserviceBus 配置:
class Program { static async Task Main(string[] args) { await CreateHostBuilder(args).RunConsoleAsync(); } public static IHostBuilder CreateHostBuilder(string[] args) { return Host.CreateDefaultBuilder(args) .UseNServiceBus(context => { var endpointConfiguration = new EndpointConfiguration("Sales"); //configure transport - configure where your message will //be published/saved //you can configure it for RabbitMq, Azure Queue, Amazon //SQS or any other cloud provider endpointConfiguration.UseTransport<LearningTransport>(); endpointConfiguration.SendFailedMessagesTo("error"); //When a message fails processing //it will be forwarded here. endpointConfiguration.AuditProcessedMessagesTo("audit"); //All messages received by an endpoint //will be forwarded to the audit queue. return endpointConfiguration; }); } }
然后使用 IMessageSession 对象发送消息:
public class HomeController : Controller { static int messagesSent; private readonly ILogger<HomeController> _log; private readonly IMessageSession _messageSession; public HomeController(IMessageSession messageSession, ILogger<HomeController> logger) { _messageSession = messageSession; _log = logger; } [HttpPost] public async Task<ActionResult> PlaceOrder() { string orderId = Guid.NewGuid().ToString().Substring(0, 8); var command = new PlaceOrder { OrderId = orderId }; // Send the command await _messageSession.Send(command) .ConfigureAwait(false); _log.LogInformation($"Sending PlaceOrder, OrderId = {orderId}"); dynamic model = new ExpandoObject(); model.OrderId = orderId; model.MessagesSent = Interlocked.Increment(ref messagesSent); return View(model); } }
最后,添加一个处理程序来接收和处理消息:
public class PlaceOrderHandler : IHandleMessages<PlaceOrder> { static readonly ILog log = LogManager.GetLogger<PlaceOrderHandler>(); static readonly Random random = new Random(); public Task Handle(PlaceOrder message, IMessageHandlerContext context) { log.Info($"Received PlaceOrder, OrderId = {message.OrderId}"); return Task.CompletedTask; } }
这是 NserviceBus 和 RabbitMQ 的基本实现。
概括
在服务之间通信时避免使用同步协议。 使用 RabbitMQ 在服务之间进行通信并在消息从源传送到目标之前临时保存它们。 使用 NserviceBus 解耦应用程序代码和消息代理,并管理长时间运行的请求。
原文:https://irfanyusanif.medium.com/how-to-communicate-between-microservice…
最新内容
- 3 hours ago
- 3 hours ago
- 3 days 4 hours ago
- 3 days 17 hours ago
- 5 days 4 hours ago
- 5 days 22 hours ago
- 5 days 22 hours ago
- 5 days 22 hours ago
- 5 days 22 hours ago
- 5 days 22 hours ago