Kubernetes调度

一、调度系统集群模式

Kubernetes调度系统本身的集群模式是主从。由Master节点负责调度，将任务调度分配到Node节点中。

二、在集群模式下，调度系统本身如何调度资源，例如如果要调度任务 A，那么应该是在调度系统进群的哪个节点上调度

Kubernetes的调度模式如下：

首先调度器会寻找nodeName为空的Pod进行作业调度，调度流程分为两部。
第一步Predicate：过滤不符合条件的节点

第二部Priority：优先级打分排序，选择优先级最高的节点。

三、任务分类

Kubernetes支持 LRS（Long Running Service）和 Batch Jobs 两种作业形态，对它们进行“分别管理”和“混合调度”。这里针对于任务分类，我们主要说一下Job和CronJob这两个对象。

1. 首先我先介绍一下Job对象中负责并行控制的两个参数：
   • spec.completions：控制任务完成的数量。它定义的是 Job 至少要完成的 Pod 数目，即 Job 的最小完成数。
   • spec.parallelism：控制任务并行数。它定义的是一个 Job 在任意时间最多可以启动多少个 Pod 同时运行
2. CronJob 他是一个专门用来管理 Job 对象的控制器。只不过，它创建和删除 Job 的依据，是 schedule 字段定义的、一个标准的Unix Cron格式的表达式。

所以对于我们对于任务的分类可以使用如下方式来实现：

• 重复任务：可以将completions设置为重复完成的数量，parallelism根据需求设定
• 定时任务：可以使用cronJob来实现定时任务。
• 一次性任务：可以将completions设置为1。
• 实时任务：可以使用job来实现。

四、任务管理

Kubernetes是通过apiserver来查询任务的运行情况。当任务的STATUS字段会记录用户任务执行的情况。

任务日志可以输出到stdout和stderr中然后通过在宿主机上部署 logging-agent 的方式来集中处理日志。

Kubernetes整个项目是声明式API，他只规定他的期望的目的。然后剩下的交给系统来处理。直到达到期望值。一个 Kubernetes 的控制器，实际上就是一个“死循环”，它不断地获取“实际状态”，然后与“期望状态”作对比，并以此为依据决定下一步的操作。

因此他允许删除任务。删除时，将任务，没有调度的任务直接从队列里丢弃。允许修改任务配置。修改调度配置以后会按照新的调度配置重新执行一遍。当删除任务时，任务先接收term信号， 再接收kill信号。

五、任务编排

Kubernetes的核心功能就是任务编排。kube-controller-manager这个组件，就是一系列控制器的集合。每一个控制器，都以独有的方式负责某种编排功能。这里的控制器，都遵循 Kubernetes 项目中的一个通用编排模式，即：控制循环（control loop）

比如，现在有一种待编排的对象 X，它有一个对应的控制器。那么，我就可以用一段 Go 语言风格的伪代码，为你描述这个控制循环：

1
2
3
4
5
6
7
8
9

for {
  实际状态 := 获取集群中对象X的实际状态（Actual State）
  期望状态 := 获取集群中对象X的期望状态（Desired State）
  if 实际状态 == 期望状态{
    什么都不做
  } else {
    执行编排动作，将实际状态调整为期望状态
  }
}

举例：
以Deployment 对象中 Replicas 字段的值为例。很明显，这些信息往往都保存在 Etcd 中。

接下来，以 Deployment 为例，我和你简单描述一下它对控制器模型的实现：

1. Deployment 控制器从 Etcd 中获取到所有携带了“app: nginx”标签的 Pod，然后统计它们的数量，这就是实际状态；
2. Deployment 对象的 Replicas 字段的值就是期望状态；
3. Deployment 控制器将两个状态做比较，然后根据比较结果，确定是创建 Pod，还是删除已有的 Pod；

可以看到，一个 Kubernetes 对象的主要编排逻辑，实际上是在第三步的“对比”阶段完成的。这个操作，通常被叫作调谐（Reconcile）。这个调谐的过程，则被称作“Reconcile Loop”（调谐循环）或者“Sync Loop”（同步循环）。

Deployment控制器，控制ReplicaSet。ReplicaSet支持Pod的“水平扩展 / 收缩”和“滚动更新”

Deployment 同样通过“控制器模式”，来操作 ReplicaSet 的个数和属性，进而实现“水平扩展 / 收缩”和“滚动更新”这两个编排动作。

比如，把replicas这个值从 3 改成 4，那么 Deployment 所对应的 ReplicaSet，就会根据修改后的值自动创建一个新的 Pod。这就是“水平扩展”了；“水平收缩”则反之。

六、任务注册与任务发现

Kubernetes有后台管理界面可以注册任务。
以Deployment为例，注册时填写的字段有：应用名称、容器镜像、pod的数量、定义的Service。还有一些高级选项如下图所示。

Kubernetes还可以以编程的形式创建任务。可以使用client-go来进行接入。

七、任务形态

Kubernetes可以通过grpc和http两种形式创建任务。

1. 使用管理控制台来创建任务时调用的是https处理任务。
2. 使用命令行在创建任务时是调用的是grpc处理任务。

八、调度模型

kube-scheduler监听kube-apiserver，查询还未分配Node的Pod（即NodeName为空的Pod）。

kubernetes的线程模型是goroutine模型，goroutine调度器的流程本质上是一个生产-消费的流程。

它的调度模型与算法也是不断优化演变的，从最初的 G-M 模型、到 G-M-P 模型，从不支持抢占，到支持协作式抢占，再到支持基于信号的异步抢占。下面简单的介绍一下GMP模型:

G：我们可以看做是goroutine

P：逻辑处理器，他上面最重要的三个结构是runNext、local run queue、global run queue。

M：系统线程（真正处理任务的CPU）

首先G我们可以看做生产者，当一个goroutine被创建时，会包装成一个G，

1. 此时生成的G会先看一下P的runnext上有没有G。
2. 如果没有的话，刚创建的G则直接放在runnext上，
3. 如果runNext上有G，看一下P上的local run queue有没有满
   1. 如果没有满，则将runnext上的G放到local run queue（本地队列中）
   2. 如果满了，则将runnext上的G+local run queue中一半的G拿出来组成一个链表放到global run queue（全局队列）中。

M我们可以看做消费者，

1. 首先runtime.schedle检测schedtick%61==0？
2. 如果schedtick%61==0，则直接从全局队列里获取队里第一个G执行。
3. schedtick%61≠0，先看P中runnext是否有值
4. 如果runnext中有值，M直接拿runnext中的G执行。
5. 如果runnext中没有值，先去看local run queue中是否有G，
6. local run queue有G则获取本地队列里的第一个G放到M上执行
7. local run queue没有G，则去看global run queue中是否有G
8. global run queue有G，则从global run queue中取（全局队列g总数 / gomaxprocs）+ 1个，但是不能超过128个G。将从全局队列中拿出来的G的第一个放到M上执行，剩下的放入到本地队列中。
9. global run queue没有G，则去看其他P中是否有G，如果有的话，从其他P的local run queue中的队头开始，偷一半。取偷过来的最后一个放到M上执行，剩下的放入当前P的本地队列中。

下面我们在看一下goroutine 调度器是如何进行抢占式调度的。Go 程序启动时，运行时会去启动一个名为 sysmon 的 M（一般称为监控线程），这个 M 的特殊之处在于它不需要绑定 P 就可以运行（以 g0 这个 G 的形式）

sysmon 每 20us~10ms 启动一次，sysmon 主要完成了这些工作：

1. 释放闲置超过 5 分钟的 span 内存；
2. 如果超过 2 分钟没有垃圾回收，强制执行；
3. 将长时间未处理的 netpoll 结果添加到任务队列；
4. 向长时间运行的 G 任务发出抢占调度；
5. 收回因 syscall 长时间阻塞的 P；

如果一个 G 任务运行 10ms，sysmon 就会认为它的运行时间太久而发出抢占式调度的请求。一旦 G 的抢占标志位被设为 true，那么等到这个 G 下一次调用函数或方法时，运行时就可以将 G 抢占并移出运行状态，放入队列中，等待下一次被调度。

九、调度策略

在第二大部分时，有介绍到对于Kubernetes的调度分为两个部分，第一部分Predicates，先对所有节点进行按照一定的策略进行过滤，第二部分Priority，对第一部分过滤出来的结果按照一定的策略进行打分，选择打分最高的节点进行调度。下面我来分别介绍一下这两部分的调度策略。

1. Predicates中的调度策略有很多种，还可以自己编写自己的调度策略，这里我就简单介绍下面几种调度策略：
   • PodFitsHostPorts：检查是否有Host Ports冲突
   • PodFitsPorts：和PodFitsHostPorts相同
   • PodFitsResources：检查Node的资源是否充足，包括允许的Pod数量、CPU、内存、GPU个数以及其他的OpaqueIntResources。
   • HostName：检查pod.Spec.NodeName是否与候选节点一致。
   • MatchNodeSelector：检查候选节点的pod.Spec.NodeSelector是否匹配
   • NoVolumeZoneConflict：检查volume zone是否冲突
   • MatchinterPodAffinity：检查是否匹配Pod的亲和性要求
   • NoDiskConflict：检查是否存在Volume冲突，仅限于GCE PD、AWS EBS、Ceph RBD以及ISCSI。
   • PodToleratesNodeTaints：检查Pod是否容忍Node Taints。
   • CheckNodeMemoryPressure：检查Pod是否可以调度到MemoryPressure的节点上。
   • CheckNodeDiskPressure：检查Pod是否可以调度到DiskPressure的节点上。
   • NoVolumeNodeConflict：检查节点是否满足Pod所引用的Volume的条件。
2. Priority中的调度策略有很多种，还可以自己编写自己的调度策略，这里我就简单介绍下面几种调度策略：
   • SelectorSpreadPriority：优先减少节点上属于同一个Service或Replication Controller的Pod数量
   • InterPodAffinityPriority：优先将Pod调度到相同的拓扑上（如同一个节点、Rack、Zone等）
   • LeastRequestedPriority：优先调度到请求资源少的节点上。
   • BalancedResourceAllocation：优先平衡各节点的资源使用。
   • NodePreferAvoidPodsPriority：alpha.kubernetes.io/preferAvoidPods 字段判断，权重为10000，避免其他优先级策略的影响。
   • NodeAffinityPriority：优先调度到匹配NodeAffinity的节点上。
   • TaintTolerationPriority：优先调度到匹配TaintToleration的节点上。
   • ServiceSpreadingPriority：尽量将同一个service的Pod分布到不同节点上，已经被SelectorSpreadPriority替代（默认未使用）
   • EqualPriority：将所有节点的优先级设置为1（默认未使用）
   • ImageLocalityPriority：尽量将使用大镜像的容器调度到已经下拉了该镜像的节点上（默认未使用）
   • MostRequestedPriority：尽量调度到已经使用过的Node上，特别适用于cluster-autoscaler（默认未使用）

Kubernetes是允许用户通过调度框架（Scheduling Framework）来自定义调度器。Scheduling Framework为我们定义了一些扩展点，用户能够实现扩展点定义的接口来定义本身的调度逻辑，并将扩展注册到扩展点上，调度器在执行调度流程时，会检查扩展plugin，如果有的话扩展plugin的话，则处理扩展plugin。

Scheduler 扩展点：

调度行为发生在一系列阶段中，这些阶段是通过以下扩展点公开的：

