پیکربندی مشتری

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

محاصره کردن = نیاز('kafkajs') // مشتری را با لیست کارگزار ایجاد کنید محاصره کردنکافکا =جدیدکافکا (<شناسه مشتری: "برنامه من", کارگزار: ['kafka1: 9092', 'kafka2: 9092']>) 

شناسه منطقی یک برنامه. می توان توسط کارگزاران برای اعمال سهمیه یا درخواست ردیابی در یک برنامه خاص استفاده کرد. مثال: رزرو-فرآیند.

مستندات Kafka ClientId را به عنوان:

Client-ID یک گروه بندی منطقی از مشتریان با نام معنی دار است که توسط برنامه مشتری انتخاب شده است. Tuple (کاربر ، مشتری-ID) یک گروه منطقی ایمن از مشتریانی را تعریف می کند که هم اصلی کاربر و هم مشتری-مشتری را به اشتراک می گذارد. سهمیه ها را می توان برای گروه های (کاربر ، مشتری-ID) ، کاربر یا مشتری-ID استفاده کرد.

شناسه مشتری

یک رشته شناسه برای انتقال به سرور هنگام درخواست. هدف از این امر این است که بتوانیم با اجازه دادن به یک نام برنامه منطقی در ورود به سیستم درخواست سرور ، منبع درخواست های فراتر از IP/Port را ردیابی کنیم.

بنابراین ClientID باید در چندین مورد در یک خوشه یا برنامه مقیاس افقی به اشتراک گذاشته شود ، اما برای هر برنامه متمایز است.

به طور معمول Kafkajs متوجه تغییرات توپولوژی خوشه ای کارگزار به طور خودکار می شود و واکنش نشان می دهد ، اما در برخی شرایط ممکن است بخواهید به جای استفاده از یک لیست پیکربندی شده ، بتوانید به صورت پویا کارگزاران بذر را واکشی کنید. در این حالت ، کارگزاران را می توان روی یک عملکرد ASYNC تنظیم کرد که به یک آرایه کارگزار برطرف می شود:

