با سری مقالات ادغام Hyperledger Fabric با Spring Boot در خدمت شما هستیم. در این مقاله، Kafka را بررسی خواهیم کرد. همچنین، نحوه استقرار Kafka در Kubernetes نیز توضیح خواهم داده خواهد شد.
سایر مقالات در مورد ادغام Hyperledger Fabric با Spring Boot از لینک های زیر قابل دسترسی هستند.
بخش 1 – مقدمه
قسمت 2 – راه اندازی خوشه Kubernetes
قسمت 3 – Fabric CA Server
بخش 4 – تولید گواهینامه ها و مصنوعات
ابتدا به اختصار Kafka را توضیح می دهم.
Kafka
آپاچی کافکا یک چارچوب متن باز برای ذخیره سازی و تجزیه و تحلیل فوری داده های بزرگ است که توسط لینکدین توسعه یافته و اکنون بخشی از آپاچی است.
از سیستم پیام رسانی (queue) برای ذخیره و تجزیه و تحلیل سریع داده های بزرگ استفاده می کند.
گره های خدمات سفارش Hyperledger fabric(OSN) از خوشه Kafka شما استفاده می کنند و یک سرویس سفارش را به شبکه بلاک چین شما ارائه می دهند.
بیایید مفاهیم موضوع، تولید کننده و مصرف کننده را در Kafka بررسی کنیم.
Topic
Topics یک ناحیه قابل نام گذاری توسط کاربر است که در آن داده ها (پیام ها) ذخیره می شود. موضوعات به پارتیشن‌ها تقسیم می‌شوند و تعداد پارتیشن‌هایی که در آنها ذخیره می‌شوند را کاربر می‌تواند تعیین کند.
Producer
از سوی دیگر، وقتی به ساختار Publish-Subscribe آپاچی کافکا نگاه می کنیم، تولیدکنندگان ناشر هستند، که می تواند به این موضوعات پیام ارسال کند. آنها می توانند داده ها، به عنوان مثال، پیام ها را به موضوعات ارسال کنند و می توانند همزمان به بیش از یک موضوع مرتبط شوند.
Consumer
از طرف دیگر، مصرف کنندگان، مشترکینی هستند که پیام های ارسال شده توسط تولیدکنندگان به موضوعات را مصرف می کنند، همانطور که از نام آن متوجه می شویم. بیش از یک تولیدکننده می توانند به یک موضوع (موضوع) پیام ارسال کنند، و بیش از یک مصرف کننده می توانند در یک موضوع گنجانده شوند و داده های ارسال شده به موضوع را بخوانند. پس از اینکه مصرف کننده این پیام های ارسال شده توسط تولیدکنندگان را خواند، این داده ها از موضوع حذف نمی شوند.
Zookeeper
Apache ZooKeeper یک پروژه منبع باز آپاچی است که به خوشه ها اجازه می دهد تا اطلاعاتی مانند پیکربندی، نام گذاری و سرویس های گروهی را روی خوشه های بزرگ توزیع کنند.
ZooKeeper از یک فروشگاه با ارزش کلیدی به صورت سلسله مراتبی استفاده می کند. برای محیط های در دسترس بالا استفاده می شود. Apache ZooKeeper به زبان جاوا نوشته شده و تحت مجوز Apache 2.0 مجوز دارد. این توسط برخی از شرکت های بزرگ مانند Rackspace، Yahoo، eBay و Reddit استفاده می شود.
Zookeeper وضعیت گره های خوشه کافکا را پیگیری می کند و همچنین موضوعات کافکا، پارتیشن ها و غیره را پیگیری می کند.
نصب کافکا در Kubernetes
برای پروژه انتقال دارایی، کافکا و zookeeper را برای اجرای 2 غلاف در kubernetes راه اندازی می کنیم.
بیایید پروژه ای را که از این لینک دانلود کرده ایم باز کنیم و به دایرکتوری که k8s در آن قرار دارد برویم.
$ cd deploy/k8s
حجم پایداری Kafka و Zookeeper
اول از همه، بیایید یک حجم پایدار برای Kafka و Zookeeper ایجاد کنیم. فایل های yaml که آن را ایجاد می کنند در پوشه زیر در پروژه زیر قرار دارند.
deploy/k8s/pv/kafka-pv.yaml
deploy/k8s/pv/zookeeper-pv.yaml
برای Kafka ؛
nfs:
path: /srv/kubedata/fabricfiles/broker/kafka1
server: 192.168.12.9
مسیری را مشخص می کند که داده های دائمی کافکا در سرور nfs و ip سرور nfs نگهداری می شود.
برای Zookeper
nfs:
path: /srv/kubedata/fabricfiles/broker/zookeeper0
server: 192.168.12.9
مسیری را مشخص می کند که داده های دائمی Zookoper در سرور nfs و ip سرور nfs نگهداری می شود.
spec:
storageClassName: default
volumeMode: Filesystem
capacity:
storage: 1Gi
ظرفیت ذخیره‌سازی محدود است و بسته به گره‌ای که یک پاد روی آن اجرا می‌شود، ممکن است متفاوت باشد: فضای ذخیره‌سازی متصل به شبکه ممکن است برای همه گره‌ها قابل دسترسی نباشد، یا ذخیره‌سازی برای شروع به یک گره محلی است. به عنوان ۱ گیگابایت اختصاص داده می‌شود.
metadata:
name: kafka1-pv
labels:
app: kafka
podindex: "0"
یک برچسب پاددکس برای پیوست کردن حجم ماندگاری با ادعاهای مربوط به حجم ماندگاری اختصاص داده شد.
metadata:
name: kafka1-pv
labels:
app: kafka
podindex: "1"
podindex: “1” نشان دهنده داده های پایدار کافکای دوم است.
metadata:
name: zookeeper1-pv
labels:
app: zookeeper
podindex: "0"
podindex: “0” نشان دهنده داده های پایدار اولین zookeeper است.podindex: “1” نشان دهنده داده های پایدارzookeeper دوم است.
ادعای حجم پایداری kafka وzookeeper
بیایید یک ادعای حجم پایدار برای کافکا و zookeeper ایجاد کنیم. فایل های yaml که آن را ایجاد می کنند در پوشه زیر در پروژه زیر قرار دارند.
deploy/k8s/pvc/kafka-pvc.yaml
deploy/k8s/pvc/zookeeper-pvc.yaml
selector:
matchLabels:
app: kafka
podindex: "0"
ادعاها می‌توانند یک انتخابگر برچسب برای فیلتر بیشتر مجموعه حجم‌ها مشخص کنند. فقط جلدهایی که برچسب‌های آنها با انتخابگر مطابقت دارد، می‌توانند به جلد ماندگاری ادعای kafka متصل شوند، باید دارای برچسب “podindex: “0” باشند.
selector:
matchLabels:
app: zookeeper
podindex: "0"
به همین ترتیب، حجم تداوم zookeeper باید دارای پاددکس باشد: برچسب “0”.
resources:
requests:
storage: 1Gi
ادعاها، مانند Pods، می توانند مقادیر خاصی از یک منبع را درخواست کنند. در این مورد، درخواست برای ذخیره سازی است. به عنوان 1GB اختصاص داده شده است.
Kafka and Zookeeper StatefulSet
بیایید یک مجموعه حالت برای Kafka وZookeeper ایجاد کنیم. StatefulSet شیء API حجم کاری است که برای مدیریت برنامه‌های حالت دار استفاده می‌شود. استقرار و مقیاس‌بندی مجموعه‌ای از Pods را مدیریت می‌کند و تضمین‌هایی در مورد ترتیب و منحصربه‌فرد بودن این پادها ارائه می‌دهد.
فایل های yaml که آن را ایجاد می کنند در پوشه زیر در پروژه زیر قرار دارند.
deploy/k8s/kafka/kafka.yaml
deploy/k8s/kafka/zookeeper.yaml
apiVersion: apps/v1
kind: StatefulSet
metadata:
name: broker
labels:
app: kafka
StatefulSet، به نام broker.
apiVersion: apps/v1
kind: StatefulSet
metadata:
name: zoo
labels:
app: zookeeper
StatefulSet با نام zoo
spec:
selector:
matchLabels:
app: kafka
serviceName: broker
replicas: 2
StatefulSet دارای مشخصاتی است که نشان می دهد 2 کپی از ظرف کافکا در Pods منحصر به فرد عرضه می شود.
selector:
matchLabels:
app: zookeeper
serviceName: zoo
replicas: 2
به همین ترتیب، StatefulSet دارای مشخصاتی است که نشان می دهد 2 نمونه از کانتینر zookeeper در Pods منحصر به فرد راه اندازی خواهند شد.
volumeClaimTemplates:
- metadata:
name: kafka
VolumeClaimTemplates با استفاده از PersistentVolumes ارائه شده توسط PersistentVolume Provisioner، فضای ذخیره‌سازی پایداری را فراهم می‌کند. باید با نام فراداده pvc یکسان باشد.
volumeClaimTemplates:
- metadata:
name: zookeeper
به همین ترتیب، باید با نام ابرداده pvc یکسان باشد.
متغیرهای محیطی زیر به کافکا اختصاص داده شده است.
env:
- name: KAFKA_MESSAGE_MAX_BYTES
value: "102760448"
- name: KAFKA_REPLICA_FETCH_MAX_BYTES
value: "102760448"
- name: KAFKA_UNCLEAN_LEADER_ELECTION_ENABLE
value: "false"
- name: KAFKA_DEFAULT_REPLICATION_FACTOR
value: "2"
- name: KAFKA_MIN_INSYNC_REPLICAS
value: "2"
- name: KAFKA_ZOOKEEPER_CONNECT
value: zoo-0.zoo:2181,zoo-1.zoo:2181
- name: KAFKA_PORT
value: "9092"
- name: GODEBUG
value: netdns=go
- name: KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS
value: "30000"
- name: KAFKA_LOG_DIRS
value: /opt/kafka/data
KAFKA_MESSAGE_MAX_BYTES: حداکثر اندازه ارسال پیام.
KAFKA_REPLICA_FETCH_MAX_BYTES: حداکثر تعداد اولیه بایت در هر موضوع+پارتیشن برای درخواست هنگام واکشی پیام از کارگزار.
KAFKA_MIN_INSYNC_REPLICAS: زمانی استفاده می شود که مشکلی در موضوع وجود داشته باشد، شاید یکی از پارتیشن ها همزمان یا آفلاین نباشد. در این حالت، وقتی KAFKA_MIN_INSYNC_REPLICAS راضی شد، خوشه یک آک ارسال می کند. بنابراین 2 کپی با KAFKA_MIN_INSYNC_REPLICAS=2 همچنان قادر به نوشتن خواهند بود.
KAFKA_ZOOKEEPER_CONNECT: به کافکا آموزش می دهد که چگونه با ZooKeeper ارتباط برقرار کند.
zoo: نام سرویس Zookeeper
2181: پورت خدمات Zookeeper
KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: حداکثر زمانی که مشتری در هنگام برقراری ارتباط با Zookeeper منتظر می ماند
KAFKA_DEFAULT_REPLICATION_FACTOR: ضریب تکرار برای مبحث ابرداده ردیف (برای اطمینان از در دسترس بودن، بالاتر تنظیم کنید).
متغیرهای محیطی زیر برای Zookeeper اختصاص داده شده است.
env:
- name: ZOO_SERVERS
value: server.0=zoo-0.zoo.default.svc.cluster.local:2888:3888 server.1=zoo-1.zoo.default.svc.cluster.local:2888:3888
- name: ZOO_4LW_COMMANDS_WHITELIST
value: srvr, mntr, ruok
- name: ZOO_MAX_SESSION_TIMEOUT
value: "40000"
- name: ZOO_TICK_TIME
value: "2000"
ZOO_SERVERS: این متغیر به شما امکان می دهد لیستی از ماشین های مجموعه Zookeeper را مشخص کنید. هر ورودی باید به این صورت مشخص شود: server.id=<address1>:<port1>:<port2>
ZOO_4LW_COMMANDS_WHITELIST: فهرستی از دستورات چهار حرفی جدا شده با کاما که کاربر می خواهد استفاده کند. یک دستور چهار حرفی معتبر باید در این لیست قرار داده شود، در غیر این صورت سرور ZooKeeper دستور را فعال نمی کند. پیش فرض ها برای srvr
ZOO_MAX_SESSION_TIMEOUT: حداکثر زمان پایان جلسه در میلی ثانیه که سرور به مشتری اجازه می دهد تا با آن مذاکره کند.
ZOO_TICK_TIME: واحد زمان اصلی بر حسب میلی ثانیه که توسط Apache ZooKeeper برای ضربان قلب استفاده می شود
ports:
- name: broker
containerPort: 9092
کافکا به یک پورت گوش می دهد: 9092 پورت پیش فرض استفاده شده توسط کافکا است.
ports:
- name: client
containerPort: 2181
- name: peer
containerPort: 2888
- name: leader-election
containerPort: 3888
Zookeeper به سه پورت گوش می دهد: 2181 برای اتصالات مشتری. 2888 برای اتصالات فالوور، اگر رهبر باشند. و 3888 برای سایر اتصالات سرور در مرحله انتخاب رهبر .
خدمات Kafka و Zookeeper
بیایید یک سرویس برای Kafka و Zookeeper ایجاد کنیم.
فایل های yaml که آن را ایجاد می کنند در پوشه زیر در پروژه زیر قرار دارند.
deploy/k8s/kafka/kafka-svc.yaml
deploy/k8s/kafka/zookeeper-svc.yaml
apiVersion: v1
kind: Service
metadata:
name: kafka
labels:
app: kafka
این مشخصات یک شیء سرویس جدید به نام “کافکا” ایجاد می کند.
apiVersion: v1
kind: Service
metadata:
name: broker
این مشخصات یک شیء هدلس سرویس جدید به نام «کارگزار» ایجاد می‌کند. سرویس بدون سر، سرویسی است با IP سرویس، اما به جای متعادل کردن بار، آی‌پی‌های Pods مرتبط ما را برمی‌گرداند. این به ما امکان می‌دهد به‌جای پروکسی مستقیماً با Pods تعامل داشته باشیم. برای پیکربندی کافکا در حالت خوشه‌ای لازم است.
ports:
- name: "broker"
targetPort: 9092
port: 9092
targetPort: کانتینر پورت. پورت پیش فرض استفاده شده توسط کافکا 9092 است.
پورت: پورت سرویس kubernetes.
apiVersion: v1
kind: Service
metadata:
name: zookeeper
labels:
app: zookeeper
این مشخصات یک شیء سرویس جدید به نام “zookeeper” ایجاد می کند.
apiVersion: v1
kind: Service
metadata:
name: zoo
spec:
type: ClusterIP
clusterIP: None
selector:
app: zookeeper
این مشخصات یک شیء سرویس بدون سر جدید به نام “zoo” ایجاد می کند. برای پیکربندی Zookeeper در حالت خوشه ای لازم است.
ports:
- name: "peer"
targetPort: 2888
port: 2888
- name: "leader-election"
targetPort: 3888
port: 3888
پورت های پیش فرض انتخابی همتا و رهبر zookeeper از طریق سرویس هدلس در معرض دید قرار می گیرند.
ports:
- name: client
protocol: TCP
targetPort: 2181
port: 2181
پورت پیش‌فرض مشتری zookeeper از طریق سرویس zookeeperدر معرض دید قرار می‌گیرد.
Kafka و Zookeeper را در Kubernetes مستقر کنید
بیایید با دستور Vagrant ssh به ماشین مجازی kubernetes master node متصل شویم.
$ vagrant ssh k8smaster
بیایید به دایرکتوری برویم که اسکریپت های نصب kubernetes در آن قرار دارند. این دایرکتوری همان پوشه deploy/k8s در پروژه است. با Vagrant، این دایرکتوری با ماشین مجازی همگام سازی می شود.
$ cd /vagrant/k8s
استقرار حجم پایداری برای zookeeper و kafka
$ kubectl apply -f pv/kafka-pv.yaml
$ kubectl apply -f pv/zookeeper-pv.yaml
پس از ایجاد pv، اجازه دهید pvc را ایجاد کنیم.
به کارگیری ادعای حجم پایدار برای zookeeper و kafka
$ kubectl apply -f pvc/kafka-pvc.yaml
$ kubectl apply -f pvc/zookeeper-pvc.yaml
استقرار statefulset برای zookeeper و kafka
$ kubectl apply -f kafka/
ایجاد Kafka و zookeeper در انتظار تکمیل است.
$ kubectl wait --for condition=ready --timeout=300s pod -l "app in (zookeeper,kafka)"
Kafka و zookeeper با موفقیت ایجاد کردند.
در نهایت، بیایید شرایط غلاف هایی را که از ایده لنز اجرا می کنیم، بررسی کنیم.
به نظر می رسد وضعیت Kafka و غلاف های zookeeper در حال اجراست.
به طور کلی به معرفی کافکا و Zookeeper پرداختیم همچنین نحوه استقرار این ابزارها در Kubernetes نیز توضیح داده شد.