• QueueSort：这些插件对调度队列中的悬决的 Pod 排序。 一次只能启用一个队列排序插件。
• PreFilter：这些插件用于在过滤之前预处理或检查 Pod 或集群的信息。 它们可以将 Pod 标记为不可调度。
• Filter：这些插件相当于调度策略中的断言（Predicates），用于过滤不能运行 Pod 的节点。 过滤器的调用顺序是可配置的。 如果没有一个节点通过所有过滤器的筛选，Pod 将会被标记为不可调度。
• PreScore：这是一个信息扩展点，可用于预打分工作。
• Score：这些插件给通过筛选阶段的节点打分。调度器会选择得分最高的节点。
• Reserve：这是一个信息扩展点，当资源已经预留给 Pod 时，会通知插件。 这些插件还实现了 Unreserve 接口，在 Reserve 期间或之后出现故障时调用。
• Permit：这些插件可以阻止或延迟 Pod 绑定。
• PreBind：这些插件在 Pod 绑定节点之前执行。
• Bind：这个插件将 Pod 与节点绑定。绑定插件是按顺序调用的，只要有一个插件完成了绑定，其余插件都会跳过。绑定插件至少需要一个。
• PostBind：这是一个信息扩展点，在 Pod 绑定了节点之后调用。

十、监控与告警

Kubernetes 项目的监控体系是以 Prometheus 项目为核心的一套统一的方案。下面是Prometheus官网的架构图。

可以看到，Prometheus 项目工作的核心，是使用 Pull （抓取）的方式去搜集被监控对象的 Metrics 数据（监控指标数据），然后，再把这些数据保存在一个 TSDB （时间序列数据库，比如 OpenTSDB、InfluxDB 等）当中，以便后续可以按照时间进行检索。

有了这套核心监控机制， Prometheus 剩下的组件就是用来配合这套机制的运行。比如 Pushgateway，可以允许被监控对象以 Push 的方式向 Prometheus 推送 Metrics 数据。而 Alertmanager，则可以根据 Metrics 信息灵活地设置报警。当然， Prometheus 最受用户欢迎的功能，还是通过 Grafana 对外暴露出的、可以灵活配置的监控数据可视化界面。

Kubernetes中Prometheus 拉取的Metrics 数据的来源大致分为三种：

1. 宿主机的监控数据。这部分数据的提供，需要借助一个由 Prometheus 维护的Node Exporter 工具。一般来说，Node Exporter 会以 DaemonSet 的方式运行在宿主机上。其实，所谓的 Exporter，就是代替被监控对象来对 Prometheus 暴露出可以被“抓取”的 Metrics 信息的一个辅助进程。
2. Kubernetes 的 API Server、kubelet 等组件的 /metrics API。除了常规的 CPU、内存的信息外，这部分信息还主要包括了各个组件的核心监控指标。比如，对于 API Server 来说，它就会在 /metrics API 里，暴露出各个 Controller 的工作队列（Work Queue）的长度、请求的 QPS 和延迟数据等等。这些信息，是检查 Kubernetes 本身工作情况的主要依据。
3. Kubernetes 相关的监控数据。这部分数据，一般叫作 Kubernetes 核心监控数据（core metrics）。这其中包括了 Pod、Node、容器、Service 等主要 Kubernetes 核心概念的 Metrics。

十一、认证和权限

Kubernetes使用Namespace来实现资源的隔离。

用户使用 kubectl、客户端库或构造 REST 请求来访问 Kubernetes API。 Kubernetes 服务账户（ServiceAccount）可以被鉴权访问 API。 当请求到达 API 时，它会经历多个阶段，如下图所示：

具体的权限控制流程如下：

• 用户携带令牌或者证书给Kubernetes的api-server发送请求要求查询修改集群资源
• Kubernetes开始认证。认证通过
• Kubernetes查询用户的授权（在 Kubernetes 项目中，负责完成授权（Authorization）工作的机制，就是 RBAC）
• 用户执行操作。过程中的一些操作（cpu、内存、硬盘、网络等….），利用准入控制来判断是否可以允许操作

对于API Server来说，可以使用Webhook 模式来实现自己的认证、鉴权和准入控制功能。也可以集成第三方的认证、鉴权和准入控制。

十二、数据存储

Kubernetes是使用etcd来进行数据存储的。

在 Kubernetes 项目中，一个 API 对象在 Etcd 里的完整资源路径，是由：Group（API 组）、Version（API 版本）和 Resource（API 资源类型）三个部分组成的。通过这样的结构，整个 Kubernetes 里的所有 API 对象，实际上就可以用如下的树形结构表示出来：

我们来看一下Kubernetes 是如何对 Resource、Group 和 Version 进行解析，从而在 Kubernetes 项目里找到 CronJob 对象的定义。

1. Kubernetes 会匹配 API 对象的组。
2. Kubernetes 会进一步匹配到 API 对象的版本号。
3. Kubernetes 会匹配 API 对象的资源类型。

在前面匹配到正确的版本之后，Kubernetes 就知道，我要创建的原来是一个 /apis/batch/v2alpha1 下的 CronJob 对象。

十三、 以一个简单的任务为例子，解释整个系统的运作过程

以Kubernetes使用默认调度器，为一个新创建出来的Pod，寻找一个最合适的节点（Node）为例。

默认调度器会首先调用一组叫作 Predicate 的调度算法，来检查每个 Node。然后，再调用一组叫作 Priority 的调度算法，来给上一步得到的结果里的每个 Node 打分。最终的调度结果，就是得分最高的那个 Node。然后调度器会在Pod的spec.nodeName 字段填上调度结果的节点名字。到这里调度器对一个 Pod 的调度就完成了。

搭建rook ceph存储集群

一、安装Rook集群

https://www.rook.io/docs/rook/v1.8/quickstart.html

1. 下载代码

   1 2	git clone --single-branch --branch v1.8.3 https://github.com/rook/rook.git cd rook/deploy/examples

2. 修改operator.yaml文件（主要是修改镜像文件下载地址，无法直接在国内网络环境下从k8s.gcr.io下载镜像，最好使用翻墙工具在自己电脑上下好镜像然后在生产服务器上直接使用）。如下图所示：

   1 2 3 4 5 6

   ROOK_CSI_CEPH_IMAGE: "testharbor.zuoyejia.com/k8s_image/cephcsi@sha256:19634b6ef9fc6df2902cf6ff0b3dbccc56a6663d0cbfd065da44ecd2f955d848" ROOK_CSI_REGISTRAR_IMAGE: "testharbor.zuoyejia.com/k8s_image/csi-node-driver-registrar@sha256:01b341312ea19cefc29f46fa0dd54255530b9039dd80834f50d582ecd93cc3ca" ROOK_CSI_RESIZER_IMAGE: "testharbor.zuoyejia.com/k8s_image/csi-resizer@sha256:d2d2e429a0a87190ee73462698a02a08e555055246ad87ad979b464b999fedae" ROOK_CSI_PROVISIONER_IMAGE: "testharbor.zuoyejia.com/k8s_image/csi-provisioner@sha256:bbae7cde811054f6a51060ba7a42d8bf2469b8c574abb50fec8b46c13e32541e" ROOK_CSI_SNAPSHOTTER_IMAGE: "testharbor.zuoyejia.com/k8s_image/csi-snapshotter@sha256:551b9692943f915b5ee4b7274e3a918692a6175bb028f1f0236a38596c46cbe0" ROOK_CSI_ATTACHER_IMAGE: "testharbor.zuoyejia.com/k8s_image/csi-attacher@sha256:221c1c6930fb1cb93b57762a74ccb59194c4c74a63c0fd49309d1158d4f8c72c"

   

3. 修改cluster.yaml文件（主要修改镜像地址、mon数量、mgr数量、dashboard的ssl）

   • mgr数量：Ceph集群需要可用，所以最好配置为2（这里有个深坑这里的高可用是主备模式，在配置dashboard的时候需要注意）

   • dashboard的ssl：修改这个是因为我们使用ingress的方式，并且在ingress的前面还有一层是华为云的ELB（华为云的ELB不支持后端协议为tcp，所以这里只能将dashboard的ssl关掉，在ELB层配置https），

     如下图所示：

     1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36

     mon:    # Set the number of mons to be started. Generally recommended to be 3.    # For highest availability, an odd number of mons should be specified.   count: 3    # The mons should be on unique nodes. For production, at least 3 nodes are recommended for this reason.    # Mons should only be allowed on the same node for test environments where data loss is acceptable.   allowMultiplePerNode: false mgr:    # When higher availability of the mgr is needed, increase the count to 2.    # In that case, one mgr will be active and one in standby. When Ceph updates which    # mgr is active, Rook will update the mgr services to match the active mgr.   count: 2   modules:      # Several modules should not need to be included in this list. The "dashboard" and "monitoring" modules      # are already enabled by other settings in the cluster CR.      - name: pg_autoscaler       enabled: true  # enable the ceph dashboard for viewing cluster status dashboard:   enabled: true    # serve the dashboard under a subpath (useful when you are accessing the dashboard via a reverse proxy)    # urlPrefix: /ceph-dashboard    # serve the dashboard at the given port.    # port: 8443    # serve the dashboard using SSL    # ssl: true  # enable prometheus alerting for cluster monitoring:    # requires Prometheus to be pre-installed   enabled: false    # namespace to deploy prometheusRule in. If empty, namespace of the cluster will be used.    # Recommended:    # If you have a single rook-ceph cluster, set the rulesNamespace to the same namespace as the cluster or keep it empty.    # If you have multiple rook-ceph clusters in the same k8s cluster, choose the same namespace (ideally, namespace with prometheus    # deployed) to set rulesNamespace for all the clusters. Otherwise, you will get duplicate alerts with multiple alert definitions.   rulesNamespace: rook-ceph

     

4. 部署Rook Operator

   1 2	cd rook/deploy/examples kubectl apply -f crds.yaml -f common.yaml -f operator.yaml

5. 创建Ceph集群

   1	kubectl apply -f cluster.yaml

6. 检测Ceph集群是否正常。如下图所示

   1	kubectl -n rook-ceph get pod

   

二、搭建Ceph Dashboard面板

1. 配置在集群外部查看Dashboard。这里我们使用service http的NodePort模式。这个时候我们会发现Dashboard并不可以访问，解决办法看2

   1 2	cd rook/deploy/examples kubectl create -f dashboard-external-http.yaml

2. 检测mgr主备高可用模式下哪个pod真正能用
   由于主备模式下有一个Pod是不能用的，所以在配置service的时候可能代理的mgr pod不能用，所以导致Dashboard不能访问。

   • 找到第一步查看上面的svc。我们看到pod所在的端口(targetPort)是7000

     1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20

     apiVersion: v1 kind: Service metadata: name: rook-ceph-mgr-dashboard-external-http namespace: rook-ceph # namespace:cluster labels:   app: rook-ceph-mgr   rook_cluster: rook-ceph # namespace:cluster spec: ports:   - name: dashboard     port: 7000     protocol: TCP     targetPort: 7000 selector:   app: rook-ceph-mgr   ceph_daemon_id: a   rook_cluster: rook-ceph sessionAffinity: None type: NodePort

   • 找到mgr的两个podIP地址

     1	kubectl -n rook-ceph get pod -owide

     

   • 使用curl访问7000端口查看哪个Pod正常返回

     1 2	curl 10.244.8.83:7000 curl 10.244.68.188:7000

     

   • 查看这两个pod的lables信息，并找到ceph_daemon_id信息

     1	kubectl get pod -n rook-ceph --show-labels

     

   • 修改dashboard-external-http.yaml文件。在selector中加上可用Pod的ceph_daemon_id

     1 2 3 4

     selector:   app: rook-ceph-mgr   ceph_daemon_id: a   rook_cluster: rook-ceph

     

3. 重新加载dashboard-external-http.yaml service。这次发现使用公网ip+Nodeport可以正常访问Dashboard

   1	kubectl apply -f dashboard-external-http.yaml

