车联网是物联网的一个主要应用方向,车辆通过连接车联网平台,实时进行消息的交互,平台可以提供车辆远程控制,故障检测,车路协同等各方面的功能。
我在车联网行业从事了很长时间的技术工作,参与了整个车联网平台的构建以及很多不同车联网应用的开发工作,这里打算以构建一个车联网平台作为例子,总结一下涉及到的架构设计方面的东西。
系统功能
在物联网中,主要应用的通信协议是MQTT,这是一个基于发布/订阅模式的物联网通信协议,具备了支持QoS,简单易实现,报文紧凑等特点。在车联网中,大部分的车企也是采用MQTT协议来进行通讯。因此在车联网的架构中,我们需要考虑设置一个MQTT Broker集群来与大量的车辆进行连接通信。目前有很多的开源的Broker,例如ActiveMQ, EMQ, RocketMQ等等,其中EMQ和RocketMQ都是国内的产品,有详尽的中文资料介绍。这里我选择EMQ作为MQTT Broker。
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集群
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,因为我们要创建一个statefulset的emqx多个节点,要实现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"
定义一个headless的service,用于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
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集群。这种方式可以提供如下好处:
- HAProxy作为一个反向代理可以隐藏emqx节点的信息,并为外部提供一个统一的地址来连接
- 可以用作MQTT over TLS的终结,减轻emqx节点处理SSL加密的计算负荷,并且简化证书部署和管理的工作
- 提供内在的MQTT支持,支持解析MQTT消息以实现粘性附着和智能负荷分配等功能
- 通过主备方式提供高可靠性
同样我也是以kustomization的方式来部署HAProxy
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
apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
resources:
- namespace.yaml
- configmap.yaml
- haproxy_deployment.yaml
- testservice.yaml
配置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
helm upgrade -f values.yaml haproxy-kubernetes-ingress -n haproxy-controller haproxytech/kubernetes-ingress
现在我们就可通过一个MQTT客户端来通过HAPROXY来连接EMQX了,服务器地址是minikubeip:1883
配置证书
未完待续。。。
原文地址:https://blog.csdn.net/gzroy/article/details/134562705
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。
如若转载,请注明出处:http://www.7code.cn/show_28146.html
如若内容造成侵权/违法违规/事实不符,请联系代码007邮箱:suwngjj01@126.com进行投诉反馈,一经查实,立即删除!