محاصره کردنکافکا =جدیدکافکا (<شناسه مشتری: "برنامه من", کارگزار: درز () => <// به عنوان مثال گرفتن کارگزاران از پروکسی استراحت مخلوط محاصره کردنCLUSTERRESPONSE =در انتظاررفتن و آوردن('https: // kafka-rest: 8082/v3/خوشه', <عناوین: 'برنامه/vnd. api+json',>). سپس(واکنش =>پاسخ. json ())محاصره کردنCLUSTERURL = CLUSTERRESPONSE. DATA [0] . links. selfمحاصره کردنBrokersResponse =در انتظاررفتن و آوردن(`$ /کارگزاران, <عناوین: 'برنامه/vnd. api+json',>). سپس(واکنش =>پاسخ. json ())محاصره کردنکارگزاران = brokersresponse. data. map (دلال => <محاصره کردن = broker. attributesبرگشت `$ :$ `>) برگشت brokers>>) 

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

از گزینه SSL می توان برای پیکربندی سوکت های TLS استفاده کرد. گزینه ها مستقیماً به TLS. Connect منتقل می شوند و برای ایجاد زمینه امن TLS استفاده می شوند ، همه گزینه ها پذیرفته می شوند.

محاصره کردنfs =نیاز('fs') جدیدکافکا (<شناسه مشتری: "برنامه من", کارگزار: ['kafka1: 9092', 'kafka2: 9092'], SSL: <رد کردن: دروغ, ca: [fs. readfilesync ('/my/custom/ca. crt', "UTF-8")], کلید: fs. readfilesync ('/my/custom/client-key. pem', "UTF-8"), گواهی: fs. readfilesync ('/my/custom/client-cert. pem', "UTF-8")>,>) 

برای اطلاعات بیشتر به TLS ایجاد زمینه ایمن مراجعه کنید. node_extra_ca_certs می تواند برای افزودن CAS سفارشی استفاده شود. از SSL استفاده کنید: اگر تنظیمات اضافی ندارید و می خواهید SSL را فعال کنید.

کافکا از استفاده از SASL برای تأیید اعتبار مشتری پشتیبانی می کند. از گزینه SASL می توان برای پیکربندی مکانیسم احراز هویت استفاده کرد. در حال حاضر ، Kafkajs از مکانیسم های Plain ، Scram-Sha-256 ، Scram-Sha-512 و AWS پشتیبانی می کند.

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

جدیدکافکا (<شناسه مشتری: "برنامه من", کارگزار: ['kafka1: 9092', 'kafka2: 9092'], // تأیید اعتبار: 10000 ، // مجدداً آستانه: 10000 ، SSL: درست است، واقعی, ساس: <سازوکار: 'جلگه', // scram-sha-256 یا scram-sha-512 نام کاربری: "نام تجاری من", کلمه عبور: 'رمز عبور من'>,>) 

جدیدکافکا (<شناسه مشتری: "برنامه من", کارگزار: ['kafka1: 9092', 'kafka2: 9092'], // تأیید اعتبار: 10000 ، // مجدداً آستانه: 10000 ، SSL: درست است، واقعی, ساس: <سازوکار: 'oauthbearer', پیش نویس: درز () => <// از یک نشانه ناامن استفاده کنید. محاصره کردنtoken = jwt. sign (<زیر: 'تست'>, 'ABC', <الگوریتم: 'هیچ یک'>) //یا واقع بینانه تر ، نشانه را از برخی از نقطه پایانی OAUTH بگیرید برگشت <ارزش: token>>>,>) 

شیء SASL باید شامل یک ویژگی به نام OAuthBeareRerprovider ، یک تابع Async باشد که برای بازگرداندن توکن حامل OAuth استفاده می شود.

توکن حامل OAUTH باید یک شی با ارزش خواص و پسوندهای (اختیاری) باشد که در طول درخواست SASL/OAuthbearer ارسال می شود.

اجرای OAutheareReprovider باید مراقبت کند که در صورت لزوم از نشانه ها استفاده مجدد و تازه سازی می شوند. اجرای مثال با استفاده از ساده-OAUTH2 چیزی شبیه به موارد زیر است:

وارد كردن از جانب 'Simple-oauth2' رابطoauthbearererprovideroptionsرشته؛ClientSecret:رشته؛میزبان:رشته؛مسیر:رشته؛refreshresholdms:عدد;> محاصره کردنoauthbearererprovider =(گزینه ها: oauthbearererprovideroptions) => <محاصره کردنمشتری =جدیدClientCredentials (، auth:>); اجازه دهیدTokenPromise:وعدهرشته>; اجازه دهیدAccessToken: AccessToken ؛درز تابع تازه کار( ) <تلاش كردن <if(AccessToken ==خالی) در انتظارclient. gettoken (<>)> if(AccessToken. Expired (گزینه ها. refreshresholdms /1000)) در انتظار accessToken.refresh()> محاصره کردنnextrefresh = AccessToken. token. expires_in *1000- گزینه ها. settimeout (() => ، nextrefresh) ؛برگشت accessToken.token.access_token;> گرفتن(خطا)خالی; پرت كردن error;>>tokenpromise = refreshtoken () ؛برگشت درز تابع ( ) <برگشت در انتظار tokenPromise>>>; محاصره کردنکافکا =جدیدکافکا (<//سایر گزینه های مورد نیازSASL:'oauthbearer'، oauthbearerprovider: oauthbearererprovider ("OAUTH-CLIENT-ID"، ClientSecret:'Oauth-Client-Secret'، میزبان:'https://my-oauth-server. com'، مسیر:'/oauth/token', // قبل از انقضا ، 15 ثانیه توکن را تازه کنیدrefreshreshold:15000,>),>,>) 

مثال AWS IAM

جدیدکافکا (<شناسه مشتری: "برنامه من", کارگزار: ['kafka1: 9092', 'kafka2: 9092'], // تأیید اعتبار: 10000 ، // مجدداً آستانه: 10000 ، SSL: درست است، واقعی, ساس: <سازوکار: 'AWS', مجوز: "Aidaiosfodnn7example", // userid یا نقش عکسبرداری: 'akiaiosfodnn7example', مخفی: "Wjalrxutnfemi/K7mdeng/bpxrficyexamplekey", جلسه: "Wharyt8i5vfquriu5zbmlcbjcaiv/eww6el9tgqmjp6qfnexampletoken" // اختیاری>,>) 

برای کسب اطلاعات بیشتر در مورد مبانی اعتبار و احراز هویت IAM ، به صفحه اعتبار امنیتی AWS مراجعه کنید - به صفحه کلید دسترسی.

استفاده از این قابلیت نیاز به kafka aws iam loginmodule stack دارد ، یا یک جایگزین سازگار برای نصب بر روی همه کارگزاران هدف است.

در مثال فوق ، مجوز باید AWS باشد: کاربر از هویت AWS IAM. به طور معمول ، می توانید این مقدار را با استفاده از دستورات دریافت کننده AWS IAM یا AWS IAM Get-Role از ابزار AWS CLI بازیابی کنید. AWS: UserID معمولاً به عنوان خاصیت کاربر یا نقش اصلی پاسخ ذکر می شود.

همچنین می توانید AWS را به صورت برنامه ای بازیابی کنید: UserID برای اعتبار موجود در حال حاضر با سرویس امنیتی AWS SDK.

یک شکست کامل را می توان در مرجع راهنمای کاربر IAM در مورد متغیرهای خط مشی یافت.

از پروتکل های رمزگذاری شده استفاده کنید

بسیار توصیه می شود که هنگام استفاده از ساده یا AWS از SSL برای رمزگذاری استفاده کنید ، در غیر این صورت اعتبارنامه ها در ClearText منتقل می شوند!

مکانیسم های احراز هویت سفارشی

اگر مکانیسم تأیید اعتبار از جعبه در Kafkajs پشتیبانی نشود ، می توان مکانیسم تأیید اعتبار سفارشی را به عنوان افزونه معرفی کرد:

<ساس: <سازوکار: ، AuthenticationProvider: () => Promise>>> 

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

زمان در میلی ثانیه برای انتظار برای اتصال موفق. مقدار پیش فرض: 1000 است.

جدیدکافکا (<شناسه مشتری: "برنامه من", کارگزار: ['kafka1: 9092', 'kafka2: 9092'], زمان اتصال: 3000>) 

زمان در میلی ثانیه برای انتظار برای یک درخواست موفق. مقدار پیش فرض: 30000 است.

جدیدکافکا (<شناسه مشتری: "برنامه من", کارگزار: ['kafka1: 9092', 'kafka2: 9092'], درخواست زمان: 25000>) 

زمان درخواست را می توان با تنظیم ForforceRequestTimeout به False غیرفعال کرد.

جدیدکافکا (<شناسه مشتری: "برنامه من", کارگزار: ['kafka1: 9092', 'kafka2: 9092'], اجرای برنامه: دروغ>) 

از گزینه آزمایش مجدد می توان برای تنظیم پیکربندی مکانیسم آزمایش مجدد استفاده کرد ، که برای امتحان مجدد اتصالات و تماس های API به کافکا (هنگام استفاده از تولید کنندگان یا مصرف کنندگان) استفاده می شود.

مکانیسم آزمایش مجدد از یک تابع تصادفی استفاده می کند که به صورت نمایی رشد می کند. مثال دقیق

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

گزینه های موجود:

جدیدکافکا (<شناسه مشتری: "برنامه من", کارگزار: ['kafka1: 9092', 'kafka2: 9092'], مجدداً: <زمان اولیه: 100, مجدداً: 8>>) 

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

The function has the following signature: (error: Error) =>وعده

• اگر این قول به حقیقت برطرف شود: مصرف کننده مجدداً راه اندازی می شود
• اگر قول به دروغ برسد: مصرف کننده مجدداً راه اندازی نمی شود
• اگر قول رد شود: مصرف کننده مجدداً راه اندازی می شود
• در صورت عدم وجود راه اندازی مجدد ارائه نشده است: مصرف کننده مجدداً راه اندازی می شود

توجه داشته باشید که این عملکرد فقط برای آنچه Kafkajs خطاهای بازپس گیری می داند ، فراخوانی می شود. در مورد خطاهای غیر قابل بازگشت ، مصرف کننده مجدداً راه اندازی نمی شود و از عملکرد راه اندازی مجدد استفاده نمی شود. برای خطاهای بازپس گیری در پروتکل Kafka به این لیست مراجعه کنید ، اما توجه داشته باشید که برخی از خطاهای اضافی هنوز هم در کافکا مجذوب می شوند ، مانند مثال خطاهای اتصال به شبکه.

kafkajs دارای یک logger stdout داخلی است که از JSON خارج می شود. همچنین یک سازنده ورود به سیستم سفارشی را می پذیرد که به شما امکان می دهد کتابخانه Logger مورد علاقه خود را ادغام کنید. 5 سطح ورود به سیستم در دسترس است: هیچ چیز ، خطا ، هشدار ، اطلاعات و اشکال زدایی. اطلاعات به طور پیش فرض پیکربندی شده است.

محاصره کردن = نیاز('kafkajs') محاصره کردنکافکا =جدیدکافکا (<شناسه مشتری: "برنامه من", کارگزار: ['kafka1: 9092', 'kafka2: 9092'], ورود به سیستم: logLevel.ERROR>) 

برای غلبه بر سطح ورود به سیستم بعد از لحظه ، با SetLoglevel در Logger فردی تماس بگیرید.

محاصره کردن = نیاز('kafkajs') محاصره کردنکافکا =جدیدکافکا (<شناسه مشتری: "برنامه من", کارگزار: ['kafka1: 9092', 'kafka2: 9092'], ورود به سیستم: logLevel.ERROR>) kafka. logger (). setloglevel (loglevel. wa)محاصره کردنتولید کننده = kafka. producer (.) producer. logger (). setloglevel (loglevel. info)محاصره کردنمصرف کننده = kafka. consumer (.) consumer. logger (). setLoglevel (loglevel. debug)محاصره کردنمدیر = kafka. admin (.) Admin. Logger (). setLoglevel (loglevel. nothing) 

متغیر محیط KAFKAJS_LOG_LEVEL نیز می تواند مورد استفاده قرار گیرد و بر پیکربندی کد ، به عنوان مثال:

kafkajs_log_level = کد گره اطلاعات. js 

توجه: برای کسب اطلاعات بیشتر در مورد نحوه سفارشی کردن سیاهههای مربوط ، به ورود به سیستم سفارشی نگاهی بیندازید

کارخانه سوکت سفارشی

برای امکان تنظیم تنظیمات سوکت سفارشی ، مشتری یک خاصیت اختیاری سوکت عملکردی را که برای ساخت هر سوکت استفاده می شود ، می پذیرد.

SocketFactory باید تابعی باشد که یک شی سازگار با net. socket را بازگرداند (به اجرای پیش فرض مراجعه کنید).

محاصره کردن = نیاز('kafkajs') // مثال کارخانه سوکت تنظیم TTL سفارشی محاصره کردنخالص =نیاز('خالص') محاصره کردنtls =نیاز('tls') محاصره کردنmycustomsocketfactory =( ) => <محاصره کردنسوکت = SSL؟tls. connect (هدف - شی. Assign (، SSL) ، onConnect): net. connect (، onConnect) socket. setKeepalive (درست است، واقعی, 30000) برگشت socket> محاصره کردنکافکا =جدیدکافکا (<شناسه مشتری: "برنامه من", کارگزار: ['kafka1: 9092', 'kafka2: 9092'], پربار: myCustomSocketFactory,>) 

پشتیبانی از ترافیک پروکسی می تواند با یک کارخانه سوکت سفارشی پیاده سازی شود. در مثال زیر ما از Proxy Chain برای ادغام با سرور پروکسی استفاده می کنیم ، اما تا زمانی که رابط کارخانه سوکت اجرا شود ، می توان از هر کتابخانه پروکسی دیگر استفاده کرد.

محاصره کردنtls =نیاز('tls') محاصره کردنخالص =نیاز('خالص') محاصره کردن = نیاز("زنجیره پروکسی") محاصره کردنsocketfective =( ) => <محاصره کردنسوکت = SSL؟جدیدtls. tlssocket ():جدیدnet. socket () createTunnel (process. env. http_proxy ،`$ :$ `) . سپس((تونلد) => <محاصره کردن[Tunnelhost ، Tunnelport] = tunneladdress. split (':') socket. setKeepalive (درست است، واقعی, 60000) Socket. Connect (هدف - شی. Assign (<میزبان: Tunnelhost ،بندر: تونلپورت ،نام ارائهکننده: host>، ssl) ، onConnect) socket. on ('بستن', () => درست است، واقعی)>)>) . catch (خطا =>Socket. emit ("خطا"، خطا))برگشت socket>