ساخت جریان های کاری رویداد محور با DynamoDB Streams

Summarize this content to 400 words in Persian Lang
معماری های مبتنی بر رویداد به ما این امکان را می دهند که کدهای پیچیده و خوانا را به اجزای قابل مدیریت تر تجزیه کنیم. با استفاده از جریان‌های DynamoDB و توابع سبک لامبدا، می‌توانیم منابع جفت شده‌ای ایجاد کنیم که به طور خودکار به رویدادها پاسخ می‌دهند، گردش کار را ساده‌سازی می‌کنند و وضوح سیستم را بهبود می‌بخشند.

1. سناریو

باب در حال توسعه یک برنامه بازی است که در آن کاربران برای انجام وظایف امتیاز کسب می کنند. هر کار بر اساس سختی طبقه‌بندی می‌شود و به بازیکنان این امکان را می‌دهد تا بر اساس سطح چالش امتیازات متفاوتی کسب کنند.

منطق باطن این امتیازات را به جدول DynamoDB اضافه می کند. باب تصمیم گرفت یک طرح تک جدولی را برای کل برنامه پیاده سازی کند.

امتیازها از طریق یک درخواست POST به نقطه پایانی API Gateway REST API ارسال می‌شوند. یک تابع Lambda بار بار را پردازش کرده و در پایگاه داده ذخیره می کند. از طرف دیگر، اگر بار به اعتبارسنجی یا بازسازی نیاز نداشته باشد، API Gateway می‌تواند مستقیماً به DynamoDB متصل شود – راه‌حلی سریع‌تر، اما نه همیشه قابل اجرا ارائه می‌دهد.

در هر دو مورد، امتیاز جدید باید به امتیاز کل کاربر اضافه شود.

2. رویکردها

باب دو گزینه اصلی برای مقابله با این چالش دارد.

2.1. راه حل سنکرون

در این رویکرد، باب یک را اضافه می کند ادغام لامبدا به نقطه پایانی API Gateway. تابع امتیاز جدید را بر اساس نوع وظیفه ارائه شده در درخواست محاسبه می کند. همچنین ممکن است نیاز باشد که نگاشت امتیاز کار مربوطه را از پایگاه داده واکشی کند. سپس، این تابع امتیاز کل موجود کاربر را از پایگاه داده بازیابی می‌کند، امتیاز جدید را اضافه می‌کند و امتیاز کل به‌روز شده را در جدول ذخیره می‌کند.

این الگو معمولا استفاده می شود. یکی از مزیت ها سادگی آن است: یک تابع تمام عملیات لازم را انجام می دهد و معماری را ساده نگه می دارد. تنها چیزی که لازم است نوشتن و استقرار کد در Lambda است.

با این حال، در حالی که معماری را ساده می کنیم، پیچیدگی کد را افزایش می دهیم. عملکرد لامبدا به سرعت می تواند سخت شود، زیرا مسئولیت های متعددی را بر عهده می گیرد. علاوه بر این، پردازش همزمان می تواند پاسخ به مشتری را به تأخیر بیندازد زیرا منطق تجاری باید قبل از بازگشت نتیجه تکمیل شود. در نهایت، در برخی از سناریوها، مشتری لزوماً نیازی به دریافت نتیجه، یعنی نمره کل به روز شده در پاسخ ندارد.

2.2. راه حل ناهمزمان

از طرف دیگر، باب می تواند طراحی کند ناهمزمان راه حل از آنجایی که مشتری به نمره کل جدید در پاسخ نیاز ندارد، نقطه پایانی می تواند به سادگی یک را برگرداند تصدیق که درخواست دریافت شد. تمام پردازش های بیشتر – محاسبات و به روز رسانی پایگاه داده – می توانند در پس زمینه انجام شوند.

این رویکرد به مشتری این امکان را می دهد که بدون منتظر ماندن برای اتمام کل فرآیند، یک پاسخ فوری دریافت کند. تنها وظیفه تابع کنترل نقطه پایان اضافه کردن امتیاز جدید به پایگاه داده است. در برخی موارد، API Gateway حتی می‌تواند مستقیماً در DynamoDB بنویسد و نیاز به Lambda را دور بزند.

سپس منطق کسب و کار را می توان در توابع جداگانه ای که از مشتری پنهان شده است تقسیم کرد. این کد هر تابع لامبدا را کوتاه و متمرکز نگه می‌دارد، زیرا هر کدام یک کار مشخص و کاملاً تعریف شده را انجام می‌دهند و ایجاد می‌کنند. کوپلینگ شل بین بخشهای مختلف منطق

اشکال اصلی رویکرد ناهمزمان، پیچیدگی معماری اضافه شده است. باب نیاز به راه اندازی و مدیریت منابع اضافی دارد که به زمان، توجه به طراحی و نگهداری بیشتری نیاز دارد.

