大型门户网站建设哪专业,亦庄建设局网站,泸州市住房和城乡建设厅官方网站,网络营销是什么工作职位不许转载
kafka 部署 把 kafka 部署到 k8s 后#xff0c;我们肯定是通过 service 从 k8s 外部访问 kafaka。这里的 service 要么是 NodePort#xff0c; 要么是 LoadBalancer 类型。我们使用的方式是 LoadBalancer。 我们先看下面这张图#xff0c;这是 kafka 在集群中的网…不许转载
kafka 部署 把 kafka 部署到 k8s 后我们肯定是通过 service 从 k8s 外部访问 kafaka。这里的 service 要么是 NodePort 要么是 LoadBalancer 类型。我们使用的方式是 LoadBalancer。 我们先看下面这张图这是 kafka 在集群中的网络拓扑。当我们通过地址 12.345.67:31090 访问到 kafka 后kafka 返回的访问地址是类似这样的 endpoint jettopro-kafka.jettopro-poc.svc.cluster.local:9092。这是 k8s 集群内部能访问的 headless service endpoint 地址从集群外部自然使用这个地址是访问不通的。 所以我们需要解决两个问题
kafka 不同的 pod 需要不通的对外能访问的地址不能使用 kafka 默认的 advertised.listeners
解决方案
问题1我们为每个 pod 创建类型是 LoadBalancer 的 service 并且监听不同的端口。这样通过 LB IP port 就能找到特定的 kafka broker。 service 示例如下
apiVersion: v1
kind: Service
metadata:name: kafka-{index}
spec:externalTrafficPolicy: Localtype: LoadBalancerselector:statefulset.kubernetes.io/pod-name: kafka-{index}ports:- protocol: TCPport: {9092index}targetPort: 9092这里如果不是云主机也可以使用NodePort类型来暴露kafka服务。
问题2我们在容器启动的时候执行脚本动态获取 pod 编号生成容器需要的环境变量 KAFKA_CFG_ADVERTISED_LISTENERS对应 kafka broker 的配置 advertised.listener
HOSTNAMEhostname -s
if [[ $HOSTNAME ~ (.*)-([0-9])$ ]]; thenORD${BASH_REMATCH[2]}PORT$((ORD 9092))#12.345.67.8 是 LB 的 ipexport KAFKA_CFG_ADVERTISED_LISTENERSPLAINTEXT://12.345.67.8:$PORT
elseecho Failed to get index from hostname $HOSTexit 1
fiapiVersion: apps/v1
kind: StatefulSet
metadata:name: kafka
spec:selector:matchLabels:app: kafkaserviceName: kafkareplicas: 3updateStrategy:type: RollingUpdatepodManagementPolicy: OrderedReadytemplate:metadata:labels:app: kafkaspec:affinity:podAntiAffinity:requiredDuringSchedulingIgnoredDuringExecution:- labelSelector:matchExpressions:- key: appoperator: Invalues:- kafkatopologyKey: kubernetes.io/hostnamecontainers:- name: kafkacommand:- bash- -ec- |HOSTNAMEhostname -sif [[ $HOSTNAME ~ (.*)-([0-9])$ ]]; thenORD${BASH_REMATCH[2]}PORT$((ORD 9092))export KAFKA_CFG_ADVERTISED_LISTENERSPLAINTEXT://12.345.67.8:$PORTelseecho Failed to get index from hostname $HOSTexit 1fiexec /entrypoint.sh /run.shimagePullPolicy: Alwaysimage: bitnami/kafka:2env:- name: ALLOW_PLAINTEXT_LISTENERvalue: yes- name: KAFKA_CFG_ZOOKEEPER_CONNECTvalue: zookeeper-0.zookeeper-hs:2181,zookeeper-1.zookeeper-hs:2181,zookeeper-2.zookeeper-hs:2181- name: KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTORvalue: 3- name: KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISRvalue: 3- name: KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTORvalue: 3ports:- containerPort: 9092volumeMounts:- name: kafka-datamountPath: /bitnamisecurityContext:runAsUser: 1000fsGroup: 1000volumeClaimTemplates:- metadata:name: kafka-dataspec:accessModes: [ ReadWriteOnce ]storageClassName: alicloud-disk-efficiencyresources:requests:storage: 20Gi
