本文展示了如何使用Azure Service Bus Queue, 实现2个ASP.NET Core Api应用之间的消息传输。

配置Azure Service Bus Queue

你可以从官网文档中了解到如何配置一个Azure Service Bus Queue.

这里我们使用Queue或者Topic来实现消息传输。Queue是一种消息传输类型，一旦一个消息被一个消费者接收了，该消息就会从Queue中被移除。

架构图

整个应用的实现如下：

• Api 1负责发送消息
• Api 2负责监听Azure Service Bus，并处理接收到的消息

实现一个Service Bus Queue

这里我们首先需要引入** Microsoft.Azure.ServiceBus ** 程序集。Microsoft.Azure.ServiceBus是Azure Service Bus的客户端库。针对Service Bus的连接字符串我们保存在项目的User Secret中。当部署项目的时候，我们可以使用Azure Key Valut来设置这个Secret值。

为了发送向Azure Service Bus Queue发送消息，我们需要创建一个

SendMessage

MyPayload

MyPayload

Message

using Microsoft.Azure.ServiceBus;
using Microsoft.Extensions.Configuration;
using Newtonsoft.Json;
using System.Text;
using System.Threading.Tasks;

namespace ServiceBusMessaging
{
public class ServiceBusSender
{
private readonly QueueClient _queueClient;
private readonly IConfiguration _configuration;
private const string QUEUE_NAME = "simplequeue";

public ServiceBusSender(IConfiguration configuration)
{
_configuration = configuration;
_queueClient = new QueueClient(
_configuration
.GetConnectionString("ServiceBusConnectionString"),
QUEUE_NAME);
}

public async Task SendMessage(MyPayload payload)
{
string data = JsonConvert.SerializeObject(payload);
Message message = new Message(Encoding.UTF8.GetBytes(data));

await _queueClient.SendAsync(message);
}
}
}

在API 1和API 2中，我们需要将

ServiceBusSender

Swagger

public void ConfigureServices(IServiceCollection services)
{
services.AddMvc();

services.AddScoped<ServiceBusSender>();

services.AddSwaggerGen(c =>
{
c.SwaggerDoc("v1", new Info
{
Version = "v1",
Title = "Payload View API",
});
});
}

接下来，我们就可以在控制器中通过构造函数注入的方式使用这个服务了。

在API1中，我们创建一个POST方法，这个方法会将API接收到

Payload

[HttpPost]
[ProducesResponseType(typeof(Payload), StatusCodes.Status200OK)]
[ProducesResponseType(typeof(Payload), StatusCodes.Status409Conflict)]
public async Task<IActionResult> Create([FromBody][Required]Payload request)
{
if (data.Any(d => d.Id == request.Id))
{
return Conflict($"data with id {request.Id} already exists");
}

data.Add(request);

// Send this to the bus for the other services
await _serviceBusSender.SendMessage(new MyPayload
{
Goals = request.Goals,
Name = request.Name,
Delete = false
});

return Ok(request);
}

从Queue中获取消息

为了监听Azure Service Bus Queue, 并处理接收到的消息，我们创建了一个新类

ServiceBusConsumer

ServiceBusConsumer

IServiceBusConsumer

Queue的连接字符串是使用

IConfiguration

RegisterOnMessageHandlerAndReceiveMessages

ProcessMessagesAsync

ProcessMessagesAsync

IProcessData

using Microsoft.Azure.ServiceBus;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ServiceBusMessaging
{
public interface IServiceBusConsumer
{
void RegisterOnMessageHandlerAndReceiveMessages();
Task CloseQueueAsync();
}

public class ServiceBusConsumer : IServiceBusConsumer
{
private readonly IProcessData _processData;
private readonly IConfiguration _configuration;
private readonly QueueClient _queueClient;
private const string QUEUE_NAME = "simplequeue";
private readonly ILogger _logger;

public ServiceBusConsumer(IProcessData processData,
IConfiguration configuration,
ILogger<ServiceBusConsumer> logger)
{
_processData = processData;
_configuration = configuration;
_logger = logger;
_queueClient = new QueueClient(
_configuration.GetConnectionString("ServiceBusConnectionString"), QUEUE_NAME);
}

public void RegisterOnMessageHandlerAndReceiveMessages()
{
var messageHandlerOptions = new MessageHandlerOptions(ExceptionReceivedHandler)
{
MaxConcurrentCalls = 1,
AutoComplete = false
};

_queueClient.RegisterMessageHandler(ProcessMessagesAsync, messageHandlerOptions);
}

private async Task ProcessMessagesAsync(Message message, CancellationToken token)
{
var myPayload = JsonConvert.DeserializeObject<MyPayload>(Encoding.UTF8.GetString(message.Body));
_processData.Process(myPayload);
await _queueClient.CompleteAsync(message.SystemProperties.LockToken);
}

private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs exceptionReceivedEventArgs)
{
_logger.LogError(exceptionReceivedEventArgs.Exception, "Message handler encountered an exception");
var context = exceptionReceivedEventArgs.ExceptionReceivedContext;

_logger.LogDebug($"- Endpoint: {context.Endpoint}");
_logger.LogDebug($"- Entity Path: {context.EntityPath}");
_logger.LogDebug($"- Executing Action: {context.Action}");

return Task.CompletedTask;
}

public async Task CloseQueueAsync()
{
await _queueClient.CloseAsync();
}
}
}

其中

IProcessData

ServiceBusMessaging

public interface IProcessData
{
void Process(MyPayload myPayload);
}

在Api 2中，我们创建一个

ProcessData

IProcessData

public class ProcessData : IProcessData
{
public void Process(MyPayload myPayload)
{
DataServiceSimi.Data.Add(new Payload
{
Name = myPayload.Name,
Goals = myPayload.Goals
});
}
}

DataServiceSimi

Payload

ViewPayloadMessagesController

DataServiceSimi

[Route("api/[controller]")]
[ApiController]
public class ViewPayloadMessagesController : ControllerBase
{
[HttpGet]
[ProducesResponseType(StatusCodes.Status200OK)]
public ActionResult<List<Payload>> Get()
{
return Ok(DataServiceSimi.Data);
}
}

最后我们还需要将

ProcessData

public void ConfigureServices(IServiceCollection services)
{
services.AddMvc();

services.AddSingleton<IServiceBusConsumer, ServiceBusConsumer>();
services.AddTransient<IProcessData, ProcessData>();
}

最终效果

现在我们分别启用2个Api项目，并在Api 1的Swagger文档界面，调用POST请求，添加一个Payload

操作完成之后，我们访问Api 2的/api/ViewPayloadMessages, 获得结果如下，Api 1发出的消息出现在了Api 2的结果集中，这说明Api 2从Azure Service Bus Queue中获取了消息，并保存在了自己的静态类

DataServiceSimi