如何在 Azure WebJobs 中修改基于队列消息的连接字符串？

Question

我有一个 Azure WebJob 需要根据它正在处理的队列消息更改它连接到的数据库：

此队列消息需要访问数据库 CustomerOneDb：

{"FormId":1, "DbName":"CustomerOneDb"}

此队列消息需要访问数据库 CustomerTwoDb：

{"FormId":2, "DbName":"CustomerTwoDb"}

我在添加

DBContext

时分配连接字符串，我想让它动态化：

hostBuilder.ConfigureServices(services =>
{
    services.AddDbContext<DbContext.MyDbContext>(
        (serviceProvider, cfg) =>
        {
            if (!cfg.IsConfigured)
            {
                var cnString = "server=localhost;Initial Catalog={DbName};Integrated Security=True;";

                // TODO: Get DB name from queue message
                const string dbName = "CustomerOneDb";

                cnString = cnString.Replace("{DbName}", dbName);

                cfg.UseSqlServer(cnString);
            }
        }
    );
});

我还没有找到任何方法从调用

UseSqlServer

的上下文访问队列消息，所以我正在寻找我can访问队列消息的地方，更新某种变量，然后从里面

AddDbContext

.

失败的尝试 #1：从标准队列处理程序中获取它。

[FunctionName("ProcessQueueMessage")]
public async Task ProcessQueueMessage([QueueTrigger("my-queue")] QueueMessage queueMessage)
{
  // Process queue message
  // This runs *after* the code that calls UseSqlServer, so it's too late
}

失败的尝试#2：通过实施

IQueueProcessorFactory

拦截队列消息：

public class QueueData
{
    public string DatabaseName { get; set; } = string.Empty;
}

public class MyQueueProcessorFactory : Microsoft.Azure.WebJobs.Host.Queues.IQueueProcessorFactory
{
    private readonly IServiceProvider _serviceProvider;

    public MyQueueProcessorFactory(IServiceProvider serviceProvider)
    {
        _serviceProvider = serviceProvider;
    }

    public QueueProcessor Create(QueueProcessorOptions context)
    {
        return new MyQueueProcessor(_serviceProvider, context);
    }
}

public class MyQueueProcessor : QueueProcessor
{
    private readonly IServiceProvider _serviceProvider;

    public MyQueueProcessor(IServiceProvider serviceProvider, QueueProcessorOptions context) : base(context)
    {
        _serviceProvider = serviceProvider;
        Context = context;
    }

    internal QueueProcessorOptions Context { get; private set; }

    protected override Task<bool> BeginProcessingMessageAsync(QueueMessage message, CancellationToken cancellationToken)
    {
        var queueData = _serviceProvider.GetRequiredService<QueueData>();
        queueData.DatabaseName = "CustomerOneDb"; // TODO: Set variable based on value inside queue message

        return base.BeginProcessingMessageAsync(message, cancellationToken);
    }
}

QueueData

是范围服务：

hostBuilder.ConfigureServices(services =>
{
    // ... other config omitted
    services.AddSingleton<IQueueProcessorFactory, MyQueueProcessorFactory>();
    services.AddScoped<QueueData>();
});

函数

BeginProcessingMessageAsync

运行before我调用

UseSqlServer

，这太棒了！但是，当我稍后检索

QueueData

的实例时，它始终是一个空字符串：

services.AddDbContext<DbContext.MyDbContext>(
    (serviceProvider, cfg) =>
    {
        if (!cfg.IsConfigured)
        {
            // ...
            var queueData = serviceProvider.GetRequiredService<QueueData>();
            var dbName = queueData.DatabaseName; // Always an empty string (unless QueueData is configured as a Singleton)
            // ...
        }
    }
);

失败尝试 #3：将

QueueData

更改为单身人士：

services.AddSingleton<QueueData>();

这有效！我可以从

BeginProcessingMessageAsync

分配数据库名称并从

AddDbContext

中检索它。 但是......这不是一个可行的解决方案，因为数据不再局限于触发调用的消息。 也就是说，同时进入的两条消息可能相互竞争以设置/检索数据库名称在同一个实例上。

可能的解决方案： 如果我能实现以下任何一个，我就可以实现基于队列消息中的信息动态设置连接字符串的目标：

• 从内部访问队列消息

  AddDbContext

  .
• 调整 QueueData 的范围，这样当我从

  BeginProcessingMessageAsync

  给它赋值时，它可以稍后在

  AddDbContext

中访问
• 更改 QueueData 以保存由唯一 ID 标识的 DbName 值的集合，稍后我可以从内部查找

  AddDbContext

  。队列消息 ID 可以工作，但无法从内部访问

  AddDbContext

  .

我对如何做这些选项感到困惑，所以我在这里寻求帮助......