4. 创建 ingress文件dashboard-ingress-http.yaml

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25

   # # This example is for Kubernetes running an ngnix-ingress # and an ACME (e.g. Let's Encrypt) certificate service # # The nginx-ingress annotations support the dashboard # running using HTTPS with a self-signed certificate # apiVersion: networking.k8s.io/v1 kind: Ingress metadata: name: rook-ceph-mgr-dashboard namespace: rook-ceph # namespace:cluster spec: ingressClassName: nginx rules:   - host: rookceph.zuoyejia.com     http:       paths:         - path: /           pathType: Prefix           backend:             service:               name: rook-ceph-mgr-dashboard-external-http               port:                 number: 7000

5. 加载ingress

   1	kubectl apply -f dashboard-ingress-http.yaml

6. 注意：本文第4、5部需要使用ingress-nginx和华为云的ELB。如果没有的话直接使用官方文档上的文件。再次提醒，service上必须加上可以访问正常Pod的选择器。

三、Ceph存储

1.1 块存储(RBD)

RDB： RADOS Block Devices

RADOS： Reliable, Autonomic Distributed Object Store

1. 配置
   RWO:（ReadWriteOnce）
   常用 块存储 。RWO模式；STS删除，pvc不会删除，需要自己手动维护

   https://www.rook.io/docs/rook/v1.8/ceph-block.html

   1	kubectl create -f deploy/examples/csi/rbd/storageclass.yaml

1.2 共享文件存储(CephFS)

1. 配置
   常用 文件存储。 RWX模式；如：10个Pod共同操作一个地方
   https://rook.io/docs/rook/v1.8/ceph-filesystem.html

   1 2 3

   cd rook kubectl apply -f filesystem.yaml kubectl create -f deploy/examples/csi/cephfs/storageclass.yaml

四、卸载Rook Ceph

参考：https://rook.io/docs/rook/v1.8/ceph-teardown.html

1. 清理集群

1

rm -rf /var/lib/rook

2.删除块和部署的文件

1
2
3
4

kubectl delete -f crds.yaml -f common.yaml -f operator.yaml
kubectl delete -f cluster.yaml
kubectl delete -n rook-ceph cephblockpool replicapool
kubectl delete storageclass rook-ceph-block

3.删除CephCluster CRD

1
2
3
4
5
6

# 1.编辑CephCluster并添加cleanupPolicy
# 2.删除CephClusterCR
# 3.确认已删除集群 CR
kubectl -n rook-ceph patch cephcluster rook-ceph --type merge -p '{"spec":{"cleanupPolicy":{"confirmation":"yes-really-destroy-data"}}}'
kubectl -n rook-ceph delete cephcluster rook-ceph
kubectl -n rook-ceph get cephcluster

4.删除Operator 及相关资源

1
2
3

kubectl delete -f operator.yaml
kubectl delete -f common.yaml
kubectl delete -f crds.yaml

5.删除主机上的数据

1
2
3
4
5
6
7
8
9
10

DISK="/dev/vdb"
sgdisk --zap-all $DISK
dd if=/dev/zero of="$DISK" bs=1M count=100 oflag=direct,dsync
# 如果是SSD
# blkdiscard $DISK
partprobe $DISK

ls /dev/mapper/ceph-* | xargs -I% -- dmsetup remove %
rm -rf /dev/ceph-*
rm -rf /dev/mapper/ceph--*

6.故障排除

• 查看Pod

  1	kubectl -n rook-ceph get pod

• 查看集群CRD

  1	kubectl -n rook-ceph get cephcluster

• 删除CRD

  1 2 3 4 5

  # 删除CRD for CRD in $(kubectl get crd -n rook-ceph | awk '/ceph.rook.io/ {print $1}'); do kubectl get -n rook-ceph "$CRD" -o name | \ xargs -I {} kubectl patch -n rook-ceph {} --type merge -p '{"metadata":{"finalizers": [null]}}' done

• 如果如果namespace仍然停留在Terminting状态，可以检查哪些资源正在阻止删除并删除finalizers并删除

  1 2	kubectl api-resources --verbs=list --namespaced -o name \ | xargs -n 1 kubectl get --show-kind --ignore-not-found -n rook-ceph

• 删除finalizers资源

  1 2 3 4 5 6

  kubectl -n rook-ceph patch configmap rook-ceph-mon-endpoints --type merge -p '{"metadata":{"finalizers": []}}' kubectl -n rook-ceph patch secrets rook-ceph-mon --type merge -p '{"metadata":{"finalizers": []}}' # 如果cluster和replicapool存在执行下面命令 kubectl -n rook-ceph patch cephclusters.ceph.rook.io rook-ceph -p '{"metadata":{"finalizers": []}}' --type=merge kubectl -n rook-ceph patch cephblockpool.ceph.rook.io replicapool -p '{"metadata":{"finalizers": []}}' --type=merge

搭建Keepalived + Nginx高可用服务集群

一、操作场景

虚拟IP（VIP）主要用于弹性云服务器的主备切换，达到高可用性HA（High Availability）的目的。当主服务器发生故障无法对外提供服务时，动态将虚拟IP切换到备服务器，继续对外提供服务。

本文档以弹性云服务器的CentOS 7.5 64位操作系统为例，实现Keepalived+Nginx高可用服务集群搭建。

二、背景知识

三、网络拓扑

数据规划如下：

表1 数据规划

实现方式如下：

• 将2台同子网的弹性云服务器配置Keepalived，一台为主服务器，一台为备份服务器。
• 将这2台弹性云服务器绑定同一个虚拟IP。
• 将虚拟IP与弹性公网IP绑定，从互联网可以访问绑定了该虚拟IP地址的主备云服务器。

图1 组网图

三、操作步骤

1. nginx

1.1 在ECS-HA1和ECS-HA2分别安装Nginx

• RPM包地址：http://nginx.org/packages/centos/7/x86_64/RPMS/

  1 2	$ wget http://nginx.org/packages/centos/7/x86_64/RPMS/nginx-1.20.2-1.el7.ngx.x86_64.rpm $ rpm -ivh nginx-1.20.2-1.el7.ngx.x86_64.rpm

• yum安装

  1 2 3 4

  $ yum install nginx $ systemctl start nginx $ systemctl enable nginx $ systemctl status nginx

1.2 配置Nginx

• 主机ECS-HA1 nginx.conf配置

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84

  # For more information on configuration, see: # * Official English Documentation: http://nginx.org/en/docs/ # * Official Russian Documentation: http://nginx.org/ru/docs/ user nginx; worker_processes auto; error_log /mnt/sse/log/nginx/error.log; pid /run/nginx.pid; # Load dynamic modules. See /usr/share/doc/nginx/README.dynamic. include /usr/share/nginx/modules/*.conf; events { worker_connections 1024; } http { log_format main '$status $body_bytes_sent "$http_referer" ' '"$http_user_agent" "$http_x_forwarded_for"' '"rl=$request_length" $remote_addr - $remote_user [$time_local] "$request"' 'rt=$request_time uct="$upstream_connect_time" uht="$upstream_header_time" urt="$upstream_response_time"'; access_log /mnt/sse/log/nginx/access.log main; sendfile on; tcp_nopush on; tcp_nodelay on; keepalive_timeout 65; types_hash_max_size 4096; include /etc/nginx/mime.types; default_type application/octet-stream; # Load modular configuration files from the /etc/nginx/conf.d directory. # See http://nginx.org/en/docs/ngx_core_module.html#include # for more information. include /etc/nginx/conf.d/*.conf; server { listen 80; listen [::]:80; server_name _; root /usr/share/nginx/html; # Load configuration files for the default server block. include /etc/nginx/default.d/*.conf; error_page 404 /404.html; location = /404.html { } error_page 500 502 503 504 /50x.html; location = /50x.html { } } # Settings for a TLS enabled server. # # server { # listen 443 ssl http2; # listen [::]:443 ssl http2; # server_name _; # root /usr/share/nginx/html; # # ssl_certificate "/etc/pki/nginx/server.crt"; # ssl_certificate_key "/etc/pki/nginx/private/server.key"; # ssl_session_cache shared:SSL:1m; # ssl_session_timeout 10m; # ssl_ciphers HIGH:!aNULL:!MD5; # ssl_prefer_server_ciphers on; # # # Load configuration files for the default server block. # include /etc/nginx/default.d/*.conf; # # error_page 404 /404.html; # location = /40x.html { # } # # error_page 500 502 503 504 /50x.html; # location = /50x.html { # } # } }

• 主机ECS-HA1 /usr/share/nginx/html/index.html修改

  1	Welcome to ECS-HA1

• 备机ECS-HA2 nginx.conf配置

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84

  # For more information on configuration, see: # * Official English Documentation: http://nginx.org/en/docs/ # * Official Russian Documentation: http://nginx.org/ru/docs/ user nginx; worker_processes auto; error_log /mnt/sse/log/nginx/error.log; pid /run/nginx.pid; # Load dynamic modules. See /usr/share/doc/nginx/README.dynamic. include /usr/share/nginx/modules/*.conf; events { worker_connections 1024; } http { log_format main '$status $body_bytes_sent "$http_referer" ' '"$http_user_agent" "$http_x_forwarded_for"' '"rl=$request_length" $remote_addr - $remote_user [$time_local] "$request"' 'rt=$request_time uct="$upstream_connect_time" uht="$upstream_header_time" urt="$upstream_response_time"'; access_log /mnt/sse/log/nginx/access.log main; sendfile on; tcp_nopush on; tcp_nodelay on; keepalive_timeout 65; types_hash_max_size 4096; include /etc/nginx/mime.types; default_type application/octet-stream; # Load modular configuration files from the /etc/nginx/conf.d directory. # See http://nginx.org/en/docs/ngx_core_module.html#include # for more information. include /etc/nginx/conf.d/*.conf; server { listen 80; listen [::]:80; server_name _; root /usr/share/nginx/html; # Load configuration files for the default server block. include /etc/nginx/default.d/*.conf; error_page 404 /404.html; location = /404.html { } error_page 500 502 503 504 /50x.html; location = /50x.html { } } # Settings for a TLS enabled server. # # server { # listen 443 ssl http2; # listen [::]:443 ssl http2; # server_name _; # root /usr/share/nginx/html; # # ssl_certificate "/etc/pki/nginx/server.crt"; # ssl_certificate_key "/etc/pki/nginx/private/server.key"; # ssl_session_cache shared:SSL:1m; # ssl_session_timeout 10m; # ssl_ciphers HIGH:!aNULL:!MD5; # ssl_prefer_server_ciphers on; # # # Load configuration files for the default server block. # include /etc/nginx/default.d/*.conf; # # error_page 404 /404.html; # location = /40x.html { # } # # error_page 500 502 503 504 /50x.html; # location = /50x.html { # } # } }

• 备机ECS-HA2 /usr/share/nginx/html/index.html修改

  1	Welcome to ECS-HA2

2. Keepalived

1.1 在ECS-HA1和ECS-HA2分别安装Keepalived

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

$ wget --no-check-certificate https://www.keepalived.org/software/keepalived-2.2.4.tar.gz
$ tar -xvzf keepalived-2.2.4.tar.gz
$ yum -y install openssl-devel
$ cd keepalived-2.2.4/
$ ./configure --prefix=/usr/local/keepalived
$ make
$ make install
$ mkdir /etc/keepalived
$ cp /usr/local/keepalived/etc/keepalived/keepalived.conf /etc/keepalived/keepalived.conf
$ cp /usr/local/keepalived/etc/sysconfig/keepalived /etc/sysconfig/keepalived
# 修改/lib/systemd/system/keepalived.service，设置EnvironmentFile=-/etc/sysconfig/keepalived
$ vim /lib/systemd/system/keepalived.service
$ systemctl start keepalived
$ systemctl enable keepalived
$ systemctl status keepalived.service

1.2 配置主机ECS-HA1Keepalived

