联网是物联网一个主要应用方向车辆通过连接联网平台实时进行消息交互平台可以提供车辆远程控制故障检测,车路协同等各方面的功能

我在车联网行业从事了很长时间技术工作,参与了整个车联网平台构建以及很多不同车联网应用开发工作这里打算以构建一个车联网平台作为例子总结一下涉及到的架构设计方面的东西。

系统功能

一个车联网平台需要实现以下的一些功能

1. 与车辆消息交互

在物联网中,主要应用通信协议是MQTT,这是一个基于发布/订阅模式的物联网通信协议,具备了支持QoS,简单易实现,报文紧凑等特点。在车联网中,大部分的车企也是采用MQTT协议来进行通讯。因此在车联网的架构中,我们需要考虑设置一个MQTT Broker集群来与大量的车辆进行连接通信。目前有很多的开源的Broker例如ActiveMQ, EMQ, RocketMQ等等,其中EMQ和RocketMQ都是国内产品,有详尽的中文资料介绍这里选择EMQ作为MQTT Broker

2. 车辆消息的消费与存储

MQTT Broker接收到车辆消息后,需要消息给到上层应用来进行处理。我们可以把这些消息保存数据库或者转发一个消息队列缓存这里选择Kafka。上层应用通过订阅Kafka主题,来获得其需要相关车辆信息,进行处理。上层应用也可以把要下发给车辆的消息发送到Kafka主题然后让MQTT Broker再转发给车辆,也可以直接通过MQTT主题发布消息的方式来直接发送给车辆。

3. V2X应用

包括了V2V, V2I, V2P等应用场景,车辆需要能和不同的数据源进行消息交互,从而为驾驶提供决策信息。我将基于这个平台展示一些V2X应用的开发设计,实现3GPP规范里面制定的一些V2X场景

4. 车辆数据分析与报表

车联网平台每天收集生成了大量的数据通过对这些数据进行发掘分析可以更好的了解业务运行的情况,同时也可以更好的为商业决策提供参考。我们可以基于目前流行的大数据处理平台例如Spark/Beam/Flink等,对数据进行即时的处理,保存到数据仓库,随后再进行各种数据分析和报表呈现

在这篇文章中,我先对以上提到的第一点功能进行介绍,搭建一个MQTT消息平台。

MQTT消息平台

选择EMQX来搭建这个平台,EMQX是国内的一个优秀的MQTT broker软件,有企业版和开源版,这里选择开源版。在官网上有介绍安装方式,在Kubernetes上是采用Operator的方式安装的,但是我这里采用kustomization方式安装,因为这样方便我进行一些设置上的改动。在我本地minikube启动了一个kubernetes cluster

安装EMQX集群

定义一个新的namespace

apiVersion: v1
kind: Namespace
metadata:
  name: emqx

为这个namespace创建一个service account并赋予相关权限

apiVersion: v1
kind: ServiceAccount
metadata:
  namespace: emqx
  name: emqx
---
kind: Role
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  namespace: emqx
  name: emqx
rules:
- apiGroups:
  - ""
  resources:
  - endpoints 
  verbs: 
  - get
  - watch
  - list
---
kind: RoleBinding
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  namespace: emqx
  name: emqx
subjects:
  - kind: ServiceAccount
    name: emqx
    namespace: emqx
roleRef:
  kind: Role
  name: emqx
  apiGroup: rbac.authorization.k8s.io

定义一个configmap,因为我们要创建一个statefulsetemqx多个节点,要实现auto cluster功能自动把这多个节点组成一个cluster,因此需要定义相关配置

apiVersion: v1
kind: ConfigMap
metadata:
  name: emqx-config
  namespace: emqx
data:
  EMQX_NAME: "emqx"
  EMQX_CLUSTER__DISCOVERY_STRATEGY: "k8s"
  EMQX_CLUSTER__K8S__SERVICE_NAME: "emqx-headless"
  EMQX_CLUSTER__K8S__NAMESPACE: "emqx"
  EMQX_CLUSTER__K8S__ADDRESS_TYPE: "hostname"
  EMQX_CLUSTER__K8S__APISERVER: "https://kubernetes.default.svc:443"
  EMQX_CLUSTER__K8S__SUFFIX: "svc.cluster.local"

定义一个headlessservice用于statefulset服务暴露通信

apiVersion: v1
kind: Service
metadata:
  name: emqx-headless
  namespace: emqx
spec:
  type: ClusterIP
  clusterIP: None
  selector:
    app: emqx
  ports:
  - name: mqtt
    port: 1883
    protocol: TCP
    targetPort: 1883
  - name: mqttssl
    port: 8883
    protocol: TCP
    targetPort: 8883
  - name: mgmt
    port: 8081
    protocol: TCP
    targetPort: 8081
  - name: websocket
    port: 8083
    protocol: TCP
    targetPort: 8083
  - name: wss
    port: 8084
    protocol: TCP
    targetPort: 8084
  - name: dashboard
    port: 18083
    protocol: TCP
    targetPort: 18083

最后定义一个statefulset里面包含了2个节点

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: emqx-statefulset
  labels:
    app: emqx
  namespace: emqx
spec:
  serviceName: emqx-headless
  updateStrategy:
    type: RollingUpdate
  replicas: 2
  selector:
    matchLabels:
      app: emqx
  template:
    metadata:
      labels:
        app: emqx
    spec:
      serviceAccountName: emqx
      containers:
      - name: emqx
        image: emqx/emqx:5.1.6
        resources:
          requests:
            memory: "1Gi"
            cpu: "250m"
          limits:
            memory: "1Gi"
            cpu: "250m"
        ports:
        - name: mqtt
          containerPort: 1883
        - name: mqttssl
          containerPort: 8883
        - name: mgmt
          containerPort: 8081
        - name: ws
          containerPort: 8083
        - name: wss
          containerPort: 8084
        - name: dashboard
          containerPort: 18083
        envFrom:
          - configMapRef:
              name: emqx-config

