具有 Net Core Web API 的 Azure 事件中心

Question

我是 Azure 事件中心的新手。我的要求是我想从将发送的第三方 Azure 事件中心之一进行监听(我只有第三方事件中心的连接字符串和我自己的存储帐户)如果之后在第三方发生某些特定事件，则向我更新在我的应用程序中，我需要检查数据库中有多少用户已订阅该事件并需要在手机中向他们发送通知(通过 Firebase SDK)。那么如何我可以在 Net Core Web API() 中实现该行为吗？

我了解到我们可以将 webhook 与 Azure 事件中心结合使用，但没有找到任何好的资源，如果您有任何资源，请与我分享。

我还了解到我们可以使用 WebJob，但没有找到任何代码资源，如果您有任何资源，请与我分享。

如果您对我有任何帮助，请先致谢。

Answer 1

您需要将 Firebase Cloud Messaging 集成到您的项目中，并使用它向用户的手机发送通知。

• 实现逻辑，根据收到的事件在数据库中查询订阅用户，并使用 Firebase SDK 发送通知。

• 您需要安装 Firebase Cloud Messaging (FCM) 用于发送通知。dotnet 添加包 FirebaseAdmin

• 通过添加第三方 Azure 事件中心的连接字符串来配置您的 appsettings.json 文件。

{
  "AzureEventHubConnectionString": "your_connection_string_here",
  "FirebaseConfigFilePath": "firebase_admin_sdk.json"
}

• firebase_admin_sdk.json 您可以通过在 Firebase 项目中添加/创建示例 Web 应用程序来获取此应用程序，导航到项目设置>服务帐户，您可以在其中看到生成新的私钥。

点击generate就可以看到下载到本地的json文件了。在 appsettings.json 中给出文件路径

• 此处，我启用服务代理来生成对我的应用程序的主键访问权限。

• 下面是 EventProcessor 类，它可以监听来自 Azure 事件中心的事件并对其进行处理，包括在收到特定事件时通过 Firebase Cloud Messaging 发送通知。

using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Consumer;
using FirebaseAdmin;
using FirebaseAdmin.Messaging;
using Google.Apis.Auth.OAuth2;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using System;
using System.Text;
using System.Threading.Tasks;

public class EventProcessor
{
    private readonly IConfiguration _configuration;
    private readonly ILogger<EventProcessor> _logger;
    private readonly FirebaseApp _firebaseApp;

    public EventProcessor(IConfiguration configuration, ILogger<EventProcessor> logger)
    {
        _configuration = configuration;
        _logger = logger;

        // Initialize Firebase Admin SDK
        _firebaseApp = FirebaseApp.Create(new AppOptions
        {
            Credential = GoogleCredential.FromFile(_configuration["FirebaseConfigFilePath"])
        });
    }

    public async Task StartProcessing()
    {
        string eventHubConnectionString = _configuration["AzureEventHubConnectionString"];
        string eventHubName = "traileventhub382";

        await using (var consumerClient = new EventHubConsumerClient(EventHubConsumerClient.DefaultConsumerGroupName, eventHubConnectionString, eventHubName))
        {
            await foreach (PartitionEvent partitionEvent in consumerClient.ReadEventsAsync())
            {
                try
                {
                    // Process the incoming event data here
                    string eventData = Encoding.UTF8.GetString(partitionEvent.Data.Body.ToArray());
                    _logger.LogInformation($"Received event: {eventData}");

                    // Example: Send a notification using Firebase Cloud Messaging
                    var message = new Message
                    {
                        Notification = new Notification
                        {
                            Title = "Notification Title",
                            Body = "Notification Body"
                        },
                        // Add necessary targeting options here
                    };

                    var response = await FirebaseMessaging.DefaultInstance.SendAsync(message);
                    _logger.LogInformation($"Notification sent: {response}");
                }
                catch (Exception ex)
                {
                    _logger.LogError($"Error processing event: {ex}");
                }
            }
        }
    }
}

在这里，我能够监听 EventHub 监控指标中捕获的消息。

结果:

BackgroundService.cs:

using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System;
using System.Threading;
using System.Threading.Tasks;

public class EventHubBackgroundService : BackgroundService
{
    private readonly IServiceProvider _serviceProvider;
    private readonly ILogger<EventHubBackgroundService> _logger;

    public EventHubBackgroundService(IServiceProvider serviceProvider, ILogger<EventHubBackgroundService> logger)
    {
        _serviceProvider = serviceProvider;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            using (var scope = _serviceProvider.CreateScope())
            {
                var eventProcessor = scope.ServiceProvider.GetRequiredService<EventProcessor>();
                await eventProcessor.StartProcessing();
            }

            // Adjust the frequency at which you want to check for new events
            await Task.Delay(TimeSpan.FromMinutes(5), stoppingToken);
        }
    }
}

• 在这里，我创建了一个范围并解析了 EventProcessor 以定期调用 StartProcessing() 方法。您可以根据需要调整延迟。