• 编辑keepalived.conf

  1	vim /etc/keepalived/keepalived.conf

• 修改keepalived.conf

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40

  ! Configuration File for keepalived global_defs { router_id master-node } vrrp_script chk_http_port { script "/etc/keepalived/chk_nginx.sh" interval 2 weight -5 fall 2 rise 1 } vrrp_instance VI_1 { state MASTER interface eth0 mcast_src_ip 192.168.192.39 virtual_router_id 51 priority 101 advert_int 1 authentication { auth_type PASS auth_pass 1111 } unicast_src_ip 192.168.192.39 virtual_ipaddress { 192.168.192.100 } notify_master "/etc/keepalived/keepalived_notify.sh MASTER" # 当切换到master状态时执行脚本 notify_backup "/etc/keepalived/keepalived_notify.sh BACKUP" # 当切换到backup状态时执行脚本 notify_fault "/etc/keepalived/keepalived_notify.sh FAULT" # 当切换到fault状态时执行脚本 notify_stop "/etc/keepalived/keepalived_notify.sh STOP" # 当切换到stop状态时执行脚本 garp_master_delay 1 # 设置当切为主状态后多久更新ARP缓存 garp_master_refresh 5 # 设置主节点发送ARP报文的时间间隔 # 跟踪接口，里面任意一块网卡出现问题，都会进入故障(FAULT)状态 track_interface { eth0 } track_script { chk_http_port } }

• 创建/etc/keepalived/chk_nginx.sh脚本文件，使用chmod +x /etc/keepalived/chk_nginx.sh将文件变成可执行文件，并增加如下代码

  1 2 3 4 5 6 7 8 9 10

  #!/bin/bash counter=$(ps -C nginx --no-heading|wc -l) if [ "${counter}" = "0" ]; then systemctl start nginx.service sleep 2 counter=$(ps -C nginx --no-heading|wc -l) if [ "${counter}" = "0" ]; then systemctl stop keepalived.service fi fi

• 创建/etc/keepalived/keepalived_notify脚本文件，使用chmod +x /etc/keepalived/keepalived_notify将文件变成可执行文件，并增加如下代码

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41

  #!/usr/bin/env bash # Copyright 2021 He Chen <chenhe@zuoyejia.com>. All rights reserved. # Use of this source code is governed by a MIT style # license that can be found in the LICENSE file. # /etc/keepalived/keepalived_notify.sh log_file=/var/log/keepalived.log sse::keepalived::mail() { # 这里可以添加email逻辑，当keepalived变动时及时告警 : sendEmail -f chenhe@zuoyejia.com -t chenhe@zuoyejia.com -s "smtp.exmail.qq.com:587" -u "作业家Nginx+Keepalived状态通知" -o message-charset=utf-8 -xu chenhe@zuoyejia.com -xp agcCeRn3v42Cohkk -m "$1" } sse::keepalived::log() { echo "[`date '+%Y-%m-%d %T'`] $1" >> ${log_file} } [ ! -d /var/keepalived/ ] && mkdir -p /var/keepalived/ case "$1" in "MASTER" ) sse::keepalived::log "notify_master" sse::keepalived::mail "notify_master" ;; "BACKUP" ) sse::keepalived::log "notify_backup" sse::keepalived::mail "notify_backup" ;; "FAULT" ) sse::keepalived::log "notify_fault" sse::keepalived::mail "notify_fault" ;; "STOP" ) sse::keepalived::log "notify_stop" sse::keepalived::mail "notify_stop" ;; *) sse::keepalived::log "keepalived_notify.sh: state error!" ;; esac

1.3 配置备机ECS-HA2 Keepalived

• 编辑keepalived.conf

  1	vim /etc/keepalived/keepalived.conf

• 修改keepalived.conf

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42

  ! Configuration File for keepalived global_defs { router_id master-node script_user root enable_script_security } vrrp_script chk_http_port { script "/etc/keepalived/chk_nginx.sh" interval 2 weight -5 fall 2 rise 1 } vrrp_instance VI_1 { state BACKUP interface eth0 mcast_src_ip 192.168.192.41 virtual_router_id 51 priority 100 advert_int 1 authentication { auth_type PASS auth_pass 1111 } unicast_src_ip 192.168.192.41 virtual_ipaddress { 192.168.192.100 } notify_master "/etc/keepalived/keepalived_notify.sh MASTER" # 当切换到master状态时执行脚本 notify_backup "/etc/keepalived/keepalived_notify.sh BACKUP" # 当切换到backup状态时执行脚本 notify_fault "/etc/keepalived/keepalived_notify.sh FAULT" # 当切换到fault状态时执行脚本 notify_stop "/etc/keepalived/keepalived_notify.sh STOP" # 当切换到stop状态时执行脚本 garp_master_delay 1 # 设置当切为主状态后多久更新ARP缓存 garp_master_refresh 5 # 设置主节点发送ARP报文的时间间隔 # 跟踪接口，里面任意一块网卡出现问题，都会进入故障(FAULT)状态 track_interface { eth0 } track_script { chk_http_port } }

• 创建/etc/keepalived/chk_nginx.sh脚本文件，使用chmod +x /etc/keepalived/chk_nginx.sh将文件变成可执行文件，并增加如下代码

  1 2 3 4 5 6 7 8 9 10

  #!/bin/bash counter=$(ps -C nginx --no-heading|wc -l) if [ "${counter}" = "0" ]; then systemctl start nginx.service sleep 2 counter=$(ps -C nginx --no-heading|wc -l) if [ "${counter}" = "0" ]; then systemctl stop keepalived.service fi fi

• 创建/etc/keepalived/keepalived_notify脚本文件，使用chmod +x /etc/keepalived/keepalived_notify将文件变成可执行文件，并增加如下代码

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41

  #!/usr/bin/env bash # Copyright 2021 He Chen <chenhe@zuoyejia.com>. All rights reserved. # Use of this source code is governed by a MIT style # license that can be found in the LICENSE file. # /etc/keepalived/keepalived_notify.sh log_file=/var/log/keepalived.log sse::keepalived::mail() { # 这里可以添加email逻辑，当keepalived变动时及时告警 : sendEmail -f chenhe@zuoyejia.com -t chenhe@zuoyejia.com -s "smtp.exmail.qq.com:587" -u "作业家Nginx+Keepalived备机状态通知" -o message-charset=utf-8 -xu chenhe@zuoyejia.com -xp agcCeRn3v42Cohkk -m "$1" } sse::keepalived::log() { echo "[`date '+%Y-%m-%d %T'`] $1" >> ${log_file} } [ ! -d /var/keepalived/ ] && mkdir -p /var/keepalived/ case "$1" in "MASTER" ) sse::keepalived::log "nginx backpu notify_master" sse::keepalived::mail "nginx backpu notify_master" ;; "BACKUP" ) sse::keepalived::log "nginx backpu notify_backup" sse::keepalived::mail "nginx backpu notify_backup" ;; "FAULT" ) sse::keepalived::log "nginx backpu notify_fault" sse::keepalived::mail "nginx backpu notify_fault" ;; "STOP" ) sse::keepalived::log "nginx backpu notify_stop" sse::keepalived::mail "nginx backpu notify_stop" ;; *) sse::keepalived::log "nginx backpu keepalived_notify.sh: state error!" ;; esac

3. sendEmail

1.1 在ECS-HA1和ECS-HA2分别安装sendEmail

1

yum install sendEmail

1.2. 以腾讯企业邮箱为例，配置安全密码

• 登录腾讯企业邮箱，打开设置

  

• 选择安全设置，生成专用密码。（如果没开启安全登录，点击开启）
  

• 将安全密码写入/etc/keepalived/keepalived_notify脚本文件中

4. 重启ECS-HA1和ECS-HA2 Nginx、Keepalived

1
2

$ nginx -s reload
$ systemctl restart keepalived

5. 绑定虚拟IP

1. 登录管理控制台。
2. 选择“服务列表 > 网络 > 虚拟私有云”。
3. 在左侧导航栏选择“子网”。
4. 在“子网”列表中，单击子网名称。
5. 选择“IP地址管理”页签，在虚拟IP所在行的操作列下单击“绑定服务器”。
6. 在弹出的页面，选择ecs HA1和ecs HA2服务器。
7. 选择“IP地址管理”页签，在虚拟IP所在行的操作列下单击“绑定弹性公网IP”。
8. 在弹出的页面，选择弹性公网IP 116.63.115.186

6. 域名解析

添加A类DNS域名解析，将sseserverzj2.zuoyejia.com绑定为上述公网ip

7. 验证结果

1. 通过管理控制台远程登录到ecs-HA1。

2. 执行以下命令，查看虚拟IP是否有绑定到ecs-HA1的eth0网卡上。
   ip addr show

   如下图所示表示虚拟IP已经绑定到ecs-HA1的eth0网卡上。
   

3. 访问验证ecs-HA1
   

4. 执行以下命令，停止ecs-HA1上的keepalived服务。
   systemctl stop keepalived.service

5. 执行以下命令，查看服务器ecs-HA2是否有接管虚拟IP。
   ip addr show
   如果所示
   

6. 访问验证ecs-HA2
   

7. 查看自己的邮箱是否有keepalived的状态邮件
   

四、常用命令

Nginx

1
2

# 不停止对客户服务的情况下，重新加载配置文件
nginx -s reload

现在的互联网应用基本上都是采用分布式系统架构进行设计的，而很多分布式系统必备的一个基础软件就是消息队列。

消息队列要能支持组件通信消息的快速读写，而 Redis 本身支持数据的高速访问，正好可以满足消息队列的读写性能需求。不过，除了性能，消息队列还有其他的要求，所以，很多人都很关心一个问题：“Redis 适合做消息队列吗？”

其实，这个问题的背后，隐含着两方面的核心问题：

• 消息队列的消息存取需求是什么？
• Redis 如何实现消息队列的需求？

只有把这两方面的知识和实践经验串连起来，才能彻底理解基于 Redis 实现消息队列的技术实践。以后需要为分布式系统组件做消息队列选型时，就可以根据组件通信量和消息通信速度的要求，选择出适合的 Redis 消息队列方案。目前使用Redis做消息队列可以有如下三种方式。

一、消息队列的消息存取需求

在分布式系统中，当两个组件要基于消息队列进行通信时，一个组件会把要处理的数据以消息的形式传递给消息队列，然后，这个组件就可以继续执行其他操作了；远端的另一个组件从消息队列中把消息读取出来，再在本地进行处理。

假设组件 1 需要对采集到的数据进行求和计算，并写入数据库，但是，消息到达的速度很快，组件 1 没有办法及时地既做采集，又做计算，并且写入数据库。所以，我们可以使用基于消息队列的通信，让组件 1 把数据 x 和 y 保存为 JSON 格式的消息，再发到消息队列，这样它就可以继续接收新的数据了。组件 2 则异步地从消息队列中把数据读取出来，在服务器 2 上进行求和计算后，再写入数据库。这个过程如下图所示：

我们一般把消息队列中发送消息的组件称为生产者（例子中的组件 1），把接收消息的组件称为消费者（例子中的组件 2），下图展示了一个通用的消息队列的架构模型：

在使用消息队列时，消费者可以异步读取生产者消息，然后再进行处理。这样一来，即使生产者发送消息的速度远远超过了消费者处理消息的速度，生产者已经发送的消息也可以缓存在消息队列中，避免阻塞生产者，这是消息队列作为分布式组件通信的一大优势。

不过，消息队列在存取消息时，必须要满足三个需求，分别是消息保序、处理重复的消息和保证消息可靠性。

1. 需求一：消息保序

   虽然消费者是异步处理消息，但是，消费者仍然需要按照生产者发送消息的顺序来处理消息，避免后发送的消息被先处理了。对于要求消息保序的场景来说，一旦出现这种消息被乱序处理的情况，就可能会导致业务逻辑被错误执行，从而给业务方造成损失。

2. 需求二：重复消息处理

   消费者从消息队列读取消息时，有时会因为网络堵塞而出现消息重传的情况。此时，消费者可能会收到多条重复的消息。对于重复的消息，消费者如果多次处理的话，就可能造成一个业务逻辑被多次执行，如果业务逻辑正好是要修改数据，那就会出现数据被多次修改的问题了。

3. 需求三：消息可靠性保证

   消费者在处理消息的时候，还可能出现因为故障或宕机导致消息没有处理完成的情况。此时，消息队列需要能提供消息可靠性的保证，也就是说，当消费者重启后，可以重新读取消息再次进行处理，否则，就会出现消息漏处理的问题了。

二、基于 List 的消息队列解决方案

首先，我们先从最简单的场景开始讲起。

如果你的业务需求足够简单，想把 Redis 当作队列来使用，肯定最先想到的就是使用 List 这个数据类型。

因为 List 底层的实现就是一个「链表」，在头部和尾部操作元素，时间复杂度都是 O(1)，这意味着它非常符合消息队列的模型。

如果把 List 当作队列，你可以这么来用。

生产者使用 LPUSH 发布消息：

1
2
3
4

127.0.0.1:6379> LPUSH queue msg1
(integer) 1
127.0.0.1:6379> LPUSH queue msg2
(integer) 2

消费者这一侧，使用 RPOP 拉取消息：

1
2
3
4

127.0.0.1:6379> RPOP queue
"msg1"
127.0.0.1:6379> RPOP queue
"msg2"

这个模型非常简单，也很容易理解。

但这里有个小问题，当队列中已经没有消息了，消费者在执行 RPOP 时，会返回 NULL。

1
2
3

127.0.0.1:6379> RPOP queue
# 没消息了
(nil)

而我们在编写消费者逻辑时，一般是一个「死循环」，这个逻辑需要不断地从队列中拉取消息进行处理，伪代码一般会这么写：

1
2
3
4
5
6
7

while true:
    msg = redis.rpop("queue")
    // 没有消息，继续循环
    if msg == null:
        continue
    // 处理消息
    handle(msg)

如果此时队列为空，那消费者依旧会频繁拉取消息，这会造成「CPU 空转」，不仅浪费 CPU 资源，还会对 Redis 造成压力。

怎么解决这个问题呢？

也很简单，当队列为空时，我们可以「休眠」一会，再去尝试拉取消息。代码可以修改成这样：

1
2
3
4
5
6
7
8

while true:
    msg = redis.rpop("queue")
    // 没有消息，休眠2s
    if msg == null:
        sleep(2)
        continue
    // 处理消息        
    handle(msg)

这就解决了 CPU 空转问题。

这个问题虽然解决了，但又带来另外一个问题：当消费者在休眠等待时，有新消息来了，那消费者处理新消息就会存在「延迟」。

假设设置的休眠时间是 2s，那新消息最多存在 2s 的延迟。

要想缩短这个延迟，只能减小休眠的时间。但休眠时间越小，又有可能引发 CPU 空转问题。

鱼和熊掌不可兼得。

那如何做，既能及时处理新消息，还能避免 CPU 空转呢？

Redis 是否存在这样一种机制：如果队列为空，消费者在拉取消息时就「阻塞等待」，一旦有新消息过来，就通知我的消费者立即处理新消息呢？

幸运的是，Redis 确实提供了「阻塞式」拉取消息的命令：BRPOP / BLPOP，这里的 B 指的是阻塞（Block）。

现在，你可以这样来拉取消息了：

1
2
3
4
5
6
7

while true:
    // 没消息阻塞等待，0表示不设置超时时间
    msg = redis.brpop("queue", 0)
    if msg == null:
        continue
    // 处理消息
    handle(msg)

使用 BRPOP 这种阻塞式方式拉取消息时，还支持传入一个「超时时间」，如果设置为 0，则表示不设置超时，直到有新消息才返回，否则会在指定的超时时间后返回 NULL。

这个方案不错，既兼顾了效率，还避免了 CPU 空转问题，一举两得。

使用 BRPOP 这种阻塞式方式拉取消息时，还支持传入一个「超时时间」，如果设置为 0，则表示不设置超时，直到有新消息才返回，否则会在指定的超时时间后返回 NULL。

这个方案不错，既兼顾了效率，还避免了 CPU 空转问题，一举两得。

解决了消息处理不及时的问题，你可以再思考一下，这种队列模型，有什么缺点？

我们一起来分析一下：

1. 不支持重复消费：消费者拉取消息后，这条消息就从 List 中删除了，无法被其它消费者再次消费，即不支持多个消费者消费同一批数据
2. 消息丢失：消费者拉取到消息后，如果发生异常宕机，那这条消息就丢失了

第一个问题是功能上的，使用 List 做消息队列，它仅仅支持最简单的，一组生产者对应一组消费者，不能满足多组生产者和消费者的业务场景。

第二个问题就比较棘手了，因为从 List 中 POP 一条消息出来后，这条消息就会立即从链表中删除了。也就是说，无论消费者是否处理成功，这条消息都没办法再次消费了。

这也意味着，如果消费者在处理消息时异常宕机，那这条消息就相当于丢失了。

针对这 2 个问题怎么解决呢？我们一个个来看。

三、基于 Pub/Sub 的消息队列解决方案

从名字就能看出来，这个模块是 Redis 专门是针对「发布/订阅」这种队列模型设计的。

它正好可以解决前面提到的第一个问题：重复消费。

即多组生产者、消费者的场景，我们来看它是如何做的。

Redis 提供了 PUBLISH / SUBSCRIBE 命令，来完成发布、订阅的操作。

假设你想开启 2 个消费者，同时消费同一批数据，就可以按照以下方式来实现。

首先，使用 SUBSCRIBE 命令，启动 2 个消费者，并「订阅」同一个队列。

1
2
3
4
5
6

# 2个消费者 都订阅一个队列
127.0.0.1:6379> SUBSCRIBE queue
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "queue"
3) (integer) 1

此时，2 个消费者都会被阻塞住，等待新消息的到来。

之后，再启动一个生产者，发布一条消息。

1
2

127.0.0.1:6379> PUBLISH queue msg1
(integer) 1

这时，2 个消费者就会解除阻塞，收到生产者发来的新消息。

1
2
3
4
5

127.0.0.1:6379> SUBSCRIBE queue
# 收到新消息
1) "message"
2) "queue"
3) "msg1"