定义一个kustomization.yaml文件,把以上定义的manifest包括进来:

apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
resources:
- namespace.yaml
- rbac.yaml
- configmap.yaml
- headless.yaml
- statefulset.yaml

最后运行kubectl apply -k即可部署,我们可以运行以下命令查看emqx cluster的状态

kubectl exec emqx-statefulset-0 -n emqx -- emqx_ctl cluster status

如果成功运行,将显示如下信息

Cluster status: #{running_nodes =>
                      ['emqx@emqx-statefulset-0.emqx-headless.emqx.svc.cluster.local',
                       'emqx@emqx-statefulset-1.emqx-headless.emqx.svc.cluster.local'],
                  stopped_nodes => []}

可见当前的EMQX cluster包括了两个节点并已成功运行。

配置HAProxy

一步我将配置一个HAProxy来作为Load balancer,连接EMQX集群。这种方式可以提供如下好处:

同样我也是以kustomization的方式部署HAProxy

定义一个namespace

apiVersion: v1
kind: Namespace
metadata:
  name: haproxy

定义一个configmap,因为haproxy启动需要读取haproxy.cfg配置文件信息,把这个文件通过configmap方式加载

apiVersion: v1
kind: ConfigMap
metadata:
  name: haproxy-config
  namespace: haproxy
data:
  haproxy.cfg: |
    global  
      log 127.0.0.1 local3 info 
      daemon  
      maxconn 10240

    defaults  
      log global 
      mode tcp 
      option tcplog 
      #option dontlognull  
      timeout connect 10000 
      # timeout > mqtt's keepalive * 1.2  
      timeout client 240s  
      timeout server 240s 
      maxconn 20000

    backend mqtt_backend
      mode tcp
      # 粘性会话负载均衡
      stick-table type string len 32 size 1000k expire 30m
      stick on req.payload(0,0),mqtt_field_value(connect,client_identifier)

      server emqx0 emqx-statefulset-0.emqx-headless.emqx.svc.cluster.local:1883
      server emqx1 emqx-statefulset-1.emqx-headless.emqx.svc.cluster.local:1883

    frontend mqtt_servers
      bind *:1883
      mode tcp
      # 拒绝非 MQTT 连接
      # tcp-request content reject unless { req.payload(0,0),mqtt_is_valid }
      default_backend mqtt_backend

定义一个deployment

apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: haproxy
  name: haproxy
  namespace: haproxy
spec:
  replicas: 1
  selector:
    matchLabels:
      app: haproxy
  template:
    metadata:
      labels:
        app: haproxy
    spec:
      containers:
      - name: haproxy
        image: haproxy:2.8
        ports:
        - name: http
          containerPort: 80
        - name: https
          containerPort: 443
        - name: haproxy-mgmt
          containerPort: 1024
        - name: mqtt
          containerPort: 1883
        - name: mqttssl
          containerPort: 8883
        - name: mgmt
          containerPort: 8081
        - name: ws
          containerPort: 8083
        - name: wss
          containerPort: 8084
        - name: dashboard
          containerPort: 18083
        volumeMounts:
        - name: haproxy-config
          mountPath: /usr/local/etc/haproxy/haproxy.cfg
          subPath: haproxy.cfg
      volumes:
      - name: haproxy-config
        configMap:
          name: haproxy-config
          items:
            - key: haproxy.cfg
              path: haproxy.cfg

最后定义一个kustomization.yaml

apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
resources:
- namespace.yaml
- configmap.yaml
- haproxy_deployment.yaml
- testservice.yaml

运行kubectl apply -k来部署

配置Ingress

在我的minikube k8s集群上暴露HAProxy的端口,使得外部可以访问MQTT。因为我想仍然暴露1883端口外部访问,所以需要在minikube启动的时候设置

minikube start --extra-config=apiserver.service-node-port-range=1-65535

然后安装HAProxy ingress通过helm方式安装

helm repo add haproxytech https://haproxytech.github.io/helm-charts
helm repo update
helm install haproxy-kubernetes-ingress haproxytech/kubernetes-ingress 
  --create-namespace 
  --namespace haproxy-controller

安装完成之后,我们需要创建一个configmap配置要暴露的TCP端口

apiVersion: v1
kind: ConfigMap
metadata:
  name: tcp
  namespace: haproxy
data:
  1883:
    haproxy/haproxy-service:1883   

读取HAProxy ingress的配置信息,保存在values.yaml文件

helm show values haproxytech/kubernetes-ingress > values.yaml

然后values.yaml里面找到以下对应位置,进行修改

    tcpPorts:
     - name: mqtt
       port: 1883
       targetPort: 1883
       nodePort: 1883
# add extra args in controller section
  extraArgs:
   - --configmap-tcp-services=haproxy/tcp

运行以下命令更新haproxy-ingress的配置

helm upgrade -f values.yaml haproxy-kubernetes-ingress -n haproxy-controller haproxytech/kubernetes-ingress

现在我们就可通过一个MQTT客户端通过HAPROXY来连接EMQX了,服务器地址minikubeip:1883

配置证书

一步我们可以配置TLS加密

未完待续。。。

原文地址:https://blog.csdn.net/gzroy/article/details/134562705

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任

如若转载,请注明出处:http://www.7code.cn/show_28146.html

如若内容造成侵权/违法违规/事实不符,请联系代码007邮箱suwngjj01@126.com进行投诉反馈,一经查实,立即删除

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注