长沙手机网站建设公司排名,网站的建设时间怎么查,中国最好的少儿编程培训机构,俄文网站开发车联网是物联网的一个主要应用方向#xff0c;车辆通过连接车联网平台#xff0c;实时进行消息的交互#xff0c;平台可以提供车辆远程控制#xff0c;故障检测#xff0c;车路协同等各方面的功能。
我在车联网行业从事了很长时间的技术工作#xff0c;参与了整个车联网…车联网是物联网的一个主要应用方向车辆通过连接车联网平台实时进行消息的交互平台可以提供车辆远程控制故障检测车路协同等各方面的功能。
我在车联网行业从事了很长时间的技术工作参与了整个车联网平台的构建以及很多不同车联网应用的开发工作这里打算以构建一个车联网平台作为例子总结一下涉及到的架构设计方面的东西。
系统功能
一个车联网平台需要实现以下的一些功能
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: emqxname: emqx
---
kind: Role
apiVersion: rbac.authorization.k8s.io/v1
metadata:namespace: emqxname: emqx
rules:
- apiGroups:- resources:- endpoints verbs: - get- watch- list
---
kind: RoleBinding
apiVersion: rbac.authorization.k8s.io/v1
metadata:namespace: emqxname: emqx
subjects:- kind: ServiceAccountname: emqxnamespace: emqx
roleRef:kind: Rolename: emqxapiGroup: rbac.authorization.k8s.io
定义一个configmap因为我们要创建一个statefulset的emqx多个节点要实现auto cluster的功能自动把这多个节点组成一个cluster因此需要定义相关的配置
apiVersion: v1
kind: ConfigMap
metadata:name: emqx-confignamespace: emqx
data:EMQX_NAME: emqxEMQX_CLUSTER__DISCOVERY_STRATEGY: k8sEMQX_CLUSTER__K8S__SERVICE_NAME: emqx-headlessEMQX_CLUSTER__K8S__NAMESPACE: emqxEMQX_CLUSTER__K8S__ADDRESS_TYPE: hostnameEMQX_CLUSTER__K8S__APISERVER: https://kubernetes.default.svc:443EMQX_CLUSTER__K8S__SUFFIX: svc.cluster.local
定义一个headless的service用于statefulset的服务暴露和通信。
apiVersion: v1
kind: Service
metadata:name: emqx-headlessnamespace: emqx
spec:type: ClusterIPclusterIP: Noneselector:app: emqxports:- name: mqttport: 1883protocol: TCPtargetPort: 1883- name: mqttsslport: 8883protocol: TCPtargetPort: 8883- name: mgmtport: 8081protocol: TCPtargetPort: 8081- name: websocketport: 8083protocol: TCPtargetPort: 8083- name: wssport: 8084protocol: TCPtargetPort: 8084- name: dashboardport: 18083protocol: TCPtargetPort: 18083
最后是定义一个statefulset里面包含了2个节点。
apiVersion: apps/v1
kind: StatefulSet
metadata:name: emqx-statefulsetlabels:app: emqxnamespace: emqx
spec:serviceName: emqx-headlessupdateStrategy:type: RollingUpdatereplicas: 2selector:matchLabels:app: emqxtemplate:metadata:labels:app: emqxspec:serviceAccountName: emqxcontainers:- name: emqximage: emqx/emqx:5.1.6resources:requests:memory: 1Gicpu: 250mlimits:memory: 1Gicpu: 250mports:- name: mqttcontainerPort: 1883- name: mqttsslcontainerPort: 8883- name: mgmtcontainerPort: 8081- name: wscontainerPort: 8083- name: wsscontainerPort: 8084- name: dashboardcontainerPort: 18083envFrom:- 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 [emqxemqx-statefulset-0.emqx-headless.emqx.svc.cluster.local,emqxemqx-statefulset-1.emqx-headless.emqx.svc.cluster.local],stopped_nodes []}
可见当前的EMQX cluster包括了两个节点并已成功运行。
配置HAProxy
下一步我将配置一个HAProxy来作为Load balancer连接EMQX集群。这种方式可以提供如下好处
HAProxy作为一个反向代理可以隐藏emqx节点的信息并为外部提供一个统一的地址来连接可以用作MQTT over TLS的终结减轻emqx节点处理SSL加密的计算负荷并且简化证书部署和管理的工作提供内在的MQTT支持支持解析MQTT消息以实现粘性附着和智能负荷分配等功能通过主备方式提供高可靠性
同样我也是以kustomization的方式来部署HAProxy
定义一个namespace
apiVersion: v1
kind: Namespace
metadata:name: haproxy
定义一个configmap因为haproxy启动需要读取haproxy.cfg配置文件的信息把这个文件通过configmap的方式来加载
apiVersion: v1
kind: ConfigMap
metadata:name: haproxy-confignamespace: haproxy
data:haproxy.cfg: |global log 127.0.0.1 local3 info daemon maxconn 10240defaults log global mode tcp option tcplog #option dontlognull timeout connect 10000 # timeout mqtts keepalive * 1.2 timeout client 240s timeout server 240s maxconn 20000backend mqtt_backendmode tcp# 粘性会话负载均衡stick-table type string len 32 size 1000k expire 30mstick on req.payload(0,0),mqtt_field_value(connect,client_identifier)server emqx0 emqx-statefulset-0.emqx-headless.emqx.svc.cluster.local:1883server emqx1 emqx-statefulset-1.emqx-headless.emqx.svc.cluster.local:1883frontend mqtt_serversbind *:1883mode 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: haproxyname: haproxynamespace: haproxy
spec:replicas: 1selector:matchLabels:app: haproxytemplate:metadata:labels:app: haproxyspec:containers:- name: haproxyimage: haproxy:2.8ports:- name: httpcontainerPort: 80- name: httpscontainerPort: 443- name: haproxy-mgmtcontainerPort: 1024- name: mqttcontainerPort: 1883- name: mqttsslcontainerPort: 8883- name: mgmtcontainerPort: 8081- name: wscontainerPort: 8083- name: wsscontainerPort: 8084- name: dashboardcontainerPort: 18083volumeMounts:- name: haproxy-configmountPath: /usr/local/etc/haproxy/haproxy.cfgsubPath: haproxy.cfgvolumes:- name: haproxy-configconfigMap:name: haproxy-configitems:- key: haproxy.cfgpath: haproxy.cfg
定义一个service暴露harpoxy的端口
apiVersion: v1
kind: Service
metadata:name: haproxy-servicenamespace: haproxy
spec:selector:app: haproxyports:- name: mqttport: 1883protocol: TCPtargetPort: mqtt- name: mqttsport: 8883protocol: TCPtargetPort: 8883- name: wsport: 8083protocol: TCPtargetPort: 8083- name: wssport: 8084protocol: TCPtargetPort: 8084- name: dashboardport: 18083protocol: TCPtargetPort: 18083
最后定义一个kustomization.yaml
apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
resources:
- namespace.yaml
- configmap.yaml
- haproxy_deployment.yaml
- service.yaml
运行kubectl apply -k来部署
配置Ingress
在我的minikube k8s集群上暴露HAProxy的端口使得外部可以访问MQTT。因为我想仍然暴露1883端口给外部访问所以需要在minikube启动的时候设置
minikube start --extra-configapiserver.service-node-port-range1-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端口通过kubectl apply -f来部署。
apiVersion: v1
kind: ConfigMap
metadata:name: tcpnamespace: haproxy
data:1883:haproxy/haproxy-service:18838883:haproxy/haproxy-service:8883 8083:haproxy/haproxy-service:80838084:haproxy/haproxy-service:808418083:haproxy/haproxy-service:18083
读取HAProxy ingress的配置信息保存在values.yaml文件
helm show values haproxytech/kubernetes-ingress values.yaml
然后在values.yaml里面找到以下对应位置进行修改 tcpPorts:- name: mqttport: 1883targetPort: 1883nodePort: 1883- name: mqttsport: 8883targetPort: 8883nodePort: 8883- name: wsport: 8083targetPort: 8083nodePort: 8083- name: wssport: 8084targetPort: 8084nodePort: 8084- name: dashboardport: 18083targetPort: 18083nodePort: 18083
# add extra args in controller sectionextraArgs:- --configmap-tcp-serviceshaproxy/tcp
运行以下命令更新haproxy-ingress的配置
helm upgrade -f values.yaml haproxy-kubernetes-ingress -n haproxy-controller haproxytech/kubernetes-ingress
现在我们就可通过一个MQTT客户端来通过HAPROXY来连接EMQX了服务器地址是minikubeip:1883通过访问minikubeip:18083可以访问EMQX的dashboard
配置证书
在实际应用中车辆和平台之间是通过TLS加密来通信的有单向认证和双向认证两种方式。单向认证指客户端需要验证服务器端是否持有受信任的证书双向认证则指双方都需要验证。这里以双向认证为例进行配置。
1. 创建根CA证书
以自签证书的方式来做首先是创建一个根CA证书如以下命令。
openssl req -newkey rsa:2048 -nodes -x509 -days 3650 -keyout root-ca.key -out root-ca.crt
用以下命令可以查看创建的证书内容
openssl x509 -noout -text -in root-ca.crt
2. 创建中间CA证书
有了根CA之后我们可以用来签发中间CA证书因为一般证书都不会直接用根CA来签发的。
输入以下命令来创建中间CA的Key和证书签发请求csr注意common name不能和之前根CA的重复
openssl req -newkey rsa:2048 -nodes -days 3650 -keyout intermediate-ca.key -out intermediate-ca.csr
因为我们这个证书是要用作CA需要定义一个扩展文件来描述是一个CA type创建一个ca-cert-extension.cnf内容如下
basicConstraints CA:TRUE
keyUsage keyCertSign, cRLSign
subjectKeyIdentifier hash
authorityKeyIdentifier keyid,issuer
然后就可以用根CA证书对这个csr进行签发生成中间CA证书
openssl x509 -req -in intermediate-ca.csr -out intermediate-ca.crt -CA root-ca.crt -CAkey root-ca.key -CAcreateserial -days 3650 -extfile ca-cert-extensions.cnf
3. 为客户端签发证书
有了中间CA证书后我们就可以为客户端创建证书了例如为一辆ID为vehicle-1的车辆签发证书。
openssl req -newkey rsa:2048 -nodes -days 365 -subj /CNvehicle-1/Ovehicle -keyout client.key -out client.csr
创建一个扩展文件client-cert-extensions.cnf声明其是client certificate type
basicConstraints CA:FALSE
keyUsage digitalSignature
extendedKeyUsage clientAuth
subjectKeyIdentifier hash
authorityKeyIdentifier keyid,issuer
然后用中间CA证书签发
openssl x509 -req -in client.csr -out client.crt -CA intermediate-ca.crt -CAkey intermediate-ca.key -CAcreateserial -days 365 -extfile client-cert-extensions.cnf
4. 为服务器端签发证书
同理为服务器端也签发证书
openssl req -newkey rsa:2048 -nodes -days 365 -subj /CNemqx/Oserver -keyout server.key -out server.csr
创建一个扩展文件server-cert-extensions.cnf
basicConstraints CA:FALSE
keyUsage digitalSignature
extendedKeyUsage serverAuth
subjectKeyIdentifier hash
authorityKeyIdentifier keyid,issuer
用中间CA证书签发
openssl x509 -req -in server.csr -out server.crt -CA intermediate-ca.crt -CAkey intermediate-ca.key -CAcreateserial -days 365 -extfile server-cert-extensions.cnf
然后把server.crt和server.key合成一个文件