باب در نهایت تصمیم گرفت که با او همراه شود ناهمزمان راه حل برای این نرم افزار

3. راه حل ناهمزمان انتخاب شده

بیایید نحوه تنظیم یک گردش کار ناهمزمان برای پردازش امتیاز را بررسی کنیم.

3.1. DynamoDB Streams

DynamoDB Streams هر کدام را می گیرد در سطح مورد تغییرات (ایجاد/به روز رسانی/حذف) در جدول DynamoDB.

این ویژگی دقیقاً همان چیزی است که ما به آن نیاز داریم: به ما امکان می‌دهد تشخیص دهیم که چه زمانی یک آیتم امتیاز جدید به جدول اضافه می‌شود و عملکردی را برای پردازش امتیاز فعال می‌کند.

فعال کردن جریان در DynamoDB با استفاده از CDK ساده است:

const table = new dynamodb.Table(scope, ‘SOME_ID’, {
// … other properties
stream: dynamodb.StreamViewType.NEW_IMAGE,
});

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

با این پیکربندی، تابع Lambda که جریان را مصرف می کند، آیتم کامل را همانطور که پس از اصلاح ظاهر می شود دریافت می کند.

3.2. اجتناب از بازگشت

با این حال، یک چالش وجود دارد: از آنجایی که ما از طراحی تک جدولی استفاده می کنیم، آیتم امتیاز جدید در همان جدول با نمره کل ذخیره می شود.

هنگامی که یک امتیاز جدید نوشته می‌شود، به جریان اضافه می‌شود و عملکرد لامبدا را برای پردازش امتیاز و به‌روزرسانی کل امتیاز در همان جدول فعال می‌کند. این به روز رسانی می تواند عملکرد را دوباره فعال کند و به طور بالقوه باعث ایجاد یک خطای حلقه یا عملکرد شود.

برای جلوگیری از این امر، نگاشت منبع رویداد را طوری پیکربندی می‌کنیم که فقط برای رویدادهای خاص فعال شود، مانند زمانی که یک آیتم امتیاز جدید ایجاد می‌شود. برای جلوگیری از بازگشت، رویدادهایی مانند به‌روزرسانی‌های امتیاز کل را حذف می‌کنیم.

3.3. طراحی میز مناسب با فیلتر

فیلترها را می توان برای ویژگی های مختلف و با استفاده از یکی از آنها اعمال کرد کلیدهای فهرست در پیکربندی فیلتر، راه اندازی را آسان تر می کند. طراحی دقیق جدول می تواند این فرآیند فیلتر را ساده کند.

در طراحی تک میز استفاده می کنیم کلیدهای پارتیشن و مرتب سازی عمومی، اغلب برچسب گذاری می شود PK و SK. در اینجا، کلید پارتیشن (PK) می تواند نوع مورد را مشخص کند. برای مثال، کلید پارتیشن هر آیتم امتیازی می تواند با آن شروع شود SCORE#…. قسمت بعد # مربوط به این مثال نیست.

ما می توانیم پیکربندی کنیم فیلتر مانند این در نگاشت منبع رویداد با استفاده از TypeScript CDK:

const scoreProcessorFn = new NodejsFunction(
// function properties here
);

scoreProcessorFn.addEventSource(
new lambdaEventSources.DynamoEventSource(table, { // Table construct from above
filters: [
lambda.FilterCriteria.filter({
eventName: lambda.FilterRule.isEqual(‘INSERT’),
dynamodb: {
Keys: {
PK: { S: lambda.FilterRule.beginsWith(‘SCORE’) },
},
},
}),
],
// other configuration properties here
}),
);

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

