مشاركة عبر

مشغل Azure IoT Hub لـAzure Functions

  • مقالة

توضح هذه المقالة كيفية العمل مع روابط Azure Functions من أجل IoT Hub. يعتمد دعم مركز إنترنت الأشياء (IoT Hub) على ربط مراكز الأحداث لـAzure.

للحصول على معلومات حول تفاصيل الإعداد والتكوين، راجع الاستعراض العام.

هام

بينما تستخدم نماذج التعليمات البرمجية التالية واجهة برمجة التطبيقات لمركز الحدث، يُطبق بناء الجملة المُعطى لدالات مركز IoT Hub.

استخدم مشغل الوظيفة للرد على حدث تم إرساله إلى تدفق حدث مركز الحدث. يجب أن يكون لديك حق الوصول للقراءة إلى مركز الأحداث الأساسي لإعداد المشغل. عند تشغيل الوظيفة، تتم كتابة الرسالة التي تم تمريرها إلى الوظيفة كسلسلة.

يتم اتخاذ قرارات تحجيم مراكز الأحداث لخطط الاستهلاك والمميز عبر التحجيم المستند إلى الهدف. لمزيد من المعلومات، راجع التحجيم المستند إلى الهدف.

للحصول على معلومات حول كيفية استجابة Azure Functions للأحداث المرسلة إلى دفق أحداث مركز الأحداث باستخدام المشغلات، راجع دمج مراكز الأحداث مع الوظائف بلا خادم على Azure.

هام

تستخدم هذه المقالة علامات التبويب لدعم إصدارات متعددة من نموذج البرمجة Node.js. يتوفر نموذج v4 بشكل عام وتم تصميمه للحصول على تجربة أكثر مرونة وبديهية لمطوري JavaScript وTypeScript. لمزيد من التفاصيل حول كيفية عمل نموذج v4، راجع دليل مطور Azure Functions Node.js. لمعرفة المزيد حول الاختلافات بين v3 وv4، راجع دليل الترحيل.

تدعم Azure Functions نموذجي برمجة ل Python. تعتمد الطريقة التي تحدد بها روابطك على نموذج البرمجة الذي اخترته.

يتيح لك نموذج برمجة Python v2 تحديد الروابط باستخدام المحسنات مباشرة في التعليمات البرمجية لدالة Python. لمزيد من المعلومات، راجع دليل مطور Python.

تدعم هذه المقالة كلا نموذجي البرمجة.

مثال

المثال التالي يوضح دالة C#‎ التي يتم تشغيلها استنادًا إلى مركز الأحداث، إذ تتم كتابة سلسلة رسالة الإدخال إلى السجلات:

{
    private readonly ILogger<EventHubsFunction> _logger;

    public EventHubsFunction(ILogger<EventHubsFunction> logger)
    {
        _logger = logger;
    }

    [Function(nameof(EventHubFunction))]
    [FixedDelayRetry(5, "00:00:10")]
    [EventHubOutput("dest", Connection = "EventHubConnection")]
    public string EventHubFunction(
        [EventHubTrigger("src", Connection = "EventHubConnection")] string[] input,
        FunctionContext context)
    {
        _logger.LogInformation("First Event Hubs triggered message: {msg}", input[0]);

        var message = $"Output message created at {DateTime.Now}";
        return message;
    }

يوضح المثال التالي دالة TypeScript لمشغل مراكز الأحداث. تقرأ الدالة بيانات التعريف للحدث وتعمل على تسجيل الرسالة.

import { app, InvocationContext } from '@azure/functions';

export async function eventHubTrigger1(message: unknown, context: InvocationContext): Promise<void> {
    context.log('Event hub function processed message:', message);
    context.log('EnqueuedTimeUtc =', context.triggerMetadata.enqueuedTimeUtc);
    context.log('SequenceNumber =', context.triggerMetadata.sequenceNumber);
    context.log('Offset =', context.triggerMetadata.offset);
}

app.eventHub('eventHubTrigger1', {
    connection: 'myEventHubReadConnectionAppSetting',
    eventHubName: 'MyEventHub',
    cardinality: 'one',
    handler: eventHubTrigger1,
});

لتلقي الأحداث في دفعة، قم بتعيين cardinality إلى many، كما هو موضح في المثال التالي.

import { app, InvocationContext } from '@azure/functions';

export async function eventHubTrigger1(messages: unknown[], context: InvocationContext): Promise<void> {
    context.log(`Event hub function processed ${messages.length} messages`);
    for (let i = 0; i < messages.length; i++) {
        context.log('Event hub message:', messages[i]);
        context.log(`EnqueuedTimeUtc = ${context.triggerMetadata.enqueuedTimeUtcArray[i]}`);
        context.log(`SequenceNumber = ${context.triggerMetadata.sequenceNumberArray[i]}`);
        context.log(`Offset = ${context.triggerMetadata.offsetArray[i]}`);
    }
}

app.eventHub('eventHubTrigger1', {
    connection: 'myEventHubReadConnectionAppSetting',
    eventHubName: 'MyEventHub',
    cardinality: 'many',
    handler: eventHubTrigger1,
});

يوضح المثال التالي دالة JavaScript مشغل مراكز الأحداث. تقرأ الدالة بيانات التعريف للحدث وتعمل على تسجيل الرسالة.

const { app } = require('@azure/functions');

app.eventHub('eventHubTrigger1', {
    connection: 'myEventHubReadConnectionAppSetting',
    eventHubName: 'MyEventHub',
    cardinality: 'one',
    handler: (message, context) => {
        context.log('Event hub function processed message:', message);
        context.log('EnqueuedTimeUtc =', context.triggerMetadata.enqueuedTimeUtc);
        context.log('SequenceNumber =', context.triggerMetadata.sequenceNumber);
        context.log('Offset =', context.triggerMetadata.offset);
    },
});

لتلقي الأحداث في دفعة، قم بتعيين cardinality إلى many، كما هو موضح في المثال التالي.

const { app } = require('@azure/functions');

app.eventHub('eventHubTrigger1', {
    connection: 'myEventHubReadConnectionAppSetting',
    eventHubName: 'MyEventHub',
    cardinality: 'many',
    handler: (messages, context) => {
        context.log(`Event hub function processed ${messages.length} messages`);
        for (let i = 0; i < messages.length; i++) {
            context.log('Event hub message:', messages[i]);
            context.log(`EnqueuedTimeUtc = ${context.triggerMetadata.enqueuedTimeUtcArray[i]}`);
            context.log(`SequenceNumber = ${context.triggerMetadata.sequenceNumberArray[i]}`);
            context.log(`Offset = ${context.triggerMetadata.offsetArray[i]}`);
        }
    },
});

هنا رمز PowerShell:

param($eventHubMessages, $TriggerMetadata)

Write-Host "PowerShell eventhub trigger function called for message array: $eventHubMessages"

$eventHubMessages | ForEach-Object { Write-Host "Processed message: $_" }

يوضح المثال التالي ربط مشغل مراكز الأحداث ودالة Python التي تستخدم الربط. تقرأ الدالة بيانات التعريف للحدث وتعمل على تسجيل الرسالة. يعتمد المثال على ما إذا كنت تستخدم نموذج برمجة v1 أو v2 Python.

import logging
import azure.functions as func

app = func.FunctionApp()

@app.function_name(name="EventHubTrigger1")
@app.event_hub_message_trigger(arg_name="myhub", 
                               event_hub_name="<EVENT_HUB_NAME>",
                               connection="<CONNECTION_SETTING>") 
def test_function(myhub: func.EventHubEvent):
    logging.info('Python EventHub trigger processed an event: %s',
                myhub.get_body().decode('utf-8'))

المثال التالي يوضح ربط مشغل مراكز الأحداث الذي يسجل نص الرسالة لمشغل مراكز الأحداث.

@FunctionName("ehprocessor")
public void eventHubProcessor(
  @EventHubTrigger(name = "msg",
                  eventHubName = "myeventhubname",
                  connection = "myconnvarname") String message,
       final ExecutionContext context )
       {
          context.getLogger().info(message);
 }

في مكتبة وقت تشغيل وظائف Java، استخدم التعليق التوضيحي EventHubTrigger على المعلمات التي تأتي قيمتها من مركز الأحداث. تؤدي المعلمات مع هذه التعليقات التوضيحية إلى تشغيل الدالة عند وصول حدث. يمكن استخدام هذا التعليق التوضيحي مع أنواع Java الأصلية أو POJOs أو القيم االخالية Optional<T>.

يوضح المثال التالي الاستخدام المكثف لـ SystemProperties وخيارات الربط الأخرى لمزيد من التعمق في الحدث إلى جانب توفير مسار BlobOutput مكوّن بشكل جيد يتسم بالتسلسل الهرمي للتاريخ.

package com.example;
import java.util.Map;
import java.time.ZonedDateTime;

import com.microsoft.azure.functions.annotation.*;
import com.microsoft.azure.functions.*;

/**
 * Azure Functions with Event Hub trigger.
 * and Blob Output using date in path along with message partition ID
 * and message sequence number from EventHub Trigger Properties
 */
public class EventHubReceiver {

    @FunctionName("EventHubReceiver")
    @StorageAccount("bloboutput")

    public void run(
            @EventHubTrigger(name = "message",
                eventHubName = "%eventhub%",
                consumerGroup = "%consumergroup%",
                connection = "eventhubconnection",
                cardinality = Cardinality.ONE)
            String message,

            final ExecutionContext context,

            @BindingName("Properties") Map<String, Object> properties,
            @BindingName("SystemProperties") Map<String, Object> systemProperties,
            @BindingName("PartitionContext") Map<String, Object> partitionContext,
            @BindingName("EnqueuedTimeUtc") Object enqueuedTimeUtc,

            @BlobOutput(
                name = "outputItem",
                path = "iotevents/{datetime:yy}/{datetime:MM}/{datetime:dd}/{datetime:HH}/" +
                       "{datetime:mm}/{PartitionContext.PartitionId}/{SystemProperties.SequenceNumber}.json")
            OutputBinding<String> outputItem) {

        var et = ZonedDateTime.parse(enqueuedTimeUtc + "Z"); // needed as the UTC time presented does not have a TZ
                                                             // indicator
        context.getLogger().info("Event hub message received: " + message + ", properties: " + properties);
        context.getLogger().info("Properties: " + properties);
        context.getLogger().info("System Properties: " + systemProperties);
        context.getLogger().info("partitionContext: " + partitionContext);
        context.getLogger().info("EnqueuedTimeUtc: " + et);

        outputItem.setValue(message);
    }
}

السمات

تستخدم كل من مكتبات المعالجة والعامل المعزول C# السمة لتكوين المشغل. يستخدم البرنامج النصي C# بدلا من ذلك ملف تكوين function.json كما هو موضح في دليل البرمجة النصية C#‎.

استخدم EventHubTriggerAttribute لتعريف مشغل على مركز الأحداث، والذي يدعم الخصائص التالية.

المعلمات ‏‏الوصف
EventHubName اسم مركز الأحداث. عندما يكون اسم مركز الأحداث موجودًا أيضًا في سلسلة الاتصال، هذه القيمة تتجاوز هذه الخاصية في وقت التشغيل. يمكن الرجوع إليها في إعدادات التطبيق، مثل %eventHubName%
مجموعة المستهلكين خاصية اختيارية تعمل على تعيين مجموعة المستهلكين المستخدمة للاشتراك في الأحداث في المركز. عند حذفها، يتم استخدام مجموعة المستهلكين $Default.
Connection اسم إعداد التطبيق أو مجموعة الإعدادات التي تحدد طريقة الاتصال بمراكز الأحداث. لمعرفة المزيد، راجع الاتصالات.

الديكور

ينطبق فقط على نموذج برمجة Python v2.

بالنسبة لوظائف Python v2 المعرفة باستخدام مصمم الديكور، الخصائص التالية على cosmos_db_trigger:

الخاصية ‏‏الوصف
arg_name اسم المتغير الذي يمثل عنصر الحدث في التعليمات البرمجية للوظيفة.
event_hub_name اسم مركز الأحداث. عندما يكون اسم مركز الأحداث موجودًا أيضًا في سلسلة الاتصال، هذه القيمة تتجاوز هذه الخاصية في وقت التشغيل.
connection اسم إعداد التطبيق أو مجموعة الإعدادات التي تحدد طريقة الاتصال بمراكز الأحداث. راجع الاتصالات.

للحصول على وظائف Python المعرفة باستخدام function.json، راجع قسم التكوين.

تعليقات توضيحية

في مكتبة وقت تشغيل دوال Java، استخدم التعليق التوضيحي EventHubTrigger، الذي يدعم الإعدادات التالية:

التكوين

ينطبق فقط على نموذج برمجة Python v1.

يوضح الجدول التالي الخصائص التي يمكنك تعيينها على الكائن الذي options تم تمريره app.eventHub() إلى الأسلوب .

الخاصية ‏‏الوصف
eventHubName اسم مركز الأحداث. عندما يكون اسم مركز الأحداث موجودًا أيضًا في سلسلة الاتصال، هذه القيمة تتجاوز هذه الخاصية في وقت التشغيل. يمكن الرجوع إليها عبر إعدادات التطبيق%eventHubName%
consumerGroup خاصية اختيارية تعمل على تعيين مجموعة المستهلكين المستخدمة للاشتراك في الأحداث في المركز. عند حذفها، يتم استخدام مجموعة المستهلكين $Default.
cardinality قم بالتعيين على many لتمكين التجميع. إذا تم حذفها أو تعيينها إلى one، سيتم تمرير رسالة واحدة إلى الوظيفة.
الاتصال اسم إعداد التطبيق أو مجموعة الإعدادات التي تحدد طريقة الاتصال بمراكز الأحداث. راجع الاتصالات.

يوضح الجدول التالي خصائص تكوين المشغل التي قمت بتعيينها في ملف function.json، والذي يختلف باختلاف إصدار وقت التشغيل.

خاصية function.json ‏‏الوصف
النوع يجب تعيينه إلى eventHubTrigger. يتم تعيين هذه الخاصية تلقائيا عند إنشاء المشغل في مدخل Microsoft Azure.
الاتجاه يجب تعيينه إلى in. يتم تعيين هذه الخاصية تلقائيا عند إنشاء المشغل في مدخل Microsoft Azure.
الاسم اسم المتغير الذي يمثل عنصر الحدث في التعليمات البرمجية للوظيفة.
eventHubName اسم مركز الأحداث. عندما يكون اسم مركز الأحداث موجودًا أيضًا في سلسلة الاتصال، هذه القيمة تتجاوز هذه الخاصية في وقت التشغيل. يمكن الرجوع إليها عبر إعدادات التطبيق%eventHubName%
consumerGroup خاصية اختيارية تعمل على تعيين مجموعة المستهلكين المستخدمة للاشتراك في الأحداث في المركز. عند حذفها، يتم استخدام مجموعة المستهلكين $Default.
cardinality قم بالتعيين على many لتمكين التجميع. إذا تم حذفها أو تعيينها إلى one، سيتم تمرير رسالة واحدة إلى الوظيفة.
الاتصال اسم إعداد التطبيق أو مجموعة الإعدادات التي تحدد طريقة الاتصال بمراكز الأحداث. راجع الاتصالات.

عندما تقوم بالتطوير محليًا، أضف إعدادات التطبيق في ملف local.settings.json في المجموعة Values.

الاستخدام

لمعرفة المزيد حول كيفية تشغيل مراكز الأحداث ومقاييس مشغل IoT Hub، راجع استهلاك الأحداث باستخدام Azure Functions.

يعتمد نوع المعلمة الذي يدعمه ربط مخرجات مراكز الأحداث على إصدار وقت تشغيل الدوال وإصدار حزمة الامتداد ونمط C# المستخدم.

عندما تريد أن تعالج الدالة حدثا واحدا، يمكن ربط مشغل مراكز الأحداث بالأنواع التالية:

النوع ‏‏الوصف
string الحدث كسلسلة. استخدم عندما يكون الحدث نصا بسيطا.
byte[] وحدات بايت الحدث.
‏‏أنواع قابلة للتسلسل إلى JSON عندما يحتوي حدث على بيانات JSON، تحاول الدالات إلغاء تسلسل بيانات JSON إلى نوع كائن CLR (POCO) قديم عادي.
Azure.Messaging.EventHubs.EventData1 كائن الحدث.
إذا كنت تقوم بالترحيل من أي إصدارات قديمة من Event Hubs SDKs، فلاحظ أن هذا الإصدار يسقط الدعم للنوع القديم Body لصالح EventBody.

عندما تريد أن تعالج الدالة دفعة من الأحداث، يمكن أن يرتبط مشغل مراكز الأحداث بالأنواع التالية:

النوع ‏‏الوصف
string[] صفيف من الأحداث من الدفعة، كسلاسل. يمثل كل إدخال حدثا واحدا.
EventData[]1 صفيف من الأحداث من الدفعة، كمثيلات Azure.Messaging.EventHubs.EventData. يمثل كل إدخال حدثا واحدا.
T[] حيث T هو نوعJSON قابل للتسلسل 1 صفيف من الأحداث من الدفعة، كمثيلات لنوع POCO مخصص. يمثل كل إدخال حدثا واحدا.

1 لاستخدام هذه الأنواع، تحتاج إلى الرجوع إلى Microsoft.Azure.Functions.Worker.Extensions.EventHubs 5.5.0 أو أحدث والتبعيات الشائعة لروابط نوع SDK.

يمكن أن يكون نوع المعلمة أحد الأنواع التالية:

  • أي أنواع Java أصلية مثل int، String، byte [ ].
  • القيم التي تقبل البيانات الفارغة باستخدام اختياري.
  • أي نوع POJO.

لمعرفة المزيد، اطلع على مرجع EventHubTrigger.

بيانات تعريف الحدث

يوفر مشغل مراكز الأحداث العديد من خصائص بيانات التعريف. يمكن استخدام بيانات التعريف كجزء من تعبيرات الربط في روابط أخرى أو معلمات في التعليمات البرمجية الخاصة بك. تأتي الخصائص من فئة EventData.

الخاصية نوع ‏‏الوصف
PartitionContext PartitionContext المثيل PartitionContext.
EnqueuedTimeUtc DateTime الوقت المحدد في UTC.
Offset string إزاحة البيانات المتعلقة بتدفق قسم مركز الأحداث. تعد الإزاحة هي علامة أو معرف لحدث داخل دفق مراكز الأحداث. يعد المعرف فريدًا داخل قسم من دفق مراكز الأحداث.
PartitionKey string القسم الذي يجب إرسال بيانات الحدث إليه.
Properties IDictionary<String,Object> خصائص المستخدم لبيانات الحدث.
SequenceNumber Int64 رقم التسلسل المنطقي للحدث.
SystemProperties IDictionary<String,Object> خصائص النظام، بما في ذلك بيانات الحدث.

راجع أمثلة التعليمات البرمجية التي تستخدم هذه الخصائص في هذه المقالة.

الاتصالات

الخاصية connection هي مرجع إلى تكوين البيئة الذي يحتوي على اسم إعداد تطبيق يحتوي على سلسلة اتصال. يمكنك الحصول على سلسلة الاتصال هذه عن طريق تحديد الزر معلومات الاتصال لـ مساحة الاسم. يجب أن تكون سلسلة الاتصال لمساحة "مراكز الأحداث"، وليس مركز الحدث نفسها.

يجب أن تحتوي سلسلة الاتصال على أذونات "قراءة" على الأقل لتنشيط الدالة.

يجب تخزين سلسلة الاتصال هذه في إعداد تطبيق باسم يطابق القيمة المحددة بواسطة خاصية connection لتكوين الربط.

إشعار

مشغل IoT Hub لا يدعم الاتصالات المستندة إلى الهوية. إذا كنت بحاجة إلى استخدام الهويات المدارة من طرف إلى طرف، فيمكنك بدلًا من ذلك استخدام توجيه مركز IoT لإرسال البيانات إلى مركز أحداث تتحكم فيه. وبهذه الطريقة، يمكن مصادقة التوجيه الصادر بهوية مدارة يمكن قراءة الحدث من مركز الحدث هذا باستخدام الهوية المدارة.

خصائص مضيف.json

يحتوي ملف host.json على إعدادات تتحكم في سلوك مشغل Event Hub. راجع إعدادات مضيف.json قسم للحصول على تفاصيل حول الإعدادات المتوفرة.

الخطوات التالية