使用 Pub/Sub 这种方案，既支持阻塞式拉取消息，还很好地满足了多组消费者，消费同一批数据的业务需求。

除此之外，Pub/Sub 还提供了「匹配订阅」模式，允许消费者根据一定规则，订阅「多个」自己感兴趣的队列。

1
2
3
4
5
6

# 订阅符合规则的队列
127.0.0.1:6379> PSUBSCRIBE queue.*
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "queue.*"
3) (integer) 1

这里的消费者，订阅了 queue.* 相关的队列消息。

之后，生产者分别向 queue.p1 和 queue.p2 发布消息。

1
2
3
4

127.0.0.1:6379> PUBLISH queue.p1 msg1
(integer) 1
127.0.0.1:6379> PUBLISH queue.p2 msg2
(integer) 1

这时再看消费者，它就可以接收到这 2 个生产者的消息了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14

127.0.0.1:6379> PSUBSCRIBE queue.*
Reading messages... (press Ctrl-C to quit)
...
# 来自queue.p1的消息
1) "pmessage"
2) "queue.*"
3) "queue.p1"
4) "msg1"

# 来自queue.p2的消息
1) "pmessage"
2) "queue.*"
3) "queue.p2"
4) "msg2"

我们可以看到，Pub/Sub 最大的优势就是，支持多组生产者、消费者处理消息。

讲完了它的优点，那它有什么缺点呢？

其实，Pub/Sub 最大问题是：丢数据。

如果发生以下场景，就有可能导致数据丢失：

1. 消费者下线
2. Redis 宕机
3. 消息堆积

究竟是怎么回事？

这其实与 Pub/Sub 的实现方式有很大关系。

Pub/Sub 在实现时非常简单，它没有基于任何数据类型，也没有做任何的数据存储，它只是单纯地为生产者、消费者建立「数据转发通道」，把符合规则的数据，从一端转发到另一端。

一个完整的发布、订阅消息处理流程是这样的：

1. 消费者订阅指定队列，Redis 就会记录一个映射关系：队列->消费者
2. 生产者向这个队列发布消息，那 Redis 就从映射关系中找出对应的消费者，把消息转发给它

整个过程中，没有任何的数据存储，一切都是实时转发的。

这种设计方案，就导致了上面提到的那些问题。

例如，如果一个消费者异常挂掉了，它再重新上线后，只能接收新的消息，在下线期间生产者发布的消息，因为找不到消费者，都会被丢弃掉。

如果所有消费者都下线了，那生产者发布的消息，因为找不到任何一个消费者，也会全部「丢弃」。

所以，当你在使用 Pub/Sub 时，一定要注意：消费者必须先订阅队列，生产者才能发布消息，否则消息会丢失。

这也是前面讲例子时，我们让消费者先订阅队列，之后才让生产者发布消息的原因。

另外，因为 Pub/Sub 没有基于任何数据类型实现，所以它也不具备「数据持久化」的能力。

也就是说，Pub/Sub 的相关操作，不会写入到 RDB 和 AOF 中，当 Redis 宕机重启，Pub/Sub 的数据也会全部丢失。

最后，我们来看 Pub/Sub 在处理「消息积压」时，为什么也会丢数据？

当消费者的速度，跟不上生产者时，就会导致数据积压的情况发生。

如果采用 List 当作队列，消息积压时，会导致这个链表很长，最直接的影响就是，Redis 内存会持续增长，直到消费者把所有数据都从链表中取出。

但 Pub/Sub 的处理方式却不一样，当消息积压时，有可能会导致消费失败和消息丢失！

这是怎么回事？

还是回到 Pub/Sub 的实现细节上来说。

每个消费者订阅一个队列时，Redis 都会在 Server 上给这个消费者在分配一个「缓冲区」，这个缓冲区其实就是一块内存。

当生产者发布消息时，Redis 先把消息写到对应消费者的缓冲区中。

之后，消费者不断地从缓冲区读取消息，处理消息。看到了么，整个过程中，没有任何的数据存储，一切都是实时转发的。

这种设计方案，就导致了上面提到的那些问题。

例如，如果一个消费者异常挂掉了，它再重新上线后，只能接收新的消息，在下线期间生产者发布的消息，因为找不到消费者，都会被丢弃掉。

如果所有消费者都下线了，那生产者发布的消息，因为找不到任何一个消费者，也会全部「丢弃」。

所以，当你在使用 Pub/Sub 时，一定要注意：消费者必须先订阅队列，生产者才能发布消息，否则消息会丢失。

这也是前面讲例子时，我们让消费者先订阅队列，之后才让生产者发布消息的原因。

另外，因为 Pub/Sub 没有基于任何数据类型实现，所以它也不具备「数据持久化」的能力。

也就是说，Pub/Sub 的相关操作，不会写入到 RDB 和 AOF 中，当 Redis 宕机重启，Pub/Sub 的数据也会全部丢失。

最后，我们来看 Pub/Sub 在处理「消息积压」时，为什么也会丢数据？

当消费者的速度，跟不上生产者时，就会导致数据积压的情况发生。

如果采用 List 当作队列，消息积压时，会导致这个链表很长，最直接的影响就是，Redis 内存会持续增长，直到消费者把所有数据都从链表中取出。