این تنظیمات تضمین می کند که هر بار یک آیتم جدید با یک کلید پارتیشن (PK) با شروع SCORE اضافه می شود (INSERT، تابع پردازش اجرا می شود. مواردی که کلید پارتیشن با آنها شروع نمی شود SCORE عملکرد را فعال نمی کند.

اگر به این تابع نیاز دارید که روی هر دو اجرا شود INSERT و MODIFY رویدادها، می توانید فیلتر را به صورت زیر پیکربندی کنید:

scoreProcessorFn.addEventSource(
new lambdaEventSources.DynamoEventSource(table, { // Table construct from above
filters: [
lambda.FilterCriteria.filter({
eventName: lambda.FilterRule.isEqual(‘INSERT’),
dynamodb: {
Keys: {
PK: { S: lambda.FilterRule.beginsWith(‘SCORE’) },
},
},
}),
lambda.FilterCriteria.filter({
eventName: lambda.FilterRule.isEqual(‘MODIFY’),
dynamodb: {
Keys: {
PK: { S: lambda.FilterRule.beginsWith(‘SCORE’) },
},
},
}),
],
// other configuration properties here
}),
);

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

با استفاده از فیلترهای مشابه، می‌توانیم سایر توابع لامبدا را پیکربندی کنیم تا فقط زمانی که موارد خاصی ایجاد یا به‌روزرسانی می‌شوند اجرا شوند و از حلقه‌های بازگشتی در معماری ما جلوگیری کنیم.

3.4. پردازش کد تابع

وقتی Lambda رکوردهای جدیدی را در جریان تشخیص می دهد، تابع را با یک فراخوانی می کند رویداد شی به عنوان اول استدلال شما می توانید یک نمونه قالب رویداد را در مستندات مشاهده کنید.

در اینجا نمونه ای از کد عملکرد پردازنده در TypeScript آمده است:

export async function handler(event: DynamoDBStreamEvent, context: Context) {
// Assuming that the batch size is set to 1 in the event source mapping, we can access the only record in the batch using its index.
// If the batch size is set to a number greater than 1, you should iterate over the records to process them all.
const score = event.Records[0].dynamodb?.NewImage;

const [, uniqueId] = score.PK.S?.split(‘#’) || [];
const [, otherId] = score.SK.S?.split(‘#’) || [];
const scoreValue = score.Score.N;
// other necessary attributes here

// processing logic here
}

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

یک نکته مهم: حتی اگر از ماژول سرویس گیرنده سند از AWS SDK استفاده کنید تا از تعیین آن خودداری کنید توصیفگرهای نوع داده (S، N، BOOL، و غیره) در کد، رکورد جریان همچنان حاوی آنها خواهد بود. هنگام بازیابی مقادیر ویژگی، حتماً این توصیفگرها را در مسیر دسترسی قرار دهید.

4. خلاصه

با استفاده از DynamoDB Streams و توابع Lambda، می‌توانیم جریان‌های کاری مبتنی بر رویداد را با هم‌پیوندی آزاد ایجاد کنیم. این رویکرد نیاز به منابع اضافی برای ایجاد و مدیریت دارد، اما منجر به کد تمیزتر و قابل نگهداری تر در هر تابع می شود.

برای جلوگیری از حلقه‌ها در معماری، می‌توانیم فیلترهایی را در نگاشت منبع رویداد تنظیم کنیم. این فیلترها تضمین می‌کنند که عملکرد تنها زمانی فعال می‌شود که رکورد استریم شرایط خاصی را برآورده می‌کند و جریان پردازش رویداد را ساده می‌کند.

5. مراجع، مطالعه بیشتر

ایجاد طرح تک جدول با Amazon DynamoDB – پست وبلاگ AWS در مورد طراحی تک جدول

اولین تابع Lambda خود را ایجاد کنید – نحوه ایجاد یک تابع Lambda

ایجاد جداول و بارگذاری داده برای نمونه کد در DynamoDB – نحوه ایجاد و پر کردن جداول DynamoDB

const table = new dynamodb.Table(scope, 'SOME_ID', {
  // ... other properties
 stream: dynamodb.StreamViewType.NEW_IMAGE,
});

const scoreProcessorFn = new NodejsFunction(
  // function properties here
);

scoreProcessorFn.addEventSource(
  new lambdaEventSources.DynamoEventSource(table, { // Table construct from above
    filters: [
      lambda.FilterCriteria.filter({
        eventName: lambda.FilterRule.isEqual('INSERT'),
        dynamodb: {
          Keys: {
            PK: { S: lambda.FilterRule.beginsWith('SCORE') },
          },
        },
      }),
    ],
    // other configuration properties here
  }),
);

scoreProcessorFn.addEventSource(
  new lambdaEventSources.DynamoEventSource(table, { // Table construct from above
    filters: [
      lambda.FilterCriteria.filter({
        eventName: lambda.FilterRule.isEqual('INSERT'),
        dynamodb: {
          Keys: {
            PK: { S: lambda.FilterRule.beginsWith('SCORE') },
          },
        },
      }),
      lambda.FilterCriteria.filter({
        eventName: lambda.FilterRule.isEqual('MODIFY'),
        dynamodb: {
          Keys: {
            PK: { S: lambda.FilterRule.beginsWith('SCORE') },
          },
        },
      }),
    ],
    // other configuration properties here
  }),
);

export async function handler(event: DynamoDBStreamEvent, context: Context) {
  // Assuming that the batch size is set to 1 in the event source mapping, we can access the only record in the batch using its index.
  // If the batch size is set to a number greater than 1, you should iterate over the records to process them all.
  const score = event.Records[0].dynamodb?.NewImage;

  const [, uniqueId] = score.PK.S?.split('#') || [];
  const [, otherId] = score.SK.S?.split('#') || [];
  const scoreValue = score.Score.N;
  // other necessary attributes here

  // processing logic here
}