公共交通 - 如何使用 Azure 服务总线计划消息

Question

我已经检查了有关使用 Azure 服务总线进行计划的文档，但我不清楚如何从"断开连接"的总线发送消息。

以下是我配置在服务器上处理消息的服务的方法：

builder.AddMassTransit(mt =>
{
mt.AddConsumers(cqrsAssembly);
mt.AddBus(context => Bus.Factory.CreateUsingAzureServiceBus(x =>
{
x.RequiresSession = true;
x.MaxConcurrentCalls = 500;
x.MessageWaitTimeout = TimeSpan.FromMinutes(5);
x.UseRenewLock(TimeSpan.FromMinutes(4));
x.UseServiceBusMessageScheduler();
var host = x.Host(serviceUri, h =>
{
h.SharedAccessSignature(s =>
{
s.KeyName = "key-name";
s.SharedAccessKey = "access-key";
s.TokenTimeToLive = TimeSpan.FromDays(1);
s.TokenScope = TokenScope.Namespace;
});
h.OperationTimeout = TimeSpan.FromMinutes(2);
});
x.ReceiveEndpoint(host, $"mt.myqueue", ep =>
{
ep.RequiresSession = true;
ep.MaxConcurrentCalls = 500;
ep.RemoveSubscriptions = true;
ep.UseMessageRetry(r =>
{
r.Interval(4, TimeSpan.FromSeconds(30));
r.Handle<TransientCommandException>();
});
ep.ConfigureConsumers(context);
});
});
});

我已经明确称UseServiceBusMessageScheduler().

在创建消息并将其发送到队列的项目中(在不同的上下文中运行，因此"仅发送"(，我们有这个：

var bus = Bus.Factory.CreateUsingAzureServiceBus(x =>
{
x.RequiresSession = true;
x.MessageWaitTimeout = TimeSpan.FromMinutes(5);
x.UseRenewLock(TimeSpan.FromMinutes(4));
x.Send<ICommand>(s => s.UseSessionIdFormatter(ctx => ctx.Message.SessionId ?? Guid.NewGuid().ToString()));

var host = x.Host(serviceUri, h =>
{
h.SharedAccessSignature(s =>
{
s.KeyName = "key-name";
s.SharedAccessKey = "key"; 
s.TokenTimeToLive = TimeSpan.FromDays(1);
s.TokenScope = TokenScope.Namespace;
});
h.OperationTimeout = TimeSpan.FromMinutes(2);
});
EndpointConvention.Map<ICommand>(new Uri($"{serviceUri.ToString()}mt.myqueue"));
EndpointConvention.Map<Command>(new Uri($"{serviceUri.ToString()}mt.myqueue"));
});

现在，要发送计划消息，我们这样做：

var dest = "what?";
await bus.ScheduleSend(dest, scheduledEnqueueTimeUtc.Value, message);

我不确定需要传递到destinationAddress中的内容.

我试过： -serviceUri- '{serviceUri}mt.myqueue">

但是检查队列，我在基本队列、skipped队列或error队列中都没有看到我的消息。

我是否缺少其他配置，如果没有，如何确定目标队列？

我使用的是公共交通的第5.5.4版，每次ScheduleSend()过载都需要它。

Answer 1

首先，是的，您的 Uri 格式是正确的。最后格式化后，您需要这样的东西：

new Uri(@"sb://yourdomain.servicebus.windows.net/yourapp/your_message_queue")

此外，请确保在配置终结点时添加。(见下面的链接(

configurator.UseServiceBusMessageScheduler();

如果您遵循公共交通文档，则计划是从 ConsumptionContext 完成的。请参阅公共交通 Azure 调度

public class ScheduleNotificationConsumer :
IConsumer<AssignSeat>
{
Uri _schedulerAddress;
Uri _notificationService;
public async Task Consume(ConsumeContext<AssignSeat> context)
{
if(context.Message.ReservationTime - DateTime.Now < TimeSpan.FromHours(8))
{
// assign the seat for the reservation
}
else
{
// seats can only be assigned eight hours before the reservation
context.ScheduleMessage(context.Message.ReservationTime - TimeSpan.FromHours(8), context.Message);
}
}
}

然而，在我们本周面临的一个用例中，我们需要从 consumptionContext 之外进行调度，或者只是不想将上下文转发到我们计划的位置。使用IBusControl.ScheduleSend时，我们没有得到错误反馈，但我们也没有真正完成任何调度。 在查看了Mass-Transit的功能之后，事实证明，从IBusControl中，它创建了一个新的调度提供程序。而从上下文中，它使用 ServiceBusScheduleMessageProvider。

因此，在我们清理这一位之前，我们现在正在做的是直接调用ServiceBusScheduleMessageProvider。

await new ServiceBusScheduleMessageProvider(_busControl).ScheduleSend(destinationUri
, scheduleDateTime.UtcDateTime
, Task.FromResult<T>(message)
, Pipe.Empty<SendContext>()
, default);

希望它有意义并有所帮助。