但 Pub/Sub 的处理方式却不一样，当消息积压时，有可能会导致消费失败和消息丢失！

这是怎么回事？

还是回到 Pub/Sub 的实现细节上来说。

每个消费者订阅一个队列时，Redis 都会在 Server 上给这个消费者在分配一个「缓冲区」，这个缓冲区其实就是一块内存。

当生产者发布消息时，Redis 先把消息写到对应消费者的缓冲区中。

之后，消费者不断地从缓冲区读取消息，处理消息。看到了么，整个过程中，没有任何的数据存储，一切都是实时转发的。

这种设计方案，就导致了上面提到的那些问题。

例如，如果一个消费者异常挂掉了，它再重新上线后，只能接收新的消息，在下线期间生产者发布的消息，因为找不到消费者，都会被丢弃掉。

如果所有消费者都下线了，那生产者发布的消息，因为找不到任何一个消费者，也会全部「丢弃」。

所以，当你在使用 Pub/Sub 时，一定要注意：消费者必须先订阅队列，生产者才能发布消息，否则消息会丢失。

这也是前面讲例子时，我们让消费者先订阅队列，之后才让生产者发布消息的原因。

另外，因为 Pub/Sub 没有基于任何数据类型实现，所以它也不具备「数据持久化」的能力。

也就是说，Pub/Sub 的相关操作，不会写入到 RDB 和 AOF 中，当 Redis 宕机重启，Pub/Sub 的数据也会全部丢失。

最后，我们来看 Pub/Sub 在处理「消息积压」时，为什么也会丢数据？

当消费者的速度，跟不上生产者时，就会导致数据积压的情况发生。

如果采用 List 当作队列，消息积压时，会导致这个链表很长，最直接的影响就是，Redis 内存会持续增长，直到消费者把所有数据都从链表中取出。

但 Pub/Sub 的处理方式却不一样，当消息积压时，有可能会导致消费失败和消息丢失！

这是怎么回事？

还是回到 Pub/Sub 的实现细节上来说。

每个消费者订阅一个队列时，Redis 都会在 Server 上给这个消费者在分配一个「缓冲区」，这个缓冲区其实就是一块内存。

当生产者发布消息时，Redis 先把消息写到对应消费者的缓冲区中。

之后，消费者不断地从缓冲区读取消息，处理消息。

但是，问题就出在这个缓冲区上。

因为这个缓冲区其实是有「上限」的（可配置），如果消费者拉取消息很慢，就会造成生产者发布到缓冲区的消息开始积压，缓冲区内存持续增长。

如果超过了缓冲区配置的上限，此时，Redis 就会「强制」把这个消费者踢下线。

这时消费者就会消费失败，也会丢失数据。

如果你有看过 Redis 的配置文件，可以看到这个缓冲区的默认配置：client-output-buffer-limit pubsub 32mb 8mb 60。

它的参数含义如下：

• 32mb：缓冲区一旦超过 32MB，Redis 直接强制把消费者踢下线
• 8mb + 60：缓冲区超过 8MB，并且持续 60 秒，Redis 也会把消费者踢下线

Pub/Sub 的这一点特点，是与 List 作队列差异比较大的。

从这里你应该可以看出，List 其实是属于「拉」模型，而 Pub/Sub 其实属于「推」模型。

List 中的数据可以一直积压在内存中，消费者什么时候来「拉」都可以。

但 Pub/Sub 是把消息先「推」到消费者在 Redis Server 上的缓冲区中，然后等消费者再来取。

当生产、消费速度不匹配时，就会导致缓冲区的内存开始膨胀，Redis 为了控制缓冲区的上限，所以就有了上面讲到的，强制把消费者踢下线的机制。

好了，现在我们总结一下 Pub/Sub 的优缺点：

1. 支持发布 / 订阅，支持多组生产者、消费者处理消息
2. 消费者下线，数据会丢失
3. 不支持数据持久化，Redis 宕机，数据也会丢失
4. 消息堆积，缓冲区溢出，消费者会被强制踢下线，数据也会丢失

有没有发现，除了第一个是优点之外，剩下的都是缺点。

所以，很多人看到 Pub/Sub 的特点后，觉得这个功能很「鸡肋」。

也正是以上原因，Pub/Sub 在实际的应用场景中用得并不多。

目前只有哨兵集群和 Redis 实例通信时，采用了 Pub/Sub 的方案，因为哨兵正好符合即时通讯的业务场景。

我们再来看一下，Pub/Sub 有没有解决，消息处理时异常宕机，无法再次消费的问题呢？

其实也不行，Pub/Sub 从缓冲区取走数据之后，数据就从 Redis 缓冲区删除了，消费者发生异常，自然也无法再次重新消费。

好，现在我们重新梳理一下，我们在使用消息队列时的需求。

当我们在使用一个消息队列时，希望它的功能如下：

• 支持阻塞等待拉取消息
• 支持发布 / 订阅模式
• 消费失败，可重新消费，消息不丢失
• 实例宕机，消息不丢失，数据可持久化
• 消息可堆积

Redis 除了 List 和 Pub/Sub 之外，还有符合这些要求的数据类型吗？

其实，Redis 的作者也看到了以上这些问题，也一直在朝着这些方向努力着。

Redis 作者在开发 Redis 期间，还另外开发了一个开源项目 disque。

这个项目的定位，就是一个基于内存的分布式消息队列中间件。

但由于种种原因，这个项目一直不温不火。

终于，在 Redis 5.0 版本，作者把 disque 功能移植到了 Redis 中，并给它定义了一个新的数据类型：Streams。

下面我们就来看看，它能符合上面提到的这些要求吗？

四、基于 Streams 的消息队列解决方案

Streams 是 Redis 专门为消息队列设计的数据类型，我们来看 Stream 是如何解决上面这些问题的。

我们依旧从简单到复杂，依次来看 Stream 在做消息队列时，是如何处理的？

1. Streams 是如何实现消息队列需求的

首先，Stream 通过 XADD 和 XREAD 完成最简单的生产、消费模型：

下面我们来看，针对前面提到的消息队列要求，Stream 都是如何解决的？

1. Streams 通过 XADD 和 XREAD 完成最简单的生产、消费模型。
2. Streams 通过BLOCK 参数支持「阻塞式」拉取消息
3. Stream 通过XGROUP和XREADGROUP支持发布 / 订阅模式
4. Stream 通过XACK标记消息为「处理完成」
5. Stream 数据会写入到 RDB 和 AOF 做持久化
6. Stream 处理消息堆积方式，采用丢弃消息的方式
   • 在发布消息时，可以指定队列的最大长度，防止队列积压导致内存爆炸。当队列长度超过上限后，旧消息会被删除，只保留固定长度的新消息。XADD mystream MAXLEN 10000 * repo 1

2. Streams与专业的消息队列对比

其实，一个专业的消息队列，必须要做到两大块：

1. 消息不丢
2. 消息可堆积

前面我们讨论的重点，很大篇幅围绕的是第一点展开的。

这里我们换个角度，从一个消息队列的「使用模型」来分析一下，怎么做，才能保证数据不丢？

使用一个消息队列，其实就分为三大块：生产者、队列中间件、消费者。

消息是否会发生丢失，其重点也就在于以下 3 个环节：

1. 生产者会不会丢消息？
2. 消费者会不会丢消息？
3. 队列中间件会不会丢消息？

1) 生产者会不会丢消息？

当生产者在发布消息时，可能发生以下异常情况：

1. 消息没发出去：网络故障或其它问题导致发布失败，中间件直接返回失败
2. 不确定是否发布成功：网络问题导致发布超时，可能数据已发送成功，但读取响应结果超时了

如果是情况 1，消息根本没发出去，那么重新发一次就好了。

如果是情况 2，生产者没办法知道消息到底有没有发成功？所以，为了避免消息丢失，它也只能继续重试，直到发布成功为止。

生产者一般会设定一个最大重试次数，超过上限依旧失败，需要记录日志报警处理。

也就是说，生产者为了避免消息丢失，只能采用失败重试的方式来处理。

但发现没有？这也意味着消息可能会重复发送。

是的，在使用消息队列时，要保证消息不丢，宁可重发，也不能丢弃。

那消费者这边，就需要多做一些逻辑了。

对于敏感业务，当消费者收到重复数据数据时，要设计幂等逻辑，保证业务的正确性。

从这个角度来看，生产者会不会丢消息，取决于生产者对于异常情况的处理是否合理。

所以，无论是 Redis 还是专业的队列中间件，生产者在这一点上都是可以保证消息不丢的。

2) 消费者会不会丢消息？

这种情况就是我们前面提到的，消费者拿到消息后，还没处理完成，就异常宕机了，那消费者还能否重新消费失败的消息？

要解决这个问题，消费者在处理完消息后，必须「告知」队列中间件，队列中间件才会把标记已处理，否则仍旧把这些数据发给消费者。

这种方案需要消费者和中间件互相配合，才能保证消费者这一侧的消息不丢。

无论是 Redis 的 Stream，还是专业的队列中间件，例如 RabbitMQ、Kafka，其实都是这么做的。

所以，从这个角度来看，Redis 也是合格的。

3) 队列中间件会不会丢消息？

前面 2 个问题都比较好处理，只要客户端和服务端配合好，就能保证生产端、消费端都不丢消息。

但是，如果队列中间件本身就不可靠呢？

毕竟生产者和消费这都依赖它，如果它不可靠，那么生产者和消费者无论怎么做，都无法保证数据不丢。

在这个方面，Redis 其实没有达到要求。

Redis 在以下 2 个场景下，都会导致数据丢失。

1. AOF 持久化配置为每秒写盘，但这个写盘过程是异步的，Redis 宕机时会存在数据丢失的可能
2. 主从复制也是异步的，主从切换时，也存在丢失数据的可能（从库还未同步完成主库发来的数据，就被提成主库）

基于以上原因我们可以看到，Redis 本身的无法保证严格的数据完整性。

所以，如果把 Redis 当做消息队列，在这方面是有可能导致数据丢失的。

再来看那些专业的消息队列中间件是如何解决这个问题的？

像 RabbitMQ 或 Kafka 这类专业的队列中间件，在使用时，一般是部署一个集群，生产者在发布消息时，队列中间件通常会写「多个节点」，以此保证消息的完整性。这样一来，即便其中一个节点挂了，也能保证集群的数据不丢失。

也正因为如此，RabbitMQ、Kafka在设计时也更复杂。毕竟，它们是专门针对队列场景设计的。

但 Redis 的定位则不同，它的定位更多是当作缓存来用，它们两者在这个方面肯定是存在差异的。

最后，我们来看消息积压怎么办？

4) 消息积压怎么办？

因为 Redis 的数据都存储在内存中，这就意味着一旦发生消息积压，则会导致 Redis 的内存持续增长，如果超过机器内存上限，就会面临被 OOM 的风险。

所以，Redis 的 Stream 提供了可以指定队列最大长度的功能，就是为了避免这种情况发生。

但 Kafka、RabbitMQ 这类消息队列就不一样了，它们的数据都会存储在磁盘上，磁盘的成本要比内存小得多，当消息积压时，无非就是多占用一些磁盘空间，相比于内存，在面对积压时也会更加「坦然」。

综上，我们可以看到，把 Redis 当作队列来使用时，始终面临的 2 个问题：

1. Redis 本身可能会丢数据
2. 面对消息积压，Redis 内存资源紧张

到这里，Redis 是否可以用作队列，我想这个答案你应该会比较清晰了。

如果你的业务场景足够简单，对于数据丢失不敏感，而且消息积压概率比较小的情况下，把 Redis 当作队列是完全可以的。

而且，Redis 相比于 Kafka、RabbitMQ，部署和运维也更加轻量。

如果你的业务场景对于数据丢失非常敏感，而且写入量非常大，消息积压时会占用很多的机器资源，那么我建议你使用专业的消息队列中间件。

3. Streams操作命令简介

Streams提供了丰富的消息队列操作命令。

• XADD：插入消息，保证有序，可以自动生成全局唯一 ID。
• XREAD：用于读取消息，可以按 ID 读取数据。
• XGROUP CREATE：创建消费组。
• XREADGROUP：按消费组形式读取消息。
• XPENDING 和 XACK：XPENDING 命令可以用来查询每个消费组内所有消费者已读取但尚未确认的消息，而 XACK 命令用于向消息队列确认消息处理已完成。

1. XADD

语法格式：

XADD key ID field value [field value ...]

• key：队列名称，如果不存在就创建
• ID：消息 id，我们使用 * 表示由 redis 生成，可以自定义，但是要自己保证递增性。
• field value [field value …]，key-value类型数据

XADD 命令可以往消息队列中插入新消息，消息的格式是键 - 值对形式。对于插入的每一条消息，Streams 可以自动为其生成一个全局唯一的 ID。我们执行下面的命令，就可以往名称为 mqstream 的消息队列中插入一条消息。

下面命令组成如下：

• 消息的键是 repo
• 值是 5。
• 消息队列名称后面的*，表示让 Redis 为插入的数据自动生成一个全局唯一的 ID。可以看到，消息的全局唯一 ID 由两部分组成1631010568190-0。
  • 1631010568190：数据插入时，以毫秒为单位计算的当前服务器时间
  • 0：表示插入消息在当前毫秒内的消息序号，这是从 0 开始编号的。

1
2
3
4
5
6

# *表示让Redis自动生成消息ID
# XADD key ID field value [field value ...]
127.0.0.1:6379> XADD mqstream * repo 5
"1631010546467-0"
127.0.0.1:6379> XADD mqstream * repo 6
"1631010568190-0"

2. XREAD

语法格式：

XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] id [id ...]

• [COUNT count]：用来获取消息的数量
• [BLOCK milliseconds]：用来设置阻塞模式和阻塞超时时间，默认为非阻塞
• id [id …]：用来设置读取的起始 ID，相当于 where id > $id，阻塞模式中可以使用 $ 来获取最新的消息 ID，非阻塞模式下无意义。
• key：队列名

当消费者需要读取消息时，可以直接使用 XREAD 命令从消息队列中读取。

XREAD 在读取消息时，可以指定一个消息 ID，并从这个消息 ID 的下一条消息开始进行读取。

例如，我们可以执行下面的命令，从 ID 号为 1631010542237-0 的消息开始，读取后续的所有消息（示例中一共 3 条）。参数说明如下：

• block：当消息队列中没有消息时，一旦设置了 BLOCK 配置项，XREAD 就会阻塞，阻塞的时长可以在 BLOCK 配置项进行设置。BLOCK 0 表示阻塞等待，不设置超时时间。

1
2
3
4
5
6
7
8
9
10
11

127.0.0.1:6379> XREAD BLOCK 0 STREAMS mqstream 1631010542237-0
1) 1) "mqstream"
   2) 1) 1) "1631010544817-0"
         2) 1) "repo"
            2) "4"
      2) 1) "1631010546467-0"
         2) 1) "repo"
            2) "5"
      3) 1) "1631010568190-0"
         2) 1) "repo"
            2) "6"

看下面命令，其中，命令最后的“$”符号表示读取最新的消息，同时，我们设置了 block 10000 的配置项，10000 的单位是毫秒，表明 XREAD 在读取最新消息时，如果没有消息到来，XREAD 将阻塞 10000 毫秒（即 10 秒），然后再返回。下面命令中的 XREAD 执行后，消息队列 mqstream 中一直没有消息，所以，XREAD 在 10 秒后返回空值（nil）。

1
2
3

127.0.0.1:6379> XREAD BLOCK 10000 STREAMS mqstream $
(nil)
(10.00s)

3. XGROUP CREATE

语法格式：

1 2	> XGROUP [CREATE key groupname id-or-$] [SETID key groupname id-or-$] [DESTROY key groupname] [CREATECONSUMER key groupname consumername] [DELCONSUMER key groupname consumername] >

• [CREATE key groupname id-or-$]：在指定的 key 中创建分组，并且指定分组读取消息的起点，如果指定了0，分组将可以读取指定 key 的所有历史消息，如果指定了 $，分组将可以读取指定 key 的新消息，将不能读取历史消息。也可以指定任意的开始 ID。
• [SETID key groupname id-or-$]：重新给已存在的分组设置消息读取的起点。例如将起点设置为 0就可以重新读取所有的历史消息
• [DESTROY key groupname]：销毁指定 key 中的一个分组
• [CREATECONSUMER key groupname consumername]：在指定的 key 和指定的分组中创建一个消费者。当某个命令提及了新的消费者名称时，也会自动创建新的消费者。
• [DELCONSUMER key groupname consumername]：在指定的 key 和指定的分组中销毁一个消费者。
• $：表示从尾部开始消费，只接受新消息，当前 Stream 消息会全部忽略。

执行下面的命令，

1. XGROUP create mqstream group1 0：创建一个名为 group1 的消费组，这个消费组消费的消息队列是 mqstream。
2. XREADGROUP group group1 consumer1 streams mqstream >：让 group1 消费组里的消费者 consumer1 从 mqstream 中读取所有消息，其中，命令最后的参数“>”，表示从第一条尚未被消费的消息开始读取。因为在 consumer1 读取消息前，group1 中没有其他消费者读取过消息，所以，consumer1 就得到 mqstream 消息队列中的所有消息了（一共 6 条）。
3. XREADGROUP group group1 consumer1 streams mqstream >：再次执行上述2命令查看返回效果
4. XREADGROUP group group1 consumer2 streams mqstream 0：需要注意的是，消息队列中的消息一旦被消费组里的一个消费者读取了，就不能再被该消费组内的其他消费者读取了。比如说，我们执行完刚才的 XREADGROUP 命令后，再执行下面的命令，让 group1 内的 consumer2 读取消息时，consumer2 读到的就是空值，因为消息已经被 consumer1 读取完了.
5. XREADGROUP group group2 consumer1 count 1 streams mqstream >：使用消费组的目的是让组内的多个消费者共同分担读取消息，所以，我们通常会让每个消费者读取部分消息，从而实现消息读取负载在多个消费者间是均衡分布的。例如，我们执行下列命令，让 group2 中的 consumer1、2、3 各自读取一条消息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

# 1
127.0.0.1:6379> XGROUP create mqstream group1 0
OK
# 2
127.0.0.1:6379> XREADGROUP group group1 consumer1 streams mqstream >
1) 1) "mqstream"
   2) 1) 1) "1631010537963-0"
         2) 1) "repo"
            2) "1"
      2) 1) "1631010540381-0"
         2) 1) "repo"
            2) "2"
      3) 1) "1631010542237-0"
         2) 1) "repo"
            2) "3"
      4) 1) "1631010544817-0"
         2) 1) "repo"
            2) "4"
      5) 1) "1631010546467-0"
         2) 1) "repo"
            2) "5"
      6) 1) "1631010568190-0"
         2) 1) "repo"
            2) "6"
# 3            
127.0.0.1:6379> XREADGROUP group group1 consumer1 streams mqstream >
(nil)
# 4
127.0.0.1:6379> XREADGROUP group group1 consumer2 streams mqstream 0
1) 1) "mqstream"
   2) (empty array)    
# 5
127.0.0.1:6379> XGROUP create mqstream group2 0
OK
127.0.0.1:6379> XREADGROUP group group2 consumer1 count 1 streams mqstream >
1) 1) "mqstream"
   2) 1) 1) "1631010537963-0"
         2) 1) "repo"
            2) "1"
127.0.0.1:6379> XREADGROUP group group2 consumer2 count 1 streams mqstream >
1) 1) "mqstream"
   2) 1) 1) "1631010540381-0"
         2) 1) "repo"
            2) "2"
127.0.0.1:6379> XREADGROUP group group2 consumer3 count 1 streams mqstream >
1) 1) "mqstream"
   2) 1) 1) "1631010542237-0"
         2) 1) "repo"
            2) "3"

4. XREADGROUP GROUP

使用 XREADGROUP GROUP 读取消费组中的消息

语法格式：

XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]

• GROUP：固定
• group：消费组名
• consumer：消费者名
• [COUNT count]：每次获取消息的数量
• [BLOCK milliseconds]：阻塞模式和超时时间
• [NOACK]：不需要确认消息，适用于不怎么重要的可以丢失的消息
• STREAMS：固定
• key [key …]：指定的队列名
• ID [ID …]：指定的消息 ID，> 指定读取所有未消费的消息，其他值指定被挂起的消息

执行如下命令：

1. XREADGROUP group group1 consumer1 streams mqstream >：让 group1 消费组里的消费者 consumer1 从 mqstream 中读取所有消息，其中，命令最后的参数“>”，表示从第一条尚未被消费的消息开始读取。因为在 consumer1 读取消息前，group1 中没有其他消费者读取过消息，所以，consumer1 就得到 mqstream 消息队列中的所有消息了（一共 6 条）。
2. XREADGROUP group group1 consumer1 streams mqstream >：再次执行上述2命令查看返回效果
3. XREADGROUP group group1 consumer2 streams mqstream 0：需要注意的是，消息队列中的消息一旦被消费组里的一个消费者读取了，就不能再被该消费组内的其他消费者读取了。比如说，我们执行完刚才的 XREADGROUP 命令后，再执行下面的命令，让 group1 内的 consumer2 读取消息时，consumer2 读到的就是空值，因为消息已经被 consumer1 读取完了.
4. XREADGROUP group group2 consumer1 count 1 streams mqstream >：使用消费组的目的是让组内的多个消费者共同分担读取消息，所以，我们通常会让每个消费者读取部分消息，从而实现消息读取负载在多个消费者间是均衡分布的。例如，我们执行下列命令，让 group2 中的 consumer1、2、3 各自读取一条消息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

# 1
127.0.0.1:6379> XREADGROUP group group1 consumer1 streams mqstream >
1) 1) "mqstream"
   2) 1) 1) "1631010537963-0"
         2) 1) "repo"
            2) "1"
      2) 1) "1631010540381-0"
         2) 1) "repo"
            2) "2"
      3) 1) "1631010542237-0"
         2) 1) "repo"
            2) "3"
      4) 1) "1631010544817-0"
         2) 1) "repo"
            2) "4"
      5) 1) "1631010546467-0"
         2) 1) "repo"
            2) "5"
      6) 1) "1631010568190-0"
         2) 1) "repo"
            2) "6"
# 2            
127.0.0.1:6379> XREADGROUP group group1 consumer1 streams mqstream >
(nil)
# 3
127.0.0.1:6379> XREADGROUP group group1 consumer2 streams mqstream 0
1) 1) "mqstream"
   2) (empty array)    
# 4
127.0.0.1:6379> XGROUP create mqstream group2 0
OK
127.0.0.1:6379> XREADGROUP group group2 consumer1 count 1 streams mqstream >
1) 1) "mqstream"
   2) 1) 1) "1631010537963-0"
         2) 1) "repo"
            2) "1"
127.0.0.1:6379> XREADGROUP group group2 consumer2 count 1 streams mqstream >
1) 1) "mqstream"
   2) 1) 1) "1631010540381-0"
         2) 1) "repo"
            2) "2"
127.0.0.1:6379> XREADGROUP group group2 consumer3 count 1 streams mqstream >
1) 1) "mqstream"
   2) 1) 1) "1631010542237-0"
         2) 1) "repo"
            2) "3"

5. XPENDING和XACK

语法格式：

XPENDING key group [start end count] [consumer]

• key，指定的 key
• group，指定的分组
• [start end count]，起始 ID 和结束 ID 还有数量
• consumer，消费者名字

为了保证消费者在发生故障或宕机再次重启后，仍然可以读取未处理完的消息，Streams 会自动使用内部队列（也称为 PENDING List）留存消费组里每个消费者读取的消息，直到消费者使用 XACK 命令通知 Streams“消息已经处理完成”。如果消费者没有成功处理消息，它就不会给 Streams 发送 XACK 命令，消息仍然会留存。此时，消费者可以在重启后，用 XPENDING 命令查看已读取、但尚未确认处理完成的消息。

执行如下命令：

1. 查看 group2 中各个消费者已读取、但尚未确认的消息个数。其中，XPENDING 返回结果的第二、三行分别表示 group2 中所有消费者读取的消息最小 ID 和最大 ID。
2. 查看某个消费者具体读取了哪些数据
3. 通过2步骤可以看到，consumer2 已读取的消息的 ID 是 1599274912765-0。一旦消息 1599274912765-0 被 consumer2 处理了，consumer2 就可以使用 XACK 命令通知 Streams，然后这条消息就会被删除。当我们再使用 XPENDING 命令查看时，就可以看到，consumer2 已经没有已读取、但尚未确认处理的消息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

# 1
127.0.0.1:6379> XPENDING mqstream group2
1) (integer) 3
2) "1631010537963-0"
3) "1631010542237-0"
4) 1) 1) "consumer1"
      2) "1"
   2) 1) "consumer2"
      2) "1"
   3) 1) "consumer3"
      2) "1"
# 2
127.0.0.1:6379> XPENDING mqstream group2 - + 10 consumer2
1) 1) "1631010540381-0"
   2) "consumer2"
   3) (integer) 1820141
   4) (integer) 1
# 3
127.0.0.1:6379> XACK mqstream group2 1631010540381-0
(integer) 1
127.0.0.1:6379> XPENDING mqstream group2 - + 10 consumer2
(empty array)

五、总结

前面介绍了 List、Pub/Sub、Stream 在做队列的使用方式

之后又把 Redis 和专业的消息队列中间件做对比，发现 Redis 的不足之处。

最后，我们得出 Redis 做队列的合适场景。

这里我也列了一个表格，总结了它们各自的优缺点。

六、后记

最后，再聊一聊关于「技术方案选型」的问题。

我们在分析 Redis 细节时，一直在提出问题，然后寻找更好的解决方案，在文章最后，又聊到一个专业的消息队列应该怎么做。

其实，我们在讨论技术选型时，就是一个关于如何取舍的问题。

而这里我想传达给你的信息是，在面对技术选型时，不要不经过思考就觉得哪个方案好，哪个方案不好。

你需要根据具体场景具体分析，这里我把这个分析过程分为 2 个层面：

1. 业务功能角度
2. 技术资源角度

这篇文章所讲到的内容，都是以业务功能角度出发做决策的。

但这里的第二点，从技术资源角度出发，其实也很重要。

技术资源的角度是说，你所处的公司环境、技术资源能否匹配这些技术方案。

这个怎么解释呢？

简单来讲，就是你所在的公司、团队，是否有匹配的资源能 hold 住这些技术方案。

我们都知道 Kafka、RabbitMQ 是非常专业的消息中间件，但它们的部署和运维，相比于 Redis 来说，也会更复杂一些。

如果你在一个大公司，公司本身就有优秀的运维团队，那么使用这些中间件肯定没问题，因为有足够优秀的人能 hold 住这些中间件，公司也会投入人力和时间在这个方向上。

但如果你是在一个初创公司，业务正处在快速发展期，暂时没有能 hold 住这些中间件的团队和人，如果贸然使用这些组件，当发生故障时，排查问题也会变得很困难，甚至会阻碍业务的发展。

而这种情形下，如果公司的技术人员对于 Redis 都很熟，综合评估来看，Redis 也基本可以满足业务 90% 的需求，那当下选择 Redis 未必不是一个好的决策。

所以，做技术选型不只是技术问题，还与人、团队、管理、组织结构有关。

也正是因为这些原因，当你在和别人讨论技术选型问题时，你会发现每个公司的做法都不相同。

毕竟每个公司所处的环境和文化不一样，做出的决策当然就会各有差异。

如果你不了解这其中的逻辑，那在做技术选型时，只会趋于表面现象，无法深入到问题根源。

而一旦你理解了这个逻辑，那么你在看待这个问题时，不仅对于技术会有更加深刻认识，对技术资源和人的把握，也会更加清晰。

希望你以后在做技术选型时，能够把这些因素也考虑在内，这对你的技术成长之路也是非常有帮助的。

Docker 部署安装Nginx

一、

MarkDown日常使用Tips

一、MarkDown表格中使用竖线

在表格中输入｜方法：使用|替换

Docker相关基础命令

一、镜像相关命令

二、仓库相关

三、容器相关

1
2

四、卷的相关命令

五、常用命令

抽空过来补上

Docker 环境安装及问题处理

一、Dokcer安装

我们要安装一个目前最主流的容器技术的实现 Docker。假设我们的操作系统是 CentOS，你可以参考https://docs.docker.com/install/linux/docker-ce/centos/这个官方文档，进行安装。

1. 第一步，删除原有版本的 Docker。

   1 2 3 4 5 6 7 8

   sudo yum remove docker \ docker-client \ docker-client-latest \ docker-common \ docker-latest \ docker-latest-logrotate \ docker-logrotate \ docker-engine

2. 第二步，存储库安装，设置存储库

   安装yum-utils包（提供yum-config-manager 实用程序）并设置稳定存储库。

   1 2 3 4

   sudo yum install -y yum-utils sudo yum-config-manager \ --add-repo \ https://download.docker.com/linux/centos/docker-ce.repo

3. 第三步，安装 Docker 引擎。

   1	sudo yum install docker-ce docker-ce-cli containerd.io

4. 第四部，启动 Docker。

   1	sudo systemctl start docker

5. 第五步，设置Docker开机自启

   1	sudo systemctl enable docker

二、关于Docker命令对于普通用户的权限问题解决方案

安装按Docker后，使用普通用户执行docker ps命令会出现如下错误。这是由于普通用户不具有使用docker权限。

1

Got permission denied while trying to connect to the Docker daemon socket at unix:///var/run/docker.sock: Get "http://%2Fvar%2Frun%2Fdocker.sock/v1.24/containers/json": dial unix /var/run/docker.sock: connect: permission denied

解决方案：

1. 添加docker用户组

   1	groupadd docker

2. 把用户加入docker用户组

   1	gpasswd -a sse docker

3. 查看是否添加成功

   1	cat /etc/group | grep '^docker'

4. 更新用户组

   1	newgrp docker

5. 测试docker命令是否可以使用

   1

   docker ps

Harbor 私有仓库构建

一、Harbor 介绍

Docker容器应用的开发和运行离不开可靠的镜像管理，虽然Docker官方也提供了公共的镜像仓库，但是从安全和效率等方面考虑，部署私有环境内的Registry也是非常必要的。Harbor是由VMware公司开源的企业级的Docker Registry管理项目，它包括权限管理(RBAC)、LDAP、日志审核、管理界面、自我注册、镜像复制和中文支持等功能

二、环境准备

1.硬件

2.软件

3.网络端口

Harbor要求在目标主机上打开以下端口

三、下载Harbor安装程序

1.打开下载页面

https://github.com/goharbor/harbor/releases

2.选择离线安装或者在线安装

3.使用tar提取安装包

1

bash $ tar xzvf harbor-offline-installer-version.tgz

四、配置HTTPS访问Harbor

1.准备HTTPS证书

yourdomain.com.crt

yourdomain.com.key

2.向Harbor和Docker提供证书

1. 将服务器证书和密钥复制到 Harbor 主机上的 certficates 文件夹中。

   1 2	cp yourdomain.com.crt /data/cert/ cp yourdomain.com.key /data/cert/

2. 转换yourdomain.com.crt为yourdomain.com.cert，供 Docker 使用。
   Docker 守护进程将.crt文件解释为 CA 证书，将.cert文件解释为客户端证书。

   1	openssl x509 -inform PEM -in yourdomain.com.crt -out yourdomain.com.cert

   

3. 重启Docker引擎

   1	systemctl restart docker

3.配置Harbor YML文件

1. 在harbor的安装目录下找到harbor.yml.tmpl并复制一份

   1	cp harbor.yml.tmpl harbor.ymls

2. 配置主机名

   1 2 3

   # The IP address or hostname to access admin UI and registry service. # DO NOT use localhost or 127.0.0.1, because Harbor needs to be accessed by external clients. hostname: yourdomain.com

3. 配置HTTPS的证书

   1 2 3 4 5 6

   https: # https port for harbor, default is 443 port: 443 # The path of cert and key files for nginx certificate: /var/lib/sse/keys/yourdomain.com.crt private_key: /var/lib/sse/keys/yourdomain.com.key

4. 配置镜像数据存放地址

   1 2	# The default data volume data_volume: /var/lib/sse/data

5. 指定管理员密码

   1 2 3 4

   # The initial password of Harbor admin # It only works in first time to install harbor # Remember Change the admin password from UI after launching Harbor. harbor_admin_password: Harbor12345

4.部署或重新配置Harbor

1. 运行prepare脚本以启用 HTTPS。

   Harbor 使用一个nginx实例作为所有服务的反向代理。您可以使用prepare脚本进行配置nginx以使用 HTTPS。该prepare在Harbor的安装包，在同级别的install.sh脚本。

   1 2	./prepare ./install.sh

2. 如果 Harbor 正在运行，请停止并删除现有实例。

   您的镜像数据保留在文件系统中，因此不会丢失任何数据。

   1	docker-compose down -v

3. 重启 Harbor

   1	docker-compose up -d

五、测试访问

1

https://yourdomain.com

Linux下迁移Mysql数据目录

最初架构设计，mysql的数据目录被放在系统盘里面，随着数据的不断增长，系统盘必须随着数据不断扩容，才能满足服务的正常运行。当系统盘扩容到1T的时候出现了瓶颈，系统盘最高支持扩容到1T。这使得不能不面临mysql数据目录迁移到数据盘问题。

一.迁移时需要考虑的事项

1. 尽可能少的迁移改动，防止在数据迁移时出错
2. 需要考虑到目前服务还在使用，所以迁移应该尽可能快的完成。
3. 迁移尽量不影响多源复制，如果复制中断，修复起来比较麻烦，而且还影响微信端的正常使用。
4. 系统盘不能缩容，需要考虑剩余的系统盘空间怎么办

二.迁移步骤

1. 在中央业务库中暂停迁移学校的多源复制

   1	stop slave for channel "school47";

2. 暂停学校服务

   1	pm2 stop 1

3. 停止mysql服务

   1	systemctl stop mysqld.service

4. 复制原数据目录的数据到将要迁移的数据目录

   1 2 3 4

   # 1.进入挂载的数据盘目录 shell> cd /mnt/sse/ # 2.将原数据目录下的文件复制到将要迁移的目录，注意这里cp要加 -a，需要保留原文件属性 shell> cp -a -r /var/lib/mysql/ ./

5. 修改/etc/my.cnf文件中的配置。

   1 2 3 4 5 6 7

   [mysqld] skip-log-bin datadir=/mnt/sse/mysql socket=/mnt/sse/mysql/mysql.sock [client] # 如果不设置这句的话，登录数据库的时候会报：ERROR 2002 (HY000): Can't connect to local MySQL server through socket '/var/lib/mysql/mysql.sock' (2) socket=/mnt/sse/mysql/mysql.sock

6. 开启mysql

   1 2	# 这里一定需要使用start,不能用restart,否则会出现问题 systemctl start mysqld