[PostgreSQL启动失败排查]的搜索结果

...程中出现问题导致无法启动时，k8s会不断重启，直到可用状态为止 故障转移：当正在运行中pod所在的节点发生故障或者宕机时，k8s会选择集群中另一个可用节点，将pod运行到可用节点上； pod数量的扩缩容：pod副本的扩容和缩容 镜像升降级：支持镜像版本的升级和降级； 配置模板 rs的所有配置如下 apiVersion: apps/v1 版本号kind: ReplicaSet 类型 metadata: 元数据name: rs名称 namespace: 所属命名空间 labels: 标签controller: rsspec: 详情描述replicas: 3 副本数量selector: 选择器，通过它指定该控制器管理哪些podmatchLabels: Labels匹配规则app: nginx-podmatchExpressions: Expressions匹配规则，key就是label的key，values的值是个数组，意思是标签值必须是此数组中的其中一个才能匹配上；- {key: app, operator: In, values: [nginx-pod]}template: 模板，当副本数量不足时，会根据下面的模板创建pod副本metadata:labels: 这里的标签必须和上面的matchLabels一致，将他们关联起来app: nginx-podspec:containers:- name: nginximage: nginx:1.17.1ports:- containerPort: 80 1、创建一个ReplicaSet 新建一个文件 rs.yaml，内容如下 apiVersion: apps/v1kind: ReplicaSet pod控制器metadata: 元数据name: pc-replicaset 名字namespace: dev 名称空间spec:replicas: 3 副本数selector: 选择器，通过它指定该控制器管理哪些podmatchLabels: Labels匹配规则app: nginx-podtemplate: 模板，当副本数量不足时，会根据下面的模板创建pod副本metadata:labels:app: nginx-podspec:containers:- name: nginximage: nginx:1.17.1 运行 kubectl create -f rs.yaml 获取replicaset kubectl get replicaset -n dev 2、扩缩容 刚刚我们已经用第一种方式创建了一个replicaSet，现在就基于原来的rs进行扩容，原来的副本数量是3个，现在我们将其扩到6个，做法也很简单，运行编辑命令 第一种方式: scale 使用scale命令实现扩缩容，后面--replicas=n直接指定目标数量即可kubectl scale rs pc-replicaset --replicas=2 -n dev 第二种方式：使用edit命令编辑rs 这种方式相当于使用vi编辑修改yaml配置的内容，进去后将replicas的值改为1，保存后自动生效kubectl edit rs pc-replicaset -n dev 3、镜像版本变更 第一种方式：scale kubectl scale rs pc-replicaset nginx=nginx:1.71.2 -n dev 第二种方式：edit 这种方式相当于使用vi编辑修改yaml配置的内容，进去后将nginx的值改为nginx:1.71.2，保存后自动生效kubectl edit rs pc-replicaset -n dev 4、删除rs 第一种方式kubectl delete -f rs.yaml 第二种方式 ,如果想要只删rs，但不删除pod，可在删除时加上--cascade=false参数（不推荐）kubectl delete rs pc-replicaset -n dev --cascade=false 2、Deployment k8s v1.2版本后加入Deployment；这种控制器不直接控制pod，而是通过管理ReplicaSet来间接管理pod；也就是Deployment管理ReplicaSet，ReplicaSet管理pod；所以 Deployment 比 ReplicaSet 功能更加强大 当我们创建了一个Deployment之后，也会自动创建一个ReplicaSet 功能 支持ReplicaSet 的所有功能 支持发布的停止、继续 支持版本的滚动更新和回退功能 配置模板 新建文件 apiVersion: apps/v1 版本号kind: Deployment 类型 metadata: 元数据name: rs名称 namespace: 所属命名空间 labels: 标签controller: deployspec: 详情描述replicas: 3 副本数量revisionHistoryLimit: 3 保留历史版本的数量，默认10，内部通过保留rs来实现paused: false 暂停部署，默认是falseprogressDeadlineSeconds: 600 部署超时时间（s），默认是600strategy: 策略type: RollingUpdate 滚动更新策略rollingUpdate: 滚动更新maxSurge: 30% 最大额外可以存在的副本数，可以为百分比，也可以为整数maxUnavailable: 30% 最大不可用状态的 Pod 的最大值，可以为百分比，也可以为整数selector: 选择器，通过它指定该控制器管理哪些podmatchLabels: Labels匹配规则app: nginx-podmatchExpressions: Expressions匹配规则- {key: app, operator: In, values: [nginx-pod]}template: 模板，当副本数量不足时，会根据下面的模板创建pod副本metadata:labels:app: nginx-podspec:containers:- name: nginximage: nginx:1.17.1ports:- containerPort: 80 1、创建和删除Deployment 创建pc-deployment.yaml，内容如下： apiVersion: apps/v1kind: Deployment metadata:name: pc-deploymentnamespace: devspec: replicas: 3selector:matchLabels:app: nginx-podtemplate:metadata:labels:app: nginx-podspec:containers:- name: nginximage: nginx:1.17.1 创建和查看 创建deployment，--record=true 表示记录整个deployment更新过程[root@k8s-master01 ~] kubectl create -f pc-deployment.yaml --record=truedeployment.apps/pc-deployment created 查看deployment READY 可用的/总数 UP-TO-DATE 最新版本的pod的数量 AVAILABLE 当前可用的pod的数量[root@k8s-master01 ~] kubectl get deploy pc-deployment -n devNAME READY UP-TO-DATE AVAILABLE AGEpc-deployment 3/3 3 3 15s 查看rs 发现rs的名称是在原来deployment的名字后面添加了一个10位数的随机串[root@k8s-master01 ~] kubectl get rs -n devNAME DESIRED CURRENT READY AGEpc-deployment-6696798b78 3 3 3 23s 查看pod[root@k8s-master01 ~] kubectl get pods -n devNAME READY STATUS RESTARTS AGEpc-deployment-6696798b78-d2c8n 1/1 Running 0 107spc-deployment-6696798b78-smpvp 1/1 Running 0 107spc-deployment-6696798b78-wvjd8 1/1 Running 0 107s 删除deployment 删除deployment，其下的rs和pod也将被删除kubectl delete -f pc-deployment.yaml 2、扩缩容 deployment的扩缩容和 ReplicaSet 的扩缩容一样，只需要将rs或者replicaSet改为deployment即可，具体请参考上面的 ReplicaSet 扩缩容 3、镜像更新 刚刚在创建时加上了--record=true参数，所以在一旦进行了镜像更新，就会新建出一个pod出来，将老的old-pod上的容器全删除，然后在新的new-pod上在新建对应数量的容器，此时old-pod是不会删除的，因为这个old-pod是要进行回退的； 镜像更新策略有2种 滚动更新（RollingUpdate）：（默认值），杀死一部分，就启动一部分，在更新过程中，存在两个版本Pod 重建更新（Recreate）：在创建出新的Pod之前会先杀掉所有已存在的Pod strategy：指定新的Pod替换旧的Pod的策略， 支持两个属性：type：指定策略类型，支持两种策略Recreate：在创建出新的Pod之前会先杀掉所有已存在的PodRollingUpdate：滚动更新，就是杀死一部分，就启动一部分，在更新过程中，存在两个版本PodrollingUpdate：当type为RollingUpdate时生效，用于为RollingUpdate设置参数，支持两个属性：maxUnavailable：用来指定在升级过程中不可用Pod的最大数量，默认为25%。maxSurge： 用来指定在升级过程中可以超过期望的Pod的最大数量，默认为25%。 重建更新 编辑pc-deployment.yaml,在spec节点下添加更新策略 spec:strategy: 策略type: Recreate 重建更新 创建deploy进行验证 变更镜像[root@k8s-master01 ~] kubectl set image deployment pc-deployment nginx=nginx:1.17.2 -n devdeployment.apps/pc-deployment image updated 观察升级过程[root@k8s-master01 ~] kubectl get pods -n dev -wNAME READY STATUS RESTARTS AGEpc-deployment-5d89bdfbf9-65qcw 1/1 Running 0 31spc-deployment-5d89bdfbf9-w5nzv 1/1 Running 0 31spc-deployment-5d89bdfbf9-xpt7w 1/1 Running 0 31spc-deployment-5d89bdfbf9-xpt7w 1/1 Terminating 0 41spc-deployment-5d89bdfbf9-65qcw 1/1 Terminating 0 41spc-deployment-5d89bdfbf9-w5nzv 1/1 Terminating 0 41spc-deployment-675d469f8b-grn8z 0/1 Pending 0 0spc-deployment-675d469f8b-hbl4v 0/1 Pending 0 0spc-deployment-675d469f8b-67nz2 0/1 Pending 0 0spc-deployment-675d469f8b-grn8z 0/1 ContainerCreating 0 0spc-deployment-675d469f8b-hbl4v 0/1 ContainerCreating 0 0spc-deployment-675d469f8b-67nz2 0/1 ContainerCreating 0 0spc-deployment-675d469f8b-grn8z 1/1 Running 0 1spc-deployment-675d469f8b-67nz2 1/1 Running 0 1spc-deployment-675d469f8b-hbl4v 1/1 Running 0 2s 滚动更新 编辑pc-deployment.yaml,在spec节点下添加更新策略 spec:strategy: 策略type: RollingUpdate 滚动更新策略rollingUpdate:maxSurge: 25% maxUnavailable: 25% 创建deploy进行验证 变更镜像[root@k8s-master01 ~] kubectl set image deployment pc-deployment nginx=nginx:1.17.3 -n dev deployment.apps/pc-deployment image updated 观察升级过程[root@k8s-master01 ~] kubectl get pods -n dev -wNAME READY STATUS RESTARTS AGEpc-deployment-c848d767-8rbzt 1/1 Running 0 31mpc-deployment-c848d767-h4p68 1/1 Running 0 31mpc-deployment-c848d767-hlmz4 1/1 Running 0 31mpc-deployment-c848d767-rrqcn 1/1 Running 0 31mpc-deployment-966bf7f44-226rx 0/1 Pending 0 0spc-deployment-966bf7f44-226rx 0/1 ContainerCreating 0 0spc-deployment-966bf7f44-226rx 1/1 Running 0 1spc-deployment-c848d767-h4p68 0/1 Terminating 0 34mpc-deployment-966bf7f44-cnd44 0/1 Pending 0 0spc-deployment-966bf7f44-cnd44 0/1 ContainerCreating 0 0spc-deployment-966bf7f44-cnd44 1/1 Running 0 2spc-deployment-c848d767-hlmz4 0/1 Terminating 0 34mpc-deployment-966bf7f44-px48p 0/1 Pending 0 0spc-deployment-966bf7f44-px48p 0/1 ContainerCreating 0 0spc-deployment-966bf7f44-px48p 1/1 Running 0 0spc-deployment-c848d767-8rbzt 0/1 Terminating 0 34mpc-deployment-966bf7f44-dkmqp 0/1 Pending 0 0spc-deployment-966bf7f44-dkmqp 0/1 ContainerCreating 0 0spc-deployment-966bf7f44-dkmqp 1/1 Running 0 2spc-deployment-c848d767-rrqcn 0/1 Terminating 0 34m 至此，新版本的pod创建完毕，就版本的pod销毁完毕 中间过程是滚动进行的，也就是边销毁边创建 4、版本回退 更新 刚刚在创建时加上了--record=true参数，所以在一旦进行了镜像更新，就会新建出一个pod出来，将老的old-pod上的容器全删除，然后在新的new-pod上在新建对应数量的容器，此时old-pod是不会删除的，因为这个old-pod是要进行回退的； 回退 在回退时会将new-pod上的容器全部删除，在将old-pod上恢复原来的容器； 回退命令 kubectl rollout： 版本升级相关功能，支持下面的选项： status 显示当前升级状态 history 显示 升级历史记录 pause 暂停版本升级过程 resume 继续已经暂停的版本升级过程 restart 重启版本升级过程 undo 回滚到上一级版本（可以使用–to-revision回滚到指定版本） 用法 查看当前升级版本的状态kubectl rollout status deploy pc-deployment -n dev 查看升级历史记录kubectl rollout history deploy pc-deployment -n dev 版本回滚 这里直接使用--to-revision=1回滚到了1版本， 如果省略这个选项，就是回退到上个版本kubectl rollout undo deployment pc-deployment --to-revision=1 -n dev 金丝雀发布 Deployment控制器支持控制更新过程中的控制，如“暂停(pause)”或“继续(resume)”更新操作。 比如有一批新的Pod资源创建完成后立即暂停更新过程，此时，仅存在一部分新版本的应用，主体部分还是旧的版本。然后，再筛选一小部分的用户请求路由到新版本的Pod应用，继续观察能否稳定地按期望的方式运行。确定没问题之后再继续完成余下的Pod资源滚动更新，否则立即回滚更新操作。这就是所谓的金丝雀发布。 金丝雀发布不是自动完成的，需要人为手动去操作，才能达到金丝雀发布的标准； 更新deployment的版本，并配置暂停deploymentkubectl set image deploy pc-deployment nginx=nginx:1.17.4 -n dev && kubectl rollout pause deployment pc-deployment -n dev 观察更新状态kubectl rollout status deploy pc-deployment -n dev　 监控更新的过程kubectl get rs -n dev -o wide 确保更新的pod没问题了，继续更新kubectl rollout resume deploy pc-deployment -n dev 如果有问题，就回退到上个版本回退到上个版本kubectl rollout undo deployment pc-deployment -n dev Horizontal Pod Autoscaler 简称HPA，使用deployment可以手动调整pod的数量来实现扩容和缩容；但是这显然不符合k8s的自动化的定位，k8s期望可以通过检测pod的使用情况，实现pod数量自动调整，于是就有了HPA控制器； HPA可以获取每个Pod利用率，然后和HPA中定义的指标进行对比，同时计算出需要伸缩的具体值，最后实现Pod的数量的调整。比如说我指定了一个规则：当我的cpu利用率达到90%或者内存使用率到达80%的时候，就需要进行调整pod的副本数量，每次添加n个pod副本； 其实HPA与之前的Deployment一样，也属于一种Kubernetes资源对象，它通过追踪分析ReplicaSet控制器的所有目标Pod的负载变化情况，来确定是否需要针对性地调整目标Pod的副本数，也就是HPA管理Deployment，Deployment管理ReplicaSet，ReplicaSet管理pod，这是HPA的实现原理。 1、安装metrics-server metrics-server可以用来收集集群中的资源使用情况 安装git[root@k8s-master01 ~] yum install git -y 获取metrics-server, 注意使用的版本[root@k8s-master01 ~] git clone -b v0.3.6 https://github.com/kubernetes-incubator/metrics-server 修改deployment, 注意修改的是镜像和初始化参数[root@k8s-master01 ~] cd /root/metrics-server/deploy/1.8+/[root@k8s-master01 1.8+] vim metrics-server-deployment.yaml按图中添加下面选项hostNetwork: trueimage: registry.cn-hangzhou.aliyuncs.com/google_containers/metrics-server-amd64:v0.3.6args:- --kubelet-insecure-tls- --kubelet-preferred-address-types=InternalIP,Hostname,InternalDNS,ExternalDNS,ExternalIP 2、安装metrics-server [root@k8s-master01 1.8+] kubectl apply -f ./ 3、查看pod运行情况 [root@k8s-master01 1.8+] kubectl get pod -n kube-systemmetrics-server-6b976979db-2xwbj 1/1 Running 0 90s 4、使用kubectl top node 查看资源使用情况 [root@k8s-master01 1.8+] kubectl top nodeNAME CPU(cores) CPU% MEMORY(bytes) MEMORY%k8s-master01 289m 14% 1582Mi 54% k8s-node01 81m 4% 1195Mi 40% k8s-node02 72m 3% 1211Mi 41% [root@k8s-master01 1.8+] kubectl top pod -n kube-systemNAME CPU(cores) MEMORY(bytes)coredns-6955765f44-7ptsb 3m 9Micoredns-6955765f44-vcwr5 3m 8Mietcd-master 14m 145Mi... 至此,metrics-server安装完成 5、 准备deployment和servie 创建pc-hpa-pod.yaml文件，内容如下： apiVersion: apps/v1kind: Deploymentmetadata:name: nginxnamespace: devspec:strategy: 策略type: RollingUpdate 滚动更新策略replicas: 1selector:matchLabels:app: nginx-podtemplate:metadata:labels:app: nginx-podspec:containers:- name: nginximage: nginx:1.17.1resources: 资源配额limits: 限制资源（上限）cpu: "1" CPU限制，单位是core数requests: 请求资源（下限）cpu: "100m" CPU限制，单位是core数 创建deployment [root@k8s-master01 1.8+] kubectl run nginx --image=nginx:1.17.1 --requests=cpu=100m -n dev 6、创建service [root@k8s-master01 1.8+] kubectl expose deployment nginx --type=NodePort --port=80 -n dev 7、查看 [root@k8s-master01 1.8+] kubectl get deployment,pod,svc -n devNAME READY UP-TO-DATE AVAILABLE AGEdeployment.apps/nginx 1/1 1 1 47sNAME READY STATUS RESTARTS AGEpod/nginx-7df9756ccc-bh8dr 1/1 Running 0 47sNAME TYPE CLUSTER-IP EXTERNAL-IP PORT(S) AGEservice/nginx NodePort 10.101.18.29 <none> 80:31830/TCP 35s 8、 部署HPA 创建pc-hpa.yaml文件，内容如下： apiVersion: autoscaling/v1kind: HorizontalPodAutoscalermetadata:name: pc-hpanamespace: devspec:minReplicas: 1 最小pod数量maxReplicas: 10 最大pod数量 ，pod数量会在1~10之间自动伸缩targetCPUUtilizationPercentage: 3 CPU使用率指标，如果cpu使用率达到3%就会进行扩容；为了测试方便，将这个数值调小一些scaleTargetRef: 指定要控制的nginx信息apiVersion: /v1kind: Deploymentname: nginx 创建hpa [root@k8s-master01 1.8+] kubectl create -f pc-hpa.yamlhorizontalpodautoscaler.autoscaling/pc-hpa created 查看hpa [root@k8s-master01 1.8+] kubectl get hpa -n devNAME REFERENCE TARGETS MINPODS MAXPODS REPLICAS AGEpc-hpa Deployment/nginx 0%/3% 1 10 1 62s 9、 测试 使用压测工具对service地址192.168.5.4:31830进行压测，然后通过控制台查看hpa和pod的变化 hpa变化 [root@k8s-master01 ~] kubectl get hpa -n dev -wNAME REFERENCE TARGETS MINPODS MAXPODS REPLICAS AGEpc-hpa Deployment/nginx 0%/3% 1 10 1 4m11spc-hpa Deployment/nginx 0%/3% 1 10 1 5m19spc-hpa Deployment/nginx 22%/3% 1 10 1 6m50spc-hpa Deployment/nginx 22%/3% 1 10 4 7m5spc-hpa Deployment/nginx 22%/3% 1 10 8 7m21spc-hpa Deployment/nginx 6%/3% 1 10 8 7m51spc-hpa Deployment/nginx 0%/3% 1 10 8 9m6spc-hpa Deployment/nginx 0%/3% 1 10 8 13mpc-hpa Deployment/nginx 0%/3% 1 10 1 14m deployment变化 [root@k8s-master01 ~] kubectl get deployment -n dev -wNAME READY UP-TO-DATE AVAILABLE AGEnginx 1/1 1 1 11mnginx 1/4 1 1 13mnginx 1/4 1 1 13mnginx 1/4 1 1 13mnginx 1/4 4 1 13mnginx 1/8 4 1 14mnginx 1/8 4 1 14mnginx 1/8 4 1 14mnginx 1/8 8 1 14mnginx 2/8 8 2 14mnginx 3/8 8 3 14mnginx 4/8 8 4 14mnginx 5/8 8 5 14mnginx 6/8 8 6 14mnginx 7/8 8 7 14mnginx 8/8 8 8 15mnginx 8/1 8 8 20mnginx 8/1 8 8 20mnginx 1/1 1 1 20m pod变化 [root@k8s-master01 ~] kubectl get pods -n dev -wNAME READY STATUS RESTARTS AGEnginx-7df9756ccc-bh8dr 1/1 Running 0 11mnginx-7df9756ccc-cpgrv 0/1 Pending 0 0snginx-7df9756ccc-8zhwk 0/1 Pending 0 0snginx-7df9756ccc-rr9bn 0/1 Pending 0 0snginx-7df9756ccc-cpgrv 0/1 ContainerCreating 0 0snginx-7df9756ccc-8zhwk 0/1 ContainerCreating 0 0snginx-7df9756ccc-rr9bn 0/1 ContainerCreating 0 0snginx-7df9756ccc-m9gsj 0/1 Pending 0 0snginx-7df9756ccc-g56qb 0/1 Pending 0 0snginx-7df9756ccc-sl9c6 0/1 Pending 0 0snginx-7df9756ccc-fgst7 0/1 Pending 0 0snginx-7df9756ccc-g56qb 0/1 ContainerCreating 0 0snginx-7df9756ccc-m9gsj 0/1 ContainerCreating 0 0snginx-7df9756ccc-sl9c6 0/1 ContainerCreating 0 0snginx-7df9756ccc-fgst7 0/1 ContainerCreating 0 0snginx-7df9756ccc-8zhwk 1/1 Running 0 19snginx-7df9756ccc-rr9bn 1/1 Running 0 30snginx-7df9756ccc-m9gsj 1/1 Running 0 21snginx-7df9756ccc-cpgrv 1/1 Running 0 47snginx-7df9756ccc-sl9c6 1/1 Running 0 33snginx-7df9756ccc-g56qb 1/1 Running 0 48snginx-7df9756ccc-fgst7 1/1 Running 0 66snginx-7df9756ccc-fgst7 1/1 Terminating 0 6m50snginx-7df9756ccc-8zhwk 1/1 Terminating 0 7m5snginx-7df9756ccc-cpgrv 1/1 Terminating 0 7m5snginx-7df9756ccc-g56qb 1/1 Terminating 0 6m50snginx-7df9756ccc-rr9bn 1/1 Terminating 0 7m5snginx-7df9756ccc-m9gsj 1/1 Terminating 0 6m50snginx-7df9756ccc-sl9c6 1/1 Terminating 0 6m50s DaemonSet 简称DS，ds可以保证在集群中的每一台节点（或指定节点）上都运行一个副本，一般适用于日志收集、节点监控等场景；也就是说，如果一个Pod提供的功能是节点级别的（每个节点都需要且只需要一个），那么这类Pod就适合使用DaemonSet类型的控制器创建。 DaemonSet控制器的特点： 每当向集群中添加一个节点时，指定的 Pod 副本也将添加到该节点上 当节点从集群中移除时，Pod 也就被垃圾回收了 配置模板 apiVersion: apps/v1 版本号kind: DaemonSet 类型 metadata: 元数据name: rs名称 namespace: 所属命名空间 labels: 标签controller: daemonsetspec: 详情描述revisionHistoryLimit: 3 保留历史版本updateStrategy: 更新策略type: RollingUpdate 滚动更新策略rollingUpdate: 滚动更新maxUnavailable: 1 最大不可用状态的 Pod 的最大值，可以为百分比，也可以为整数selector: 选择器，通过它指定该控制器管理哪些podmatchLabels: Labels匹配规则app: nginx-podmatchExpressions: Expressions匹配规则- {key: app, operator: In, values: [nginx-pod]}template: 模板，当副本数量不足时，会根据下面的模板创建pod副本metadata:labels:app: nginx-podspec:containers:- name: nginximage: nginx:1.17.1ports:- containerPort: 80 1、创建ds 创建pc-daemonset.yaml，内容如下： apiVersion: apps/v1kind: DaemonSet metadata:name: pc-daemonsetnamespace: devspec: selector:matchLabels:app: nginx-podtemplate:metadata:labels:app: nginx-podspec:containers:- name: nginximage: nginx:1.17.1 运行 创建daemonset[root@k8s-master01 ~] kubectl create -f pc-daemonset.yamldaemonset.apps/pc-daemonset created 查看daemonset[root@k8s-master01 ~] kubectl get ds -n dev -o wideNAME DESIRED CURRENT READY UP-TO-DATE AVAILABLE AGE CONTAINERS IMAGES pc-daemonset 2 2 2 2 2 24s nginx nginx:1.17.1 查看pod,发现在每个Node上都运行一个pod[root@k8s-master01 ~] kubectl get pods -n dev -o wideNAME READY STATUS RESTARTS AGE IP NODE pc-daemonset-9bck8 1/1 Running 0 37s 10.244.1.43 node1 pc-daemonset-k224w 1/1 Running 0 37s 10.244.2.74 node2 2、删除daemonset [root@k8s-master01 ~] kubectl delete -f pc-daemonset.yamldaemonset.apps "pc-daemonset" deleted Job 主要用于负责批量处理一次性(每个任务仅运行一次就结束)任务。当然，你也可以运行多次，配置好即可，Job特点如下： 当Job创建的pod执行成功结束时，Job将记录成功结束的pod数量 当成功结束的pod达到指定的数量时，Job将完成执行 配置模板 apiVersion: batch/v1 版本号kind: Job 类型 metadata: 元数据name: rs名称 namespace: 所属命名空间 labels: 标签controller: jobspec: 详情描述completions: 1 指定job需要成功运行Pods的次数。默认值: 1parallelism: 1 指定job在任一时刻应该并发运行Pods的数量。默认值: 1activeDeadlineSeconds: 30 指定job可运行的时间期限，超过时间还未结束，系统将会尝试进行终止。backoffLimit: 6 指定job失败后进行重试的次数。默认是6manualSelector: true 是否可以使用selector选择器选择pod，默认是falseselector: 选择器，通过它指定该控制器管理哪些podmatchLabels: Labels匹配规则app: counter-podmatchExpressions: Expressions匹配规则- {key: app, operator: In, values: [counter-pod]}template: 模板，当副本数量不足时，会根据下面的模板创建pod副本metadata:labels:app: counter-podspec:restartPolicy: Never 重启策略只能设置为Never或者OnFailurecontainers:- name: counterimage: busybox:1.30command: ["bin/sh","-c","for i in 9 8 7 6 5 4 3 2 1; do echo $i;sleep 2;done"] 关于重启策略设置的说明：（这里只能设置为Never或者OnFailure） 如果指定为OnFailure，则job会在pod出现故障时重启容器，而不是创建pod，failed次数不变 如果指定为Never，则job会在pod出现故障时创建新的pod，并且故障pod不会消失，也不会重启，failed次数加1 如果指定为Always的话，就意味着一直重启，意味着job任务会重复去执行了，当然不对，所以不能设置为Always 1、创建一个job 创建pc-job.yaml，内容如下： apiVersion: batch/v1kind: Job metadata:name: pc-jobnamespace: devspec:manualSelector: trueselector:matchLabels:app: counter-podtemplate:metadata:labels:app: counter-podspec:restartPolicy: Nevercontainers:- name: counterimage: busybox:1.30command: ["bin/sh","-c","for i in 9 8 7 6 5 4 3 2 1; do echo $i;sleep 3;done"] 创建 创建job[root@k8s-master01 ~] kubectl create -f pc-job.yamljob.batch/pc-job created 查看job[root@k8s-master01 ~] kubectl get job -n dev -o wide -wNAME COMPLETIONS DURATION AGE CONTAINERS IMAGES SELECTORpc-job 0/1 21s 21s counter busybox:1.30 app=counter-podpc-job 1/1 31s 79s counter busybox:1.30 app=counter-pod 通过观察pod状态可以看到，pod在运行完毕任务后，就会变成Completed状态[root@k8s-master01 ~] kubectl get pods -n dev -wNAME READY STATUS RESTARTS AGEpc-job-rxg96 1/1 Running 0 29spc-job-rxg96 0/1 Completed 0 33s 接下来，调整下pod运行的总数量和并行数量 即：在spec下设置下面两个选项 completions: 6 指定job需要成功运行Pods的次数为6 parallelism: 3 指定job并发运行Pods的数量为3 然后重新运行job，观察效果，此时会发现，job会每次运行3个pod，总共执行了6个pod[root@k8s-master01 ~] kubectl get pods -n dev -wNAME READY STATUS RESTARTS AGEpc-job-684ft 1/1 Running 0 5spc-job-jhj49 1/1 Running 0 5spc-job-pfcvh 1/1 Running 0 5spc-job-684ft 0/1 Completed 0 11spc-job-v7rhr 0/1 Pending 0 0spc-job-v7rhr 0/1 Pending 0 0spc-job-v7rhr 0/1 ContainerCreating 0 0spc-job-jhj49 0/1 Completed 0 11spc-job-fhwf7 0/1 Pending 0 0spc-job-fhwf7 0/1 Pending 0 0spc-job-pfcvh 0/1 Completed 0 11spc-job-5vg2j 0/1 Pending 0 0spc-job-fhwf7 0/1 ContainerCreating 0 0spc-job-5vg2j 0/1 Pending 0 0spc-job-5vg2j 0/1 ContainerCreating 0 0spc-job-fhwf7 1/1 Running 0 2spc-job-v7rhr 1/1 Running 0 2spc-job-5vg2j 1/1 Running 0 3spc-job-fhwf7 0/1 Completed 0 12spc-job-v7rhr 0/1 Completed 0 12spc-job-5vg2j 0/1 Completed 0 12s 2、删除 删除jobkubectl delete -f pc-job.yaml CronJob 简称为CJ，CronJob控制器以 Job控制器资源为其管控对象，并借助它管理pod资源对象，Job控制器定义的作业任务在其控制器资源创建之后便会立即执行，但CronJob可以以类似于Linux操作系统的周期性任务作业计划的方式控制其运行时间点及重复运行的方式。也就是说，CronJob可以在特定的时间点(反复的)去运行job任务。可以理解为定时任务 配置模板 apiVersion: batch/v1beta1 版本号kind: CronJob 类型 metadata: 元数据name: rs名称 namespace: 所属命名空间 labels: 标签controller: cronjobspec: 详情描述schedule: cron格式的作业调度运行时间点,用于控制任务在什么时间执行concurrencyPolicy: 并发执行策略，用于定义前一次作业运行尚未完成时是否以及如何运行后一次的作业failedJobHistoryLimit: 为失败的任务执行保留的历史记录数，默认为1successfulJobHistoryLimit: 为成功的任务执行保留的历史记录数，默认为3startingDeadlineSeconds: 启动作业错误的超时时长jobTemplate: job控制器模板，用于为cronjob控制器生成job对象;下面其实就是job的定义metadata:spec:completions: 1parallelism: 1activeDeadlineSeconds: 30backoffLimit: 6manualSelector: trueselector:matchLabels:app: counter-podmatchExpressions: 规则- {key: app, operator: In, values: [counter-pod]}template:metadata:labels:app: counter-podspec:restartPolicy: Never containers:- name: counterimage: busybox:1.30command: ["bin/sh","-c","for i in 9 8 7 6 5 4 3 2 1; do echo $i;sleep 20;done"] cron表达式写法 需要重点解释的几个选项：schedule: cron表达式，用于指定任务的执行时间/1 <分钟> <小时> <日> <月份> <星期>分钟 值从 0 到 59.小时 值从 0 到 23.日 值从 1 到 31.月 值从 1 到 12.星期 值从 0 到 6, 0 代表星期日多个时间可以用逗号隔开； 范围可以用连字符给出；可以作为通配符； /表示每... 例如1 // 每个小时的第一分钟执行/1 // 每分钟都执行concurrencyPolicy:Allow: 允许Jobs并发运行(默认)Forbid: 禁止并发运行，如果上一次运行尚未完成，则跳过下一次运行Replace: 替换，取消当前正在运行的作业并用新作业替换它 1、创建cronJob 创建pc-cronjob.yaml，内容如下： apiVersion: batch/v1beta1kind: CronJobmetadata:name: pc-cronjobnamespace: devlabels:controller: cronjobspec:schedule: "/1 " 每分钟执行一次jobTemplate:metadata:spec:template:spec:restartPolicy: Nevercontainers:- name: counterimage: busybox:1.30command: ["bin/sh","-c","for i in 9 8 7 6 5 4 3 2 1; do echo $i;sleep 3;done"] 运行 创建cronjob[root@k8s-master01 ~] kubectl create -f pc-cronjob.yamlcronjob.batch/pc-cronjob created 查看cronjob[root@k8s-master01 ~] kubectl get cronjobs -n devNAME SCHEDULE SUSPEND ACTIVE LAST SCHEDULE AGEpc-cronjob /1 False 0 <none> 6s 查看job[root@k8s-master01 ~] kubectl get jobs -n devNAME COMPLETIONS DURATION AGEpc-cronjob-1592587800 1/1 28s 3m26spc-cronjob-1592587860 1/1 28s 2m26spc-cronjob-1592587920 1/1 28s 86s 查看pod[root@k8s-master01 ~] kubectl get pods -n devpc-cronjob-1592587800-x4tsm 0/1 Completed 0 2m24spc-cronjob-1592587860-r5gv4 0/1 Completed 0 84spc-cronjob-1592587920-9dxxq 1/1 Running 0 24s 2、删除cronjob kubectl delete -f pc-cronjob.yaml pod调度 什么是调度 默认情况下，一个pod在哪个node节点上运行，是通过scheduler组件采用相应的算法计算出来的，这个过程是不受人工控制的； 调度规则 但是在实际使用中，我们想控制某些pod定向到达某个节点上，应该怎么做呢？其实k8s提供了四类调度规则 调度方式 描述 自动调度 通过scheduler组件采用相应的算法计算得出运行在哪个节点上 定向调度 运行到指定的node节点上，通过NodeName、NodeSelector实现 亲和性调度 跟谁关系好就调度到哪个节点上 1、nodeAffinity ：节点亲和性，调度到关系好的节点上 2、podAffinity：pod亲和性，调度到关系好的pod所在的节点上 3、PodAntAffinity：pod反清河行，调度到关系差的那个pod所在的节点上 污点（容忍）调度 污点是站在node的角度上的，比如果nodeA有一个污点，大家都别来，此时nodeA会拒绝master调度过来的pod 定向调度 指的是利用在pod上声明nodeName或nodeSelector的方式将pod调度到指定的pod节点上，因为这种定向调度是强制性的，所以如果node节点不存在的话，也会向上面进行调度，只不过pod会运行失败； 1、定向调度-> nodeName nodeName 是将pod强制调度到指定名称的node节点上，这种方式跳过了scheduler的调度逻辑，直接将pod调度到指定名称的节点上，配置文件内容如下 apiVersion: v1 版本号kind: Pod 资源类型metadata: name: pod-namenamespace: devspec: containers: - image: nginx:1.17.1name: nginx-containernodeName: node1 调度到node1节点上 2、定向调度 -> NodeSelector NodeSelector是将pod调度到添加了指定label标签的node节点上，它是通过k8s的label-selector机制实现的，也就是说，在创建pod之前，会由scheduler用matchNodeSelecto调度策略进行label标签的匹配，找出目标node，然后在将pod调度到目标node； 要实验NodeSelector，首先得给node节点加上label标签 kubectl label nodes node1 nodetag=node1 配置文件内容如下 apiVersion: v1 版本号kind: Pod 资源类型metadata: name: pod-namenamespace: devspec: containers: - image: nginx:1.17.1name: nginx-containernodeSelector: nodetag: node1 调度到具有nodetag=node1标签的节点上 本篇文章为转载内容。原文链接：https://blog.csdn.net/qq_27184497/article/details/121765387。 该文由互联网用户投稿提供，文中观点代表作者本人意见，并不代表本站的立场。 作为信息平台，本站仅提供文章转载服务，并不拥有其所有权，也不对文章内容的真实性、准确性和合法性承担责任。 如发现本文存在侵权、违法、违规或事实不符的情况，请及时联系我们，我们将第一时间进行核实并删除相应内容。

...};/ 可以从失败中恢复Driver，不过还需要指定Driver这个进程运行在Cluster，并且在提交应用程序的时候制定--supervise;/JavaStreamingContext javassc = JavaStreamingContext.getOrCreate(checkpointDirectory, factory);/ 第三步：创建Spark Streaming输入数据来源input Stream: 1、数据输入来源可以基于File、HDFS、Flume、Kafka、Socket等 2、在这里我们指定数据来源于网络Socket端口，Spark Streaming连接上该端口并在运行的时候一直监听该端口的数据 (当然该端口服务首先必须存在），并且在后续会根据业务需要不断有数据产生（当然对于Spark Streaming 应用程序的运行而言，有无数据其处理流程都是一样的） 3、如果经常在每间隔5秒钟没有数据的话不断启动空的Job其实会造成调度资源的浪费，因为并没有数据需要发生计算；所以 实际的企业级生成环境的代码在具体提交Job前会判断是否有数据，如果没有的话就不再提交Job；///创建Kafka元数据来让Spark Streaming这个Kafka Consumer利用Map<String, String> kafkaParameters = new HashMap<String, String>();kafkaParameters.put("metadata.broker.list", "Master:9092,Worker1:9092,Worker2:9092");Set<String> topics = new HashSet<String>();topics.add("SparkStreamingDirected");JavaPairInputDStream<String, String> adClickedStreaming = KafkaUtils.createDirectStream(javassc, String.class, String.class, StringDecoder.class, StringDecoder.class,kafkaParameters, topics);/因为要对黑名单进行过滤，而数据是在RDD中的，所以必然使用transform这个函数； 但是在这里我们必须使用transformToPair，原因是读取进来的Kafka的数据是Pair<String,String>类型, 另一个原因是过滤后的数据要进行进一步处理，所以必须是读进的Kafka数据的原始类型 在此再次说明，每个Batch Duration中实际上讲输入的数据就是被一个且仅被一个RDD封装的，你可以有多个 InputDStream，但其实在产生job的时候，这些不同的InputDStream在Batch Duration中就相当于Spark基于HDFS 数据操作的不同文件来源而已罢了。/JavaPairDStream<String, String> filteredadClickedStreaming = adClickedStreaming.transformToPair(new Function<JavaPairRDD<String,String>, JavaPairRDD<String,String>>() {public JavaPairRDD<String, String> call(JavaPairRDD<String, String> rdd) throws Exception {/ 在线黑名单过滤思路步骤： 1、从数据库中获取黑名单转换成RDD，即新的RDD实例封装黑名单数据； 2、然后把代表黑名单的RDD的实例和Batch Duration产生的RDD进行Join操作， 准确的说是进行leftOuterJoin操作，也就是说使用Batch Duration产生的RDD和代表黑名单的RDD实例进行 leftOuterJoin操作，如果两者都有内容的话，就会是true，否则的话就是false 我们要留下的是leftOuterJoin结果为false； /final List<String> blackListNames = new ArrayList<String>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();jdbcWrapper.doQuery("SELECT FROM blacklisttable", null, new ExecuteCallBack() {public void resultCallBack(ResultSet result) throws Exception {while(result.next()){blackListNames.add(result.getString(1));} }});List<Tuple2<String, Boolean>> blackListTuple = new ArrayList<Tuple2<String,Boolean>>();for(String name : blackListNames) {blackListTuple.add(new Tuple2<String, Boolean>(name, true));}List<Tuple2<String, Boolean>> blacklistFromListDB = blackListTuple; //数据来自于查询的黑名单表并且映射成为<String, Boolean>JavaSparkContext jsc = new JavaSparkContext(rdd.context());/ 黑名单的表中只有userID，但是如果要进行join操作的话就必须是Key-Value，所以在这里我们需要 基于数据表中的数据产生Key-Value类型的数据集合/JavaPairRDD<String, Boolean> blackListRDD = jsc.parallelizePairs(blacklistFromListDB);/ 进行操作的时候肯定是基于userID进行join，所以必须把传入的rdd进行mapToPair操作转化成为符合格式的RDD/JavaPairRDD<String, Tuple2<String, String>> rdd2Pair = rdd.mapToPair(new PairFunction<Tuple2<String,String>, String, Tuple2<String, String>>() {public Tuple2<String, Tuple2<String, String>> call(Tuple2<String, String> t) throws Exception {// TODO Auto-generated method stubString userID = t._2.split("\t")[2];return new Tuple2<String, Tuple2<String,String>>(userID, t);} });JavaPairRDD<String, Tuple2<Tuple2<String, String>, Optional<Boolean>>> joined = rdd2Pair.leftOuterJoin(blackListRDD);JavaPairRDD<String, String> result = joined.filter(new Function<Tuple2<String,Tuple2<Tuple2<String,String>,Optional<Boolean>>>, Boolean>() {public Boolean call(Tuple2<String, Tuple2<Tuple2<String, String>, Optional<Boolean>>> tuple)throws Exception {// TODO Auto-generated method stubOptional<Boolean> optional = tuple._2._2;if(optional.isPresent() && optional.get()){return false;} else {return true;} }}).mapToPair(new PairFunction<Tuple2<String,Tuple2<Tuple2<String,String>,Optional<Boolean>>>, String, String>() {public Tuple2<String, String> call(Tuple2<String, Tuple2<Tuple2<String, String>, Optional<Boolean>>> t)throws Exception {// TODO Auto-generated method stubreturn t._2._1;} });return result;} });//广告点击的基本数据格式：timestamp、ip、userID、adID、province、cityJavaPairDStream<String, Long> pairs = filteredadClickedStreaming.mapToPair(new PairFunction<Tuple2<String,String>, String, Long>() {public Tuple2<String, Long> call(Tuple2<String, String> t) throws Exception {String[] splited=t._2.split("\t");String timestamp = splited[0]; //YYYY-MM-DDString ip = splited[1];String userID = splited[2];String adID = splited[3];String province = splited[4];String city = splited[5]; String clickedRecord = timestamp + "_" +ip + "_"+userID+"_"+adID+"_"+province +"_"+city;return new Tuple2<String, Long>(clickedRecord, 1L);} });/ 第4.3步：在单词实例计数为1基础上，统计每个单词在文件中出现的总次数/JavaPairDStream<String, Long> adClickedUsers= pairs.reduceByKey(new Function2<Long, Long, Long>() {public Long call(Long i1, Long i2) throws Exception{return i1 + i2;} });/判断有效的点击，复杂化的采用机器学习训练模型进行在线过滤 简单的根据ip判断1天不超过100次；也可以通过一个batch duration的点击次数判断是否非法广告点击，通过一个batch来判断是不完整的，还需要一天的数据也可以每一个小时来判断。/JavaPairDStream<String, Long> filterClickedBatch = adClickedUsers.filter(new Function<Tuple2<String,Long>, Boolean>() {public Boolean call(Tuple2<String, Long> v1) throws Exception {if (1 < v1._2){//更新一些黑名单的数据库表return false;} else { return true;} }});//filterClickedBatch.print();//写入数据库filterClickedBatch.foreachRDD(new Function<JavaPairRDD<String,Long>, Void>() {public Void call(JavaPairRDD<String, Long> rdd) throws Exception {rdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String,Long>>>() {public void call(Iterator<Tuple2<String, Long>> partition) throws Exception {//使用数据库连接池的高效读写数据库的方式将数据写入数据库mysql//例如一次插入 1000条 records，使用insertBatch 或 updateBatch//插入的用户数据信息：userID,adID,clickedCount,time//这里面有一个问题，可能出现两条记录的key是一样的，此时需要更新累加操作List<UserAdClicked> userAdClickedList = new ArrayList<UserAdClicked>();while(partition.hasNext()) {Tuple2<String, Long> record = partition.next();String[] splited = record._1.split("\t");UserAdClicked userClicked = new UserAdClicked();userClicked.setTimestamp(splited[0]);userClicked.setIp(splited[1]);userClicked.setUserID(splited[2]);userClicked.setAdID(splited[3]);userClicked.setProvince(splited[4]);userClicked.setCity(splited[5]);userAdClickedList.add(userClicked);}final List<UserAdClicked> inserting = new ArrayList<UserAdClicked>();final List<UserAdClicked> updating = new ArrayList<UserAdClicked>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();//表的字段timestamp、ip、userID、adID、province、city、clickedCountfor(final UserAdClicked clicked : userAdClickedList) {jdbcWrapper.doQuery("SELECT clickedCount FROM adclicked WHERE"+ " timestamp =? AND userID = ? AND adID = ?",new Object[]{clicked.getTimestamp(), clicked.getUserID(),clicked.getAdID()}, new ExecuteCallBack() {public void resultCallBack(ResultSet result) throws Exception {// TODO Auto-generated method stubif(result.next()) {long count = result.getLong(1);clicked.setClickedCount(count);updating.add(clicked);} else {inserting.add(clicked);clicked.setClickedCount(1L);} }});}//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> insertParametersList = new ArrayList<Object[]>();for(UserAdClicked insertRecord : inserting) {insertParametersList.add(new Object[] {insertRecord.getTimestamp(),insertRecord.getIp(),insertRecord.getUserID(),insertRecord.getAdID(),insertRecord.getProvince(),insertRecord.getCity(),insertRecord.getClickedCount()});}jdbcWrapper.doBatch("INSERT INTO adclicked VALUES(?, ?, ?, ?, ?, ?, ?)", insertParametersList);//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> updateParametersList = new ArrayList<Object[]>();for(UserAdClicked updateRecord : updating) {updateParametersList.add(new Object[] {updateRecord.getTimestamp(),updateRecord.getIp(),updateRecord.getUserID(),updateRecord.getAdID(),updateRecord.getProvince(),updateRecord.getCity(),updateRecord.getClickedCount() + 1});}jdbcWrapper.doBatch("UPDATE adclicked SET clickedCount = ? WHERE"+ " timestamp =? AND ip = ? AND userID = ? AND adID = ? "+ "AND province = ? AND city = ?", updateParametersList);} });return null;} });//再次过滤，从数据库中读取数据过滤黑名单JavaPairDStream<String, Long> blackListBasedOnHistory = filterClickedBatch.filter(new Function<Tuple2<String,Long>, Boolean>() {public Boolean call(Tuple2<String, Long> v1) throws Exception {//广告点击的基本数据格式：timestamp,ip,userID,adID,province,cityString[] splited = v1._1.split("\t"); //提取key值String date =splited[0];String userID =splited[2];String adID =splited[3];//查询一下数据库同一个用户同一个广告id点击量超过50次列入黑名单//接下来 根据date、userID、adID条件去查询用户点击广告的数据表，获得总的点击次数//这个时候基于点击次数判断是否属于黑名单点击int clickedCountTotalToday = 81 ;if (clickedCountTotalToday > 50) {return true;}else {return false ;} }});//map操作，找出用户的idJavaDStream<String> blackListuserIDBasedInBatchOnhistroy =blackListBasedOnHistory.map(new Function<Tuple2<String,Long>, String>() {public String call(Tuple2<String, Long> v1) throws Exception {// TODO Auto-generated method stubreturn v1._1.split("\t")[2];} });//有一个问题，数据可能重复，在一个partition里面重复，这个好办；//但多个partition不能保证一个用户重复，需要对黑名单的整个rdd进行去重操作。//rdd去重了，partition也就去重了，一石二鸟，一箭双雕// 找出了黑名单，下一步就写入黑名单数据库表中JavaDStream<String> blackListUniqueuserBasedInBatchOnhistroy = blackListuserIDBasedInBatchOnhistroy.transform(new Function<JavaRDD<String>, JavaRDD<String>>() {public JavaRDD<String> call(JavaRDD<String> rdd) throws Exception {// TODO Auto-generated method stubreturn rdd.distinct();} });// 下一步写入到数据表中blackListUniqueuserBasedInBatchOnhistroy.foreachRDD(new Function<JavaRDD<String>, Void>() {public Void call(JavaRDD<String> rdd) throws Exception {rdd.foreachPartition(new VoidFunction<Iterator<String>>() {public void call(Iterator<String> t) throws Exception {// TODO Auto-generated method stub//插入的用户信息可以只包含：useID//此时直接插入黑名单数据表即可。//写入数据库List<Object[]> blackList = new ArrayList<Object[]>();while(t.hasNext()) {blackList.add(new Object[]{t.next()});}JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();jdbcWrapper.doBatch("INSERT INTO blacklisttable values (?)", blackList);} });return null;} });/广告点击累计动态更新,每个updateStateByKey都会在Batch Duration的时间间隔的基础上进行广告点击次数的更新， 更新之后我们一般都会持久化到外部存储设备上，在这里我们存储到MySQL数据库中/JavaPairDStream<String, Long> updateStateByKeyDSteam = filteredadClickedStreaming.mapToPair(new PairFunction<Tuple2<String,String>, String, Long>() {public Tuple2<String, Long> call(Tuple2<String, String> t)throws Exception {String[] splited=t._2.split("\t");String timestamp = splited[0]; //YYYY-MM-DDString ip = splited[1];String userID = splited[2];String adID = splited[3];String province = splited[4];String city = splited[5]; String clickedRecord = timestamp + "_" +ip + "_"+userID+"_"+adID+"_"+province +"_"+city;return new Tuple2<String, Long>(clickedRecord, 1L);} }).updateStateByKey(new Function2<List<Long>, Optional<Long>, Optional<Long>>() {public Optional<Long> call(List<Long> v1, Optional<Long> v2)throws Exception {// v1:当前的Key在当前的Batch Duration中出现的次数的集合，例如{1，1，1，。。。，1}// v2:当前的Key在以前的Batch Duration中积累下来的结果；Long clickedTotalHistory = 0L; if(v2.isPresent()){clickedTotalHistory = v2.get();}for(Long one : v1) {clickedTotalHistory += one;}return Optional.of(clickedTotalHistory);} });updateStateByKeyDSteam.foreachRDD(new Function<JavaPairRDD<String,Long>, Void>() {public Void call(JavaPairRDD<String, Long> rdd) throws Exception {rdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String,Long>>>() {public void call(Iterator<Tuple2<String, Long>> partition) throws Exception {//使用数据库连接池的高效读写数据库的方式将数据写入数据库mysql//例如一次插入 1000条 records，使用insertBatch 或 updateBatch//插入的用户数据信息：timestamp、adID、province、city//这里面有一个问题，可能出现两条记录的key是一样的，此时需要更新累加操作List<AdClicked> AdClickedList = new ArrayList<AdClicked>();while(partition.hasNext()) {Tuple2<String, Long> record = partition.next();String[] splited = record._1.split("\t");AdClicked adClicked = new AdClicked();adClicked.setTimestamp(splited[0]);adClicked.setAdID(splited[1]);adClicked.setProvince(splited[2]);adClicked.setCity(splited[3]);adClicked.setClickedCount(record._2);AdClickedList.add(adClicked);}final List<AdClicked> inserting = new ArrayList<AdClicked>();final List<AdClicked> updating = new ArrayList<AdClicked>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();//表的字段timestamp、ip、userID、adID、province、city、clickedCountfor(final AdClicked clicked : AdClickedList) {jdbcWrapper.doQuery("SELECT clickedCount FROM adclickedcount WHERE"+ " timestamp = ? AND adID = ? AND province = ? AND city = ?",new Object[]{clicked.getTimestamp(), clicked.getAdID(),clicked.getProvince(), clicked.getCity()}, new ExecuteCallBack() {public void resultCallBack(ResultSet result) throws Exception {// TODO Auto-generated method stubif(result.next()) {long count = result.getLong(1);clicked.setClickedCount(count);updating.add(clicked);} else {inserting.add(clicked);clicked.setClickedCount(1L);} }});}//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> insertParametersList = new ArrayList<Object[]>();for(AdClicked insertRecord : inserting) {insertParametersList.add(new Object[] {insertRecord.getTimestamp(),insertRecord.getAdID(),insertRecord.getProvince(),insertRecord.getCity(),insertRecord.getClickedCount()});}jdbcWrapper.doBatch("INSERT INTO adclickedcount VALUES(?, ?, ?, ?, ?)", insertParametersList);//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> updateParametersList = new ArrayList<Object[]>();for(AdClicked updateRecord : updating) {updateParametersList.add(new Object[] {updateRecord.getClickedCount(),updateRecord.getTimestamp(),updateRecord.getAdID(),updateRecord.getProvince(),updateRecord.getCity()});}jdbcWrapper.doBatch("UPDATE adclickedcount SET clickedCount = ? WHERE"+ " timestamp =? AND adID = ? AND province = ? AND city = ?", updateParametersList);} });return null;} });/ 对广告点击进行TopN计算，计算出每天每个省份Top5排名的广告 因为我们直接对RDD进行操作，所以使用了transfomr算子；/updateStateByKeyDSteam.transform(new Function<JavaPairRDD<String,Long>, JavaRDD<Row>>() {public JavaRDD<Row> call(JavaPairRDD<String, Long> rdd) throws Exception {JavaRDD<Row> rowRDD = rdd.mapToPair(new PairFunction<Tuple2<String,Long>, String, Long>() {public Tuple2<String, Long> call(Tuple2<String, Long> t)throws Exception {// TODO Auto-generated method stubString[] splited=t._1.split("_");String timestamp = splited[0]; //YYYY-MM-DDString adID = splited[3];String province = splited[4];String clickedRecord = timestamp + "_" + adID + "_" + province;return new Tuple2<String, Long>(clickedRecord, t._2);} }).reduceByKey(new Function2<Long, Long, Long>() {public Long call(Long v1, Long v2) throws Exception {// TODO Auto-generated method stubreturn v1 + v2;} }).map(new Function<Tuple2<String,Long>, Row>() {public Row call(Tuple2<String, Long> v1) throws Exception {// TODO Auto-generated method stubString[] splited=v1._1.split("_");String timestamp = splited[0]; //YYYY-MM-DDString adID = splited[3];String province = splited[4];return RowFactory.create(timestamp, adID, province, v1._2);} });StructType structType = DataTypes.createStructType(Arrays.asList(DataTypes.createStructField("timestamp", DataTypes.StringType, true),DataTypes.createStructField("adID", DataTypes.StringType, true),DataTypes.createStructField("province", DataTypes.StringType, true),DataTypes.createStructField("clickedCount", DataTypes.LongType, true)));HiveContext hiveContext = new HiveContext(rdd.context());DataFrame df = hiveContext.createDataFrame(rowRDD, structType);df.registerTempTable("topNTableSource");DataFrame result = hiveContext.sql("SELECT timestamp, adID, province, clickedCount, FROM"+ " (SELECT timestamp, adID, province,clickedCount, "+ "ROW_NUMBER() OVER(PARTITION BY province ORDER BY clickeCount DESC) rank "+ "FROM topNTableSource) subquery "+ "WHERE rank <= 5");return result.toJavaRDD();} }).foreachRDD(new Function<JavaRDD<Row>, Void>() {public Void call(JavaRDD<Row> rdd) throws Exception {// TODO Auto-generated method stubrdd.foreachPartition(new VoidFunction<Iterator<Row>>() {public void call(Iterator<Row> t) throws Exception {// TODO Auto-generated method stubList<AdProvinceTopN> adProvinceTopN = new ArrayList<AdProvinceTopN>();while(t.hasNext()) {Row row = t.next();AdProvinceTopN item = new AdProvinceTopN();item.setTimestamp(row.getString(0));item.setAdID(row.getString(1));item.setProvince(row.getString(2));item.setClickedCount(row.getLong(3));adProvinceTopN.add(item);}// final List<AdProvinceTopN> inserting = new ArrayList<AdProvinceTopN>();// final List<AdProvinceTopN> updating = new ArrayList<AdProvinceTopN>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();Set<String> set = new HashSet<String>();for(AdProvinceTopN item: adProvinceTopN){set.add(item.getTimestamp() + "_" + item.getProvince());}//表的字段timestamp、adID、province、clickedCountArrayList<Object[]> deleteParametersList = new ArrayList<Object[]>();for(String deleteRecord : set) {String[] splited = deleteRecord.split("_");deleteParametersList.add(new Object[]{splited[0],splited[1]});}jdbcWrapper.doBatch("DELETE FROM adprovincetopn WHERE timestamp = ? AND province = ?", deleteParametersList);//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> insertParametersList = new ArrayList<Object[]>();for(AdProvinceTopN insertRecord : adProvinceTopN) {insertParametersList.add(new Object[] {insertRecord.getClickedCount(),insertRecord.getTimestamp(),insertRecord.getAdID(),insertRecord.getProvince()});}jdbcWrapper.doBatch("INSERT INTO adprovincetopn VALUES (?, ?, ?, ?)", insertParametersList);} });return null;} });/ 计算过去半个小时内广告点击的趋势 广告点击的基本数据格式：timestamp、ip、userID、adID、province、city/filteredadClickedStreaming.mapToPair(new PairFunction<Tuple2<String,String>, String, Long>() {public Tuple2<String, Long> call(Tuple2<String, String> t)throws Exception {String splited[] = t._2.split("\t");String adID = splited[3];String time = splited[0]; //Todo:后续需要重构代码实现时间戳和分钟的转换提取。此处需要提取出该广告的点击分钟单位return new Tuple2<String, Long>(time + "_" + adID, 1L);} }).reduceByKeyAndWindow(new Function2<Long, Long, Long>() {public Long call(Long v1, Long v2) throws Exception {// TODO Auto-generated method stubreturn v1 + v2;} }, new Function2<Long, Long, Long>() {public Long call(Long v1, Long v2) throws Exception {// TODO Auto-generated method stubreturn v1 - v2;} }, Durations.minutes(30), Durations.milliseconds(5)).foreachRDD(new Function<JavaPairRDD<String,Long>, Void>() {public Void call(JavaPairRDD<String, Long> rdd) throws Exception {// TODO Auto-generated method stubrdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String,Long>>>() {public void call(Iterator<Tuple2<String, Long>> partition)throws Exception {List<AdTrendStat> adTrend = new ArrayList<AdTrendStat>();// TODO Auto-generated method stubwhile(partition.hasNext()) {Tuple2<String, Long> record = partition.next();String[] splited = record._1.split("_");String time = splited[0];String adID = splited[1];Long clickedCount = record._2;/ 在插入数据到数据库的时候具体需要哪些字段？time、adID、clickedCount; 而我们通过J2EE技术进行趋势绘图的时候肯定是需要年、月、日、时、分这个维度的，所以我们在这里需要 年月日、小时、分钟这些时间维度；/AdTrendStat adTrendStat = new AdTrendStat();adTrendStat.setAdID(adID);adTrendStat.setClickedCount(clickedCount);adTrendStat.set_date(time); //Todo:获取年月日adTrendStat.set_hour(time); //Todo:获取小时adTrendStat.set_minute(time);//Todo:获取分钟adTrend.add(adTrendStat);}final List<AdTrendStat> inserting = new ArrayList<AdTrendStat>();final List<AdTrendStat> updating = new ArrayList<AdTrendStat>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();//表的字段timestamp、ip、userID、adID、province、city、clickedCountfor(final AdTrendStat trend : adTrend) {final AdTrendCountHistory adTrendhistory = new AdTrendCountHistory();jdbcWrapper.doQuery("SELECT clickedCount FROM adclickedtrend WHERE"+ " date =? AND hour = ? AND minute = ? AND AdID = ?",new Object[]{trend.get_date(), trend.get_hour(), trend.get_minute(),trend.getAdID()}, new ExecuteCallBack() {public void resultCallBack(ResultSet result) throws Exception {// TODO Auto-generated method stubif(result.next()) {long count = result.getLong(1);adTrendhistory.setClickedCountHistoryLong(count);updating.add(trend);} else { inserting.add(trend);} }});}//表的字段date、hour、minute、adID、clickedCountList<Object[]> insertParametersList = new ArrayList<Object[]>();for(AdTrendStat insertRecord : inserting) {insertParametersList.add(new Object[] {insertRecord.get_date(),insertRecord.get_hour(),insertRecord.get_minute(),insertRecord.getAdID(),insertRecord.getClickedCount()});}jdbcWrapper.doBatch("INSERT INTO adclickedtrend VALUES(?, ?, ?, ?, ?)", insertParametersList);//表的字段date、hour、minute、adID、clickedCountList<Object[]> updateParametersList = new ArrayList<Object[]>();for(AdTrendStat updateRecord : updating) {updateParametersList.add(new Object[] {updateRecord.getClickedCount(),updateRecord.get_date(),updateRecord.get_hour(),updateRecord.get_minute(),updateRecord.getAdID()});}jdbcWrapper.doBatch("UPDATE adclickedtrend SET clickedCount = ? WHERE"+ " date =? AND hour = ? AND minute = ? AND AdID = ?", updateParametersList);} });return null;} });;/ Spark Streaming 执行引擎也就是Driver开始运行，Driver启动的时候是位于一条新的线程中的，当然其内部有消息循环体，用于 接收应用程序本身或者Executor中的消息，/javassc.start();javassc.awaitTermination();javassc.close();}private static JavaStreamingContext createContext(String checkpointDirectory, SparkConf conf) {// If you do not see this printed, that means the StreamingContext has been loaded// from the new checkpointSystem.out.println("Creating new context");// Create the context with a 5 second batch sizeJavaStreamingContext ssc = new JavaStreamingContext(conf, Durations.seconds(10));ssc.checkpoint(checkpointDirectory);return ssc;} }class JDBCWrapper {private static JDBCWrapper jdbcInstance = null;private static LinkedBlockingQueue<Connection> dbConnectionPool = new LinkedBlockingQueue<Connection>();static {try {Class.forName("com.mysql.jdbc.Driver");} catch (ClassNotFoundException e) {// TODO Auto-generated catch blocke.printStackTrace();} }public static JDBCWrapper getJDBCInstance() {if(jdbcInstance == null) {synchronized (JDBCWrapper.class) {if(jdbcInstance == null) {jdbcInstance = new JDBCWrapper();} }}return jdbcInstance; }private JDBCWrapper() {for(int i = 0; i < 10; i++){try {Connection conn = DriverManager.getConnection("jdbc:mysql://Master:3306/sparkstreaming","root", "root");dbConnectionPool.put(conn);} catch (Exception e) {// TODO Auto-generated catch blocke.printStackTrace();} } }public synchronized Connection getConnection() {while(0 == dbConnectionPool.size()){try {Thread.sleep(20);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} }return dbConnectionPool.poll();}public int[] doBatch(String sqlText, List<Object[]> paramsList){Connection conn = getConnection();PreparedStatement preparedStatement = null;int[] result = null;try {conn.setAutoCommit(false);preparedStatement = conn.prepareStatement(sqlText);for(Object[] parameters: paramsList) {for(int i = 0; i < parameters.length; i++){preparedStatement.setObject(i + 1, parameters[i]);} preparedStatement.addBatch();}result = preparedStatement.executeBatch();conn.commit();} catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();} finally {if(preparedStatement != null) {try {preparedStatement.close();} catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();} }if(conn != null) {try {dbConnectionPool.put(conn);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} }}return result; }public void doQuery(String sqlText, Object[] paramsList, ExecuteCallBack callback){Connection conn = getConnection();PreparedStatement preparedStatement = null;ResultSet result = null;try {preparedStatement = conn.prepareStatement(sqlText);for(int i = 0; i < paramsList.length; i++){preparedStatement.setObject(i + 1, paramsList[i]);} result = preparedStatement.executeQuery();try {callback.resultCallBack(result);} catch (Exception e) {// TODO Auto-generated catch blocke.printStackTrace();} } catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();} finally {if(preparedStatement != null) {try {preparedStatement.close();} catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();} }if(conn != null) {try {dbConnectionPool.put(conn);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} }} }}interface ExecuteCallBack {void resultCallBack(ResultSet result) throws Exception;}class UserAdClicked {private String timestamp;private String ip;private String userID;private String adID;private String province;private String city;private Long clickedCount;public String getTimestamp() {return timestamp;}public void setTimestamp(String timestamp) {this.timestamp = timestamp;}public String getIp() {return ip;}public void setIp(String ip) {this.ip = ip;}public String getUserID() {return userID;}public void setUserID(String userID) {this.userID = userID;}public String getAdID() {return adID;}public void setAdID(String adID) {this.adID = adID;}public String getProvince() {return province;}public void setProvince(String province) {this.province = province;}public String getCity() {return city;}public void setCity(String city) {this.city = city;}public Long getClickedCount() {return clickedCount;}public void setClickedCount(Long clickedCount) {this.clickedCount = clickedCount;} }class AdClicked {private String timestamp;private String adID;private String province;private String city;private Long clickedCount;public String getTimestamp() {return timestamp;}public void setTimestamp(String timestamp) {this.timestamp = timestamp;}public String getAdID() {return adID;}public void setAdID(String adID) {this.adID = adID;}public String getProvince() {return province;}public void setProvince(String province) {this.province = province;}public String getCity() {return city;}public void setCity(String city) {this.city = city;}public Long getClickedCount() {return clickedCount;}public void setClickedCount(Long clickedCount) {this.clickedCount = clickedCount;} }class AdProvinceTopN {private String timestamp;private String adID;private String province;private Long clickedCount;public String getTimestamp() {return timestamp;}public void setTimestamp(String timestamp) {this.timestamp = timestamp;}public String getAdID() {return adID;}public void setAdID(String adID) {this.adID = adID;}public String getProvince() {return province;}public void setProvince(String province) {this.province = province;}public Long getClickedCount() {return clickedCount;}public void setClickedCount(Long clickedCount) {this.clickedCount = clickedCount;} }class AdTrendStat {private String _date;private String _hour;private String _minute;private String adID;private Long clickedCount;public String get_date() {return _date;}public void set_date(String _date) {this._date = _date;}public String get_hour() {return _hour;}public void set_hour(String _hour) {this._hour = _hour;}public String get_minute() {return _minute;}public void set_minute(String _minute) {this._minute = _minute;}public String getAdID() {return adID;}public void setAdID(String adID) {this.adID = adID;}public Long getClickedCount() {return clickedCount;}public void setClickedCount(Long clickedCount) {this.clickedCount = clickedCount;} }class AdTrendCountHistory{private Long clickedCountHistoryLong;public Long getClickedCountHistoryLong() {return clickedCountHistoryLong;}public void setClickedCountHistoryLong(Long clickedCountHistoryLong) {this.clickedCountHistoryLong = clickedCountHistoryLong;} } 本篇文章为转载内容。原文链接：https://blog.csdn.net/tom_8899_li/article/details/71194434。 该文由互联网用户投稿提供，文中观点代表作者本人意见，并不代表本站的立场。 作为信息平台，本站仅提供文章转载服务，并不拥有其所有权，也不对文章内容的真实性、准确性和合法性承担责任。 如发现本文存在侵权、违法、违规或事实不符的情况，请及时联系我们，我们将第一时间进行核实并删除相应内容。

...ysql.conf 启动 logstash.bat Logstash多实例运行 0x04 搜素服务：查询课程媒资接口 需求分析 Api接口定义 Service Controller 测试 三、在线学习：接口开发 0x01 需求分析 0x02 搭建开发环境 0x03 Api接口 0x04 服务端开发 需求分析 搜索服务注册Eureka 搜索服务客户端 自定义错误代码 Service Controller 测试 0x05 前端开发 需求分析 api方法 配置代理 视频播放页面 简单的测试 完整的测试 1、上传文件 一些问题 ~~方案1：删除本地分块文件重新尝试上传~~ 方案2：检查前端提交的MD5值是否正确 2、为课程计划选择媒资信息 3、前端门户测试 四、待完善的一些功能 😁 认识作者 一、学习页面：查询课程计划 0x01 需求分析 到目前为止，我们已可以编辑课程计划信息并上传课程视频，下一步我们要实现在线学习页面动态读取章节对应的视频并进行播放。在线学习页面所需要的信息有两类： 课程计划信息 课程学习信息（视频地址、学习进度等） 如下图： 在线学习集成媒资管理的需求如下： 1、在线学习页面显示课程计划 2、点击课程计划播放该课程计划对应的视频 本章节实现学习页面动态显示课程计划，进入不同课程的学习页面右侧动态显示当前课程的课程计划。 0x02 Api接口 课程计划信息从哪里获取？ 在课程发布完成后会自动发布到一个 course_pub 的表中，logstash 会自动将课程发布后的信息自动采集到 ES 索引库中，这些信息也包含课程计划信息。 所以考虑性能要求，课程发布后对课程的查询统一从 ES 索引库中查询。 前端通过请求 搜索服务 获取课程信息，需要单独在 搜索服务 中定义课程信息查询接口。 本接口接收课程id，查询课程所有信息返回给前端。 我们在搜素服务 API 下添加以下方法 @ApiOperation("根据id搜索课程发布信息")public Map<String,CoursePub> getdetail(String id); 返回的课程信息为 json 结构：key 为课程id，value 为课程内容。 0x03 服务端开发 在搜索服务中开发查询课程信息接口。 Controller 在搜素服务下添加以下方法 / 根据id搜索课程发布信息 @param id 课程id @return JSON数据/@Override@GetMapping("/getdetail/{id}")public Map<String, CoursePub> getdetail(@PathVariable("id")String id) {return esCourseService.getdetail(id);} Service / 根据id搜索课程发布信息 @param id 课程id @return JSON数据/public Map<String, CoursePub> getdetail(String id) {//设置索引SearchRequest searchRequest = new SearchRequest(es_index);//设置类型searchRequest.types(es_type);//创建搜索源对象SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();//设置查询条件,根据id进行查询searchSourceBuilder.query(QueryBuilders.termQuery("id",id));//这里不使用source的原字段过滤,查询所有字段// searchSourceBuilder.fetchSource(new String[]{"name", "grade", "charge","pic"}, newString[]{});//设置搜索源对象searchRequest.source(searchSourceBuilder);//执行搜索SearchResponse searchResponse = null;try {searchResponse = restHighLevelClient.search(searchRequest);} catch (IOException e) {e.printStackTrace();}//获取搜索结果SearchHits hits = searchResponse.getHits();SearchHit[] searchHits = hits.getHits(); //获取最优结果Map<String,CoursePub> map = new HashMap<>();for (SearchHit hit: searchHits) {//从搜索结果中取值并添加到coursePub对象Map<String, Object> sourceAsMap = hit.getSourceAsMap();String courseId = (String) sourceAsMap.get("id");String name = (String) sourceAsMap.get("name");String grade = (String) sourceAsMap.get("grade");String charge = (String) sourceAsMap.get("charge");String pic = (String) sourceAsMap.get("pic");String description = (String) sourceAsMap.get("description");String teachplan = (String) sourceAsMap.get("teachplan");CoursePub coursePub = new CoursePub();coursePub.setId(courseId);coursePub.setName(name);coursePub.setPic(pic);coursePub.setGrade(grade);coursePub.setTeachplan(teachplan);coursePub.setDescription(description);//设置map对象map.put(courseId,coursePub);}return map;} 测试 使用 swagger-ui 或 postman 测试查询课程信息接口。 0x04 前端开发 配置NGINX虚拟主机 学习中心的二级域名为 ucenter.xuecheng.com ，我们在 nginx 中配置 ucenter 虚拟主机。 学成网用户中心server {listen 80;server_name ucenter.xuecheng.com;个人中心location / {proxy_pass http://ucenter_server_pool;} } 前端ucenterupstream ucenter_server_pool{server 127.0.0.1:7081 weight=10;server 127.0.0.1:13000 weight=10;} 在学习中心要调用搜索的 API，使用 Nginx 解决代理，如下图： 在 ucenter 虚拟主机下配置搜索 Api 代理路径 后台搜索（公开api）upstream search_server_pool{server 127.0.0.1:40100 weight=10;} 学成网用户中心server {listen 80;server_name ucenter.xuecheng.com;个人中心location / {proxy_pass http://ucenter_server_pool;}后端搜索服务location /openapi/search/ {proxy_pass http://search_server_pool/search/;} } 前端 API 方法 在学习中心 xc-ui-pc-leanring 对课程信息的查询属于基础常用功能，所以我们将课程查询的 api 方法定义在base 模块下，如下图： 在system.js 中定义课程查询方法： import http from './public'export const course_view = id => {return http.requestGet('/openapi/search/course/getdetail/'+id);} 前端 API 方法调用 在 learning_video.vue 页面中调用课程信息查询接口得到课程计划，将课程计划json 串转成对象。 xc-ui-pc-leanring/src/module/course/page/learning_video.vue 1、定义视图 课程计划 <!--课程计划部分代码--><div class="navCont"><div class="course-weeklist"><div class="nav nav-stacked" v-for="(teachplan_first, index) in teachplanList"><div class="tit nav-justified text-center"><i class="pull-left glyphicon glyphicon-th-list"></i>{ {teachplan_first.pname} }<i class="pull-right"></i></div><li v-if="teachplan_first.children!=null" v-for="(teachplan_second, index) in teachplan_first.children"><i class="glyphicon glyphicon-check"></i><a :href="url" @click="study(teachplan_second.id)">{ {teachplan_second.pname} }</a></li><!-- <div class="tit nav-justified text-center"><i class="pull-left glyphicon glyphicon-th-list"></i>第一章<i class="pull-right"></i></div><li ><i class="glyphicon glyphicon-check"></i><a :href="url" >第一节</a></li>--><!--<li><i class="glyphicon glyphicon-unchecked"></i>为什么分为A、B、C部分</li>--></div></div></div> 课程名称 <div class="top text-center">{ {coursename} }</div> 定义数据对象 data() {return {url:'',//当前urlcourseId:'',//课程idchapter:'',//章节Idcoursename:'',//课程名称coursepic:'',//课程图片teachplanList:[],//课程计划playerOptions: {//播放参数autoplay: false,controls: true,sources: [{type: "application/x-mpegURL",src: ''}]},} } 在 created 钩子方法中获取课程信息 created(){//当前请求的urlthis.url = window.location//课程idthis.courseId = this.$route.params.courseId//章节idthis.chapter = this.$route.params.chapter//查询课程信息systemApi.course_view(this.courseId).then((view_course)=>{if(!view_course || !view_course[this.courseId]){this.$message.error("获取课程信息失败，请重新进入此页面！")return ;} let courseInfo = view_course[this.courseId]console.log(courseInfo)this.coursename = courseInfo.nameif(courseInfo.teachplan){let teachplan = JSON.parse(courseInfo.teachplan);this.teachplanList = teachplan.children;} })}, 测试 在浏览器请求：http://ucenter.xuecheng.com//learning/4028e581617f945f01617f9dabc40000/0 4028e581617f945f01617f9dabc40000：第一个参数为课程 id，测试时从 ES索引库找一个课程 id 0：第二个参数为课程计划 id，此参数用于点击课程计划播放视频。 如果出现跨域问题，但是确定已经配置了跨域，请尝试结束所以 nginx.exe 的进程 和 清空浏览器缓存。 如果还没有解决？重启电脑试试。 二、学习页面：获取视频播放地址 0x01 需求分析 用户进入在线学习页面，点击课程计划将播放该课程计划对应的教学视频。 业务流程如下： 业务流程说明： 1、用户进入在线学习页面，页面请求搜索服务获取课程信息（包括课程计划信息）并且在页面展示。 2、在线学习请求学习服务获取视频播放地址。 3、学习服务校验当前用户是否有权限学习，如果没有权限学习则提示用户。 4、学习服务校验通过，请求搜索服务获取课程媒资信息。 5、搜索服务请求ElasticSearch获取课程媒资信息。 为什么要请求 ElasticSearch 查询课程媒资信息？ 出于性能的考虑，公开查询课程信息从搜索服务查询，分摊 mysql 数据库的访问压力。 什么时候将课程媒资信息存储到 ElasticSearch 中？ 课程媒资信息是在课程发布的时候存入 ElasticSearch，因为课程发布后课程信息将基本不再修改。 0x02 课程发布：储存媒资信息 需求分析 课程媒资信息是在课程发布的时候存入 ElasticSearch 索引库，因为课程发布后课程信息将基本不再修改，具体的业务流程如下。 1、课程发布，向课程媒资信息表写入数据。 1）根据课程 id 删除 teachplanMediaPub 中的数据 2）根据课程 id 查询 teachplanMedia 数据 3）将查询到的 teachplanMedia 数据插入到 teachplanMediaPub 中 2、Logstash 定时扫描课程媒资信息表，并将课程媒资信息写入索引库。 数据模型 在 xc_course 数据库创建课程计划媒资发布表： CREATE TABLE teachplan_media_pub (teachplan_id varchar(32) NOT NULL COMMENT '课程计划id',media_id varchar(32) NOT NULL COMMENT '媒资文件id',media_fileoriginalname varchar(128) NOT NULL COMMENT '媒资文件的原始名称',media_url varchar(256) NOT NULL COMMENT '媒资文件访问地址',courseid varchar(32) NOT NULL COMMENT '课程Id',timestamp timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT'logstash使用',PRIMARY KEY (teachplan_id)) ENGINE=InnoDB DEFAULT CHARSET=utf8 数据模型类如下： package com.xuecheng.framework.domain.course;import lombok.Data;import lombok.ToString;import org.hibernate.annotations.GenericGenerator;import javax.persistence.;import java.io.Serializable;import java.util.Date;@Data@ToString@Entity@Table(name="teachplan_media_pub")@GenericGenerator(name = "jpa-assigned", strategy = "assigned")public class TeachplanMediaPub implements Serializable {private static final long serialVersionUID = -916357110051689485L;@Id@GeneratedValue(generator = "jpa-assigned")@Column(name="teachplan_id")private String teachplanId;@Column(name="media_id")private String mediaId;@Column(name="media_fileoriginalname")private String mediaFileOriginalName;@Column(name="media_url")private String mediaUrl;@Column(name="courseid")private String courseId;@Column(name="timestamp")private Date timestamp;//时间戳} Dao 创建 TeachplanMediaPub 表的 Dao，向 TeachplanMediaPub 存储信息采用先删除该课程的媒资信息，再添加该课程的媒资信息，所以这里定义根据课程 id 删除课程计划媒资方法： public interface TeachplanMediaPubRepository extends JpaRepository<TeachplanMediaPub, String> {//根据课程id删除课程计划媒资信息long deleteByCourseId(String courseId);} 从TeachplanMedia查询课程计划媒资信息 //从TeachplanMedia查询课程计划媒资信息public interface TeachplanMediaRepository extends JpaRepository<TeachplanMedia, String> {List<TeachplanMedia> findByCourseId(String courseId);} Service 编写保存课程计划媒资信息方法，并在课程发布时调用此方法。 1、保存课程计划媒资信息方法 本方法采用先删除该课程的媒资信息，再添加该课程的媒资信息，在 CourseService 下定义该方法 //保存课程计划媒资信息private void saveTeachplanMediaPub(String courseId){//查询课程媒资信息List<TeachplanMedia> byCourseId = teachplanMediaRepository.findByCourseId(courseId);if(byCourseId == null) return; //没有查询到媒资数据则直接结束该方法//将课程计划媒资信息储存到待索引表//删除原有的索引信息teachplanMediaPubRepository.deleteByCourseId(courseId);//一个课程可能会有多个媒资信息,遍历并使用list进行储存List<TeachplanMediaPub> teachplanMediaPubList = new ArrayList<>();for (TeachplanMedia teachplanMedia: byCourseId) {TeachplanMediaPub teachplanMediaPub = new TeachplanMediaPub();BeanUtils.copyProperties(teachplanMedia, teachplanMediaPub);teachplanMediaPubList.add(teachplanMediaPub);}//保存所有信息teachplanMediaPubRepository.saveAll(teachplanMediaPubList);} 2、课程发布时调用此方法 修改课程发布的 coursePublish 方法： ....//保存课程计划媒资信息到待索引表saveTeachplanMediaPub(courseId);//页面urlString pageUrl = cmsPostPageResult.getPageUrl();return new CoursePublishResult(CommonCode.SUCCESS,pageUrl);..... 测试 测试课程发布后是否成功将课程媒资信息存储到 teachplan_media_pub 中，测试流程如下： 1、指定一个课程 2、为课程计划添加课程媒资 3、执行课程发布 4、观察课程计划媒资信息是否存储至 teachplan_media_pub 中 注意：由于此测试仅用于测试发布课程计划媒资信息的功能，可暂时将 cms页面发布的功能暂时屏蔽，提高测试效率。 测试结果如下 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Vrzs5589-1595567273126)(https://qnoss.codeyee.com/20200704_15/image7)] 0x03 Logstash：扫描课程计划媒资 Logstash 定时扫描课程媒资信息表，并将课程媒资信息写入索引库。 创建索引 1、创建 xc_course_media 索引 2、并向此索引创建如下映射 POST: http://localhost:9200/xc_course_media/doc/_mapping {"properties" : {"courseid" : {"type" : "keyword"},"teachplan_id" : {"type" : "keyword"},"media_id" : {"type" : "keyword"},"media_url" : {"index" : false,"type" : "text"},"media_fileoriginalname" : {"index" : false,"type" : "text"} }} 索引创建成功 创建模板文件 在 logstach 的 config 目录文件 xc_course_media_template.json 文件路径为 %ES_ROOT_DIR%/logstash6.8.8/config/xc_course_media_template.json %ES_ROOT_DIR% 为 ElasticSearch 和 logstash 的安装目录 内容如下： {"mappings" : {"doc" : {"properties" : {"courseid" : {"type" : "keyword"},"teachplan_id" : {"type" : "keyword"},"media_id" : {"type" : "keyword"},"media_url" : {"index" : false,"type" : "text"},"media_fileoriginalname" : {"index" : false,"type" : "text"} }},"template" : "xc_course_media"} } 配置 mysql.conf 在logstash的 config 目录下配置 mysql_course_media.conf 文件供 logstash 使用，logstash 会根据 mysql_course_media.conf 文件的配置的地址从 MySQL 中读取数据向 ES 中写入索引。 参考https://www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html 配置输入数据源和输出数据源。 input {stdin {} jdbc {jdbc_connection_string => "jdbc:mysql://localhost:3306/xc_course?useUnicode=true&characterEncoding=utf-8&useSSL=true&serverTimezone=UTC" 数据库信息jdbc_user => "root"jdbc_password => "123123" MYSQL 驱动地址,修改为maven仓库对应的位置jdbc_driver_library => "D:/soft/apache-maven-3.5.4/repository/mysql/mysql-connector-java/5.1.40/mysql-connector-java-5.1.40.jar" the name of the driver class for mysqljdbc_driver_class => "com.mysql.jdbc.Driver"jdbc_paging_enabled => "true"jdbc_page_size => "50000"要执行的sql文件statement_filepath => "/conf/course.sql"statement => "select from teachplan_media_pub where timestamp > date_add(:sql_last_value,INTERVAL 8 HOUR)"定时配置schedule => " "record_last_run => truelast_run_metadata_path => "D:/soft/elasticsearch/logstash-6.8.8/config/xc_course_media_metadata"} } output {elasticsearch {ES的ip地址和端口hosts => "localhost:9200"hosts => ["localhost:9200","localhost:9202","localhost:9203"]ES索引库名称index => "xc_course_media"document_id => "%{teachplan_id}"document_type => "doc"template => "D:/soft/elasticsearch/logstash-6.8.8/config/xc_course_media_template.json"template_name =>"xc_course_media"template_overwrite =>"true"} stdout {日志输出codec => json_lines} } 启动 logstash.bat 启动 logstash.bat 采集 teachplan_media_pub 中的数据，向 ES 写入索引。 logstash.bat -f ../config/mysql_course_media.conf 课程发布成功后，Logstash 会自动参加 teachplan_media_pub 表中新增的数据，效果如下 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ILPBxfXi-1595567273134)(https://qnoss.codeyee.com/20200704_15/image10)] Logstash多实例运行 由于之前我们还启动了一个 Logstash 对课程的发布信息进行采集，所以如果想两个 logstash 实例同时运行，因为每个实例都有一个.lock文件，所以不能使用同一个目录来存放数据，所以我们需要使用 --path.data= 为每个实例指定单独的数据目录，具体的代码如下： 该配置是在windows下进行的 课程发布实例 logstash_start_course_pub.bat @title logstash in course_publogstash.bat -f ..\config\mysql.conf --path.data=../data/course_pub 课程计划媒体发布实例 logstash_start_teachplan_media.bat @title logstash i n teachplan_media_publogstash.bat -f ../config/mysql_course_media.conf --path.data=../data/teachplan_media/ 同时运行效果如下 0x04 搜素服务：查询课程媒资接口 需求分析 搜索服务 提供查询课程媒资接口，此接口供学习服务调用。 Api接口定义 @ApiOperation("根据课程计划查询媒资信息")public TeachplanMediaPub getmedia(String teachplanId); Service 1、配置课程计划媒资索引库等信息 在 application.yml 中配置 xuecheng:elasticsearch:hostlist: ${eshostlist:127.0.0.1:9200} 多个结点中间用逗号分隔course:index: xc_coursetype: docsource_field: id,name,grade,mt,st,charge,valid,pic,qq,price,price_old,status,studymodel,teachmode,expires,pub_time,start_time,end_timemedia:index: xc_course_mediatype: docsource_field: courseid,media_id,media_url,teachplan_id,media_fileoriginalname 2、service 方法开发 在 课程搜索服务 中定义课程媒资查询接口，为了适应后续需求，service 参数定义为数组，可一次查询多个课程计划的媒资信息。 / 根据一个或者多个课程计划id查询媒资信息 @param teachplanIds 课程id @return QueryResponseResult/public QueryResponseResult<TeachplanMediaPub> getmedia(String [] teachplanIds){//设置索引SearchRequest searchRequest = new SearchRequest(media_index);//设置类型searchRequest.types(media_type);//创建搜索源对象SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();//源字段过滤String[] media_index_arr = media_field.split(",");searchSourceBuilder.fetchSource(media_index_arr, new String[]{});//查询条件,根据课程计划id查询(可以传入多个课程计划id)searchSourceBuilder.query(QueryBuilders.termsQuery("teachplan_id", teachplanIds));searchRequest.source(searchSourceBuilder);SearchResponse searchResponse = null;try {searchResponse = restHighLevelClient.search(searchRequest);} catch (IOException e) {e.printStackTrace();}//获取结果SearchHits hits = searchResponse.getHits();long totalHits = hits.getTotalHits();SearchHit[] searchHits = hits.getHits();//数据列表List<TeachplanMediaPub> teachplanMediaPubList = new ArrayList<>();for(SearchHit hit:searchHits){TeachplanMediaPub teachplanMediaPub =new TeachplanMediaPub();Map<String, Object> sourceAsMap = hit.getSourceAsMap();//取出课程计划媒资信息String courseid = (String) sourceAsMap.get("courseid");String media_id = (String) sourceAsMap.get("media_id");String media_url = (String) sourceAsMap.get("media_url");String teachplan_id = (String) sourceAsMap.get("teachplan_id");String media_fileoriginalname = (String) sourceAsMap.get("media_fileoriginalname");teachplanMediaPub.setCourseId(courseid);teachplanMediaPub.setMediaUrl(media_url);teachplanMediaPub.setMediaFileOriginalName(media_fileoriginalname);teachplanMediaPub.setMediaId(media_id);teachplanMediaPub.setTeachplanId(teachplan_id);//将对象加入到列表中teachplanMediaPubList.add(teachplanMediaPub);}//构建返回课程媒资信息对象QueryResult<TeachplanMediaPub> queryResult = new QueryResult<>();queryResult.setList(teachplanMediaPubList);queryResult.setTotal(totalHits);return new QueryResponseResult<TeachplanMediaPub>(CommonCode.SUCCESS,queryResult);} Controller / 根据课程计划id搜索发布后的媒资信息 @param teachplanId @return/@GetMapping(value="/getmedia/{teachplanId}")@Overridepublic TeachplanMediaPub getmedia(@PathVariable("teachplanId") String teachplanId) {//为了service的拓展性,所以我们service接收的是数组作为参数,以便后续开发查询多个ID的接口String[] teachplanIds = new String[]{teachplanId};//通过service查询ES获取课程媒资信息QueryResponseResult<TeachplanMediaPub> mediaPubQueryResponseResult = esCourseService.getmedia(teachplanIds);QueryResult<TeachplanMediaPub> queryResult = mediaPubQueryResponseResult.getQueryResult();if(queryResult!=null&& queryResult.getList()!=null&& queryResult.getList().size()>0){//返回课程计划对应课程媒资return queryResult.getList().get(0);} return new TeachplanMediaPub();} 测试 使用 swagger-ui 和 postman 测试课程媒资查询接口。 三、在线学习：接口开发 0x01 需求分析 根据下边的业务流程，本章节完成前端学习页面请求学习服务获取课程视频地址，并自动播放视频。 0x02 搭建开发环境 1、创建数据库 创建 xc_learning 数据库，学习数据库将记录学生的选课信息、学习信息。 导入：资料/xc_learning.sql 2、创建学习服务工程 参考课程管理服务工程结构，创建学习服务工程： 导入：资料/xc-service-learning.zip 项目工程结构如下 0x03 Api接口 此 api 接口是课程学习页面请求学习服务获取课程学习地址。 定义返回值类型： package com.xuecheng.framework.domain.learning.response;import com.xuecheng.framework.model.response.ResponseResult;import com.xuecheng.framework.model.response.ResultCode;import lombok.Data;import lombok.NoArgsConstructor;import lombok.ToString;@Data@ToString@NoArgsConstructorpublic class GetMediaResult extends ResponseResult {public GetMediaResult(ResultCode resultCode, String fileUrl) {super(resultCode);this.fileUrl = fileUrl;}//媒资文件播放地址private String fileUrl;} 定义接口，学习服务根据传入课程 ID、章节 Id(课程计划 ID)来取学习地址。 @Api(value = "录播课程学习管理",description = "录播课程学习管理")public interface CourseLearningControllerApi {@ApiOperation("获取课程学习地址")public GetMediaResult getMediaPlayUrl(String courseId,String teachplanId);} 0x04 服务端开发 需求分析 学习服务根据传入课程ID、章节Id(课程计划ID)请求搜索服务获取学习地址。 搜索服务注册Eureka 学习服务要调用搜索服务查询课程媒资信息，所以需要将搜索服务注册到 eureka 中。 1、查看服务名称是否为 xc-service-search 注意修改application.xml中的服务名称：spring:application:name: xc‐service‐search 2、配置搜索服务的配置文件 application.yml，加入 Eureka 配置 如下： eureka:client:registerWithEureka: true 服务注册开关fetchRegistry: true 服务发现开关serviceUrl: Eureka客户端与Eureka服务端进行交互的地址，多个中间用逗号分隔defaultZone: ${EUREKA_SERVER:http://localhost:50101/eureka/,http://localhost:50102/eureka/}instance:prefer-ip-address: true 将自己的ip地址注册到Eureka服务中ip-address: ${IP_ADDRESS:127.0.0.1}instance-id: ${spring.application.name}:${server.port} 指定实例idribbon:MaxAutoRetries: 2 最大重试次数，当Eureka中可以找到服务，但是服务连不上时将会重试，如果eureka中找不到服务则直接走断路器MaxAutoRetriesNextServer: 3 切换实例的重试次数OkToRetryOnAllOperations: false 对所有操作请求都进行重试，如果是get则可以，如果是post，put等操作没有实现幂等的情况下是很危险的,所以设置为falseConnectTimeout: 5000 请求连接的超时时间ReadTimeout: 6000 请求处理的超时时间 3、添加 eureka 依赖 <dependency><groupId>org.springframework.cloud</groupId><artifactId>spring‐cloud‐starter‐netflix‐eureka‐client</artifactId></dependency> 4、修改启动类，在class上添加如下注解： @EnableDiscoveryClient 搜索服务客户端 在 学习服务 创建搜索服务的客户端接口，此接口会生成代理对象，调用搜索服务： package com.xuecheng.learning.client;import com.xuecheng.framework.domain.course.TeachplanMediaPub;import org.springframework.cloud.openfeign.FeignClient;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.PathVariable;@FeignClient(value = "xc‐service‐search")public interface CourseSearchClient {@GetMapping(value="/getmedia/{teachplanId}")public TeachplanMediaPub getmedia(@PathVariable("teachplanId") String teachplanId);} 自定义错误代码 我们在 com.xuecheng.framework.domain.learning.response 包下自定义一个错误消息模型 package com.xuecheng.framework.domain.learning.response;import com.xuecheng.framework.model.response.ResultCode;import lombok.ToString;@ToStringpublic enum LearningCode implements ResultCode {LEARNING_GET_MEDIA_ERROR(false,23001,"学习中心获取媒资信息错误！");//操作代码boolean success;//操作代码int code;//提示信息String message;private LearningCode(boolean success, int code, String message){this.success = success;this.code = code;this.message = message;}@Overridepublic boolean success() {return success;}@Overridepublic int code() {return code;}@Overridepublic String message() {return message;} } 该消息模型基于 ResultCode 来实现，代码如下 package com.xuecheng.framework.model.response;/ Created by mrt on 2018/3/5. 10000-- 通用错误代码 22000-- 媒资错误代码 23000-- 用户中心错误代码 24000-- cms错误代码 25000-- 文件系统/public interface ResultCode {//操作是否成功,true为成功，false操作失败boolean success();//操作代码int code();//提示信息String message(); 从 ResultCode 中我们可以看出，我们约定了用户中心的错误代码使用 23000，所以我们定义的一些错误信息的代码就从 23000 开始计数。 Service 在学习服务中定义 service 方法，此方法远程请求课程管理服务、媒资管理服务获取课程学习地址。 package com.xuecheng.learning.service.impl;import com.netflix.discovery.converters.Auto;import com.xuecheng.framework.domain.course.TeachplanMediaPub;import com.xuecheng.framework.domain.learning.response.GetMediaResult;import com.xuecheng.framework.exception.ExceptionCast;import com.xuecheng.framework.model.response.CommonCode;import com.xuecheng.learning.client.CourseSearchClient;import com.xuecheng.learning.service.LearningService;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Service;@Servicepublic class LearningServiceImpl implements LearningService {@AutowiredCourseSearchClient courseSearchClient;/ 远程调用搜索服务获取已发布媒体信息中的url @param courseId 课程id @param teachplanId 媒体信息id @return/@Overridepublic GetMediaResult getMediaPlayUrl(String courseId, String teachplanId) {//校验学生权限,是否已付费等//远程调用搜索服务进行查询媒体信息TeachplanMediaPub mediaPub = courseSearchClient.getmedia(teachplanId);if(mediaPub == null) ExceptionCast.cast(CommonCode.FAIL);return new GetMediaResult(CommonCode.SUCCESS, mediaPub.getMediaUrl());} } Controller 调用 service 根据课程计划 id 查询视频播放地址： @RestController@RequestMapping("/learning/course")public class CourseLearningController implements CourseLearningControllerApi {@AutowiredLearningService learningService;@Override@GetMapping("/getmedia/{courseId}/{teachplanId}")public GetMediaResult getMediaPlayUrl(@PathVariable String courseId, @PathVariable String teachplanId) {//获取课程学习地址return learningService.getMedia(courseId, teachplanId);} } 测试 使用 swagger-ui 或postman 测试学习服务查询课程视频地址接口。 0x05 前端开发 需求分析 需要在学习中心前端页面需要完成如下功能： 1、进入课程学习页面需要带上 课程 Id参数及课程计划Id的参数，其中 课程 Id 参数必带，课程计划 Id 可以为空。 2、进入页面根据 课程 Id 取出该课程的课程计划显示在右侧。 3、进入页面后判断如果请求参数中有课程计划 Id 则播放该章节的视频。 4、进入页面后判断如果 课程计划id 为0则需要取出本课程第一个 课程计划的Id，并播放第一个课程计划的视频。 进入到模块 xc-ui-pc-leanring/src/module/course api方法 let sysConfig = require('@/../config/sysConfig')let apiUrl = sysConfig.xcApiUrlPre;/获取播放地址/export const get_media = (courseId,chapter) => {return http.requestGet(apiUrl+'/api/learning/course/getmedia/'+courseId+'/'+chapter);} 配置代理 在 Nginx 中的 ucenter.xuecheng.com 虚拟主机中配置 /api/learning/ 的路径转发，此url 请转发到学习服务。 学习服务upstream learning_server_pool{server 127.0.0.1:40600 weight=10;}学成网用户中心server {listen 80;server_name ucenter.xuecheng.com;个人中心location / {proxy_pass http://ucenter_server_pool;}后端搜索服务location /openapi/search/ {proxy_pass http://search_server_pool/search/; }学习服务location ^~ /api/learning/ {proxy_pass http://learning_server_pool/learning/;} } 视频播放页面 1、如果传入的课程计划id为0则取出第一个课程计划id 在 created 钩子方法中完成 created(){//当前请求的urlthis.url = window.location//课程idthis.courseId = this.$route.params.courseId//章节idthis.chapter = this.$route.params.chapter//查询课程信息systemApi.course_view(this.courseId).then((view_course)=>{if(!view_course || !view_course[this.courseId]){this.$message.error("获取课程信息失败，请重新进入此页面！")return ;}let courseInfo = view_course[this.courseId]console.log(courseInfo)this.coursename = courseInfo.nameif(courseInfo.teachplan){console.log("准备开始播放视频")let teachplan = JSON.parse(courseInfo.teachplan);this.teachplanList = teachplan.children;//开始学习if(this.chapter == "0" || !this.chapter){//取出第一个教学计划this.chapter = this.getFirstTeachplan();console.log("第一个教学计划id为 ",this.chapter);this.study(this.chapter);}else{this.study(this.chapter);} }})}, 取出第一个章节 id，用户未输入课程计划 id 或者输入为 0 时，播放第一个。 //取出第一个章节getFirstTeachplan(){for(var i=0;i<this.teachplanList.length;i++){let firstTeachplan = this.teachplanList[i];//如果当前children存在，则取出第一个返回if(firstTeachplan.children && firstTeachplan.children.length>0){let secondTeachplan = firstTeachplan.children[0];return secondTeachplan.id;} }return ;}, 开始学习： //开始学习study(chapter){// 获取播放地址courseApi.get_media(this.courseId,chapter).then((res)=>{if(res.success){let fileUrl = sysConfig.videoUrl + res.fileUrl//播放视频this.playvideo(fileUrl)}else if(res.message){this.$message.error(res.message)}else{this.$message.error("播放视频失败，请刷新页面重试")} }).catch(res=>{this.$message.error("播放视频失败，请刷新页面重试")});}, 2、点击右侧课程章节切换播放 在原有代码基础上添加 click 事件，点击调用开始学习方法（study）。 <li v‐if="teachplan_first.children!=null" v‐for="(teachplan_second, index) inteachplan_first.children"><i class="glyphicon glyphicon‐check"></i><a :href="url" @click="study(teachplan_second.id)">{ {teachplan_second.pname} }</a></li> 3、地址栏路由url变更 这里需要注意一个问题，在用户点击课程章节切换播放时，地址栏的 url 也应该同步改变为当前所选择的课程计划 id 4、在线学习按钮 将 learnstatus 默认更改为 1，这样就能显示出马上学习的按钮，方便我们后续的集成测试。 文件路径为 xc-ui-pc-static-portal/include/course_detail_dynamic.html 部分代码块如下 <script>var body= new Vue({ //创建一个Vue的实例el: "body", //挂载点是id="app"的地方data: {editLoading: false,title:'测试',courseId:'',charge:'',//203001免费,203002收费learnstatus: 1 ,//课程状态，1：马上学习，2：立即报名、3：立即购买course:{},companyId:'template',company_stat:[],course_stat:{"s601001":"","s601002":"","s601003":""} }, 简单的测试 访问在线学习页面：http://ucenter.xuecheng.com//learning/课程id/课程计划id 通过 url 传入两个参数：课程id 和 课程计划id 如果没有课程计划则传入0 测试项目如下： 1、传入正确的课程id、课程计划id，自动播放本章节的视频 2、传入正确的课程id、课程计划id传入0，自动播放第一个视频 3、传入错误的课程id 或 课程计划id，提示错误信息。 4、通过右侧章节目录切换章节及播放视频。 访问： http://ucenter.xuecheng.com//learning/4028e58161bcf7f40161bcf8b77c0000/4028e58161bd18ea0161bd1f73190008 传入正确的课程id、课程计划id，自动播放本章节的视频 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Ef0xxym7-1595567273153)(https://qnoss.codeyee.com/20200704_15/image17)] 传入正确的课程id、课程计划id传入0，自动播放第一个视频 访问 http://ucenter.xuecheng.com//learning/4028e58161bcf7f40161bcf8b77c0000/0 识别出第一个课程计划的 id 需要注意的是这里的 chapter 参数是我自己在 study 函数里加上去的，可以忽略。 传入错误的课程id或课程计划id，提示错误信息。 通过右侧章节目录切换章节及播放视频。 点击章节即可播放，但是点击制定章节后 url 没有发生改变，这个问题暂时还没有解决，关注笔记后面的内容。 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-TOGdxwb4-1595567273158)(https://qnoss.codeyee.com/20200704_15/image20)] 完整的测试 准备工作 启动 RabbitMQ，启动 Logstash、ElasticSearch 建议把所有后端服务都开起来 启动 前端静态门户、启动 nginx 、启动课程管理前端 我们整理一下测试的流程 上传两个媒资视频文件，用于测试 进入到课程管理，为课程计划选择媒资信息 发布课程，等待 logstash 将数据采集到 ElasticSearch 的索引库中 进入学成网主页，点击课程，进入到搜索门户页面 搜索课程，进入到课程详情页面 点击开始学习，进入到课程学习页面，选择课程计划中的一个章节进行学习。 1、上传文件 首先我们使用之前开发的媒资管理模块，上传两个视频文件用于测试。 第一个文件上传成功 一些问题 在上传第二个文件时，发生了错误，我们来检查一下问题出在了哪里 在媒体服务的控制台中可以看到，在 mergeChunks 方法在校验文件 md5 时候抛出了异常 我们在 MD5 校验这里打个断点，重新上传文件，分析一下问题所在。 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-OpEMZGI8-1595567273166)(https://qnoss.codeyee.com/20200704_15/image23)] 单步调试后发现，合并文件后的MD5值与用户上传的源文件值不相等 方案1：删除本地分块文件重新尝试上传 考虑到可能是在用户上传完 视频的分块文件时发生了一些问题，导致合并文件后与源文件的大小不等，导致MD5也不相同，这里我们把这个视频上传到本地的文件全部删除，在媒资上传页面重新上传文件。 对比所有分块文件的字节大小和本地源文件的大小，完全是相等的 删除所有文件后重新上传，md5值还是不等，考虑从调试一下文件合并的代码。 方案2：检查前端提交的MD5值是否正确 在查阅是否有其他的MD5值获取方案时，发现了一个使用 windows 本地命令获取文件MD5值的方法 certutil -hashfile .\19-在线学习接口-集成测试.avi md5 惊奇的发现，TM的原来是前端那边转换的MD5值不正确，后端这边是没有问题的。 从前面的图可以看出，本地和后端转换的都是以一个 f6f0 开头的MD5值 那么问题就出现在前端了，还需要花一些时间去分析一下，这里暂时就先告一段落，因为上传了几个文件测试中只有这一个文件出现了问题。 2、为课程计划选择媒资信息 进入到一个课程的管理页面 http://localhost:12000//course/manage/baseinfo/4028e58161bcf7f40161bcf8b77c0000 将刚才我们上传的媒资文件的信息和课程计划绑定 选择效果如下 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-epKaqzCD-1595567273178)(https://qnoss.codeyee.com/20200704_15/image29)] 2、发布课程，等待 logstash 从 course_pub 以及 teachplan_media_pub 表中采集数据到 ElasticSearch 当中 发布成功后，我们可以从 teachplan_media_pub 表中看到刚才我们发布的媒资信息 再观察 Logstash 的控制台，发现两个 Logstash 的实例都对更新的课程发布信息进行了采集 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-hTUve2ik-1595567273183)(https://qnoss.codeyee.com/20200704_15/image32)] 3、前端门户测试 打开我们的门户主站 http://www.xuecheng.com/ [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4wZe9R84-1595567273185)(https://qnoss.codeyee.com/20200704_15/image33)] 点击导航栏的课程，进入到我们的搜索门户页面 如果无法进入到搜索门户，请检查你的 xc-ui-pc-portal 前端工程是否已经启动 进入到搜索门户后，可以看到一些初始化时搜索的课程数据，默认是搜索第一页的数据，每页2个课程。 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-BJ1AKoJb-1595567273187)(https://qnoss.codeyee.com/20200704_15/image34)] 我们可以测试搜索一下前面我们选择媒资信息时所用的课程 点击课程，进入到课程详情页面，然后再点击开始学习。 点击马上学习后，会进入到该课程的在线学习页面，默认自动播放我们第一个课程计划中的视频。 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-tcuLWnf2-1595567273193)(https://qnoss.codeyee.com/20200704_15/image37)] 我们可以在右侧的目录中选择第二个课程计划，会自动播放所选的课程计划所对应的媒资视频播放地址，该 播放地址正是我们刚才通过 Logstash 自动采集到 ElasticSearch 的索引信息，效果图如下 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Cvi9Dr0Y-1595567273195)(https://qnoss.codeyee.com/20200704_15/image38)] 四、待完善的一些功能 课程发布前，校验课程计划里面是否包含二级课程计划 课程发布前，校验课程计划信息里面是否全部包含媒资信息 删除媒资信息，并且同步删除ES中的索引 在获取该课程的播放地址时校验用户的合法、 在线学习页面，点击右侧目录中的课程计划同时改变url中的课程计划地址 视频文件 19-在线学习接口-集成测试.avi 前端上传时提交的MD5值不正确 😁 认识作者 作者：👦 LCyee ，全干型代码🐕 自建博客：https://www.codeyee.com 记录学习以及项目开发过程中的笔记与心得，记录认知迭代的过程，分享想法与观点。 CSDN 博客：https://blog.csdn.net/codeyee 记录和分享一些开发过程中遇到的问题以及解决的思路。 欢迎加入微服务练习生的队伍，一起交流项目学习过程中的一些问题、分享学习心得等，不定期组织一起刷题、刷项目，共同见证成长。 本篇文章为转载内容。原文链接：https://blog.csdn.net/codeyee/article/details/107558901。 该文由互联网用户投稿提供，文中观点代表作者本人意见，并不代表本站的立场。 作为信息平台，本站仅提供文章转载服务，并不拥有其所有权，也不对文章内容的真实性、准确性和合法性承担责任。 如发现本文存在侵权、违法、违规或事实不符的情况，请及时联系我们，我们将第一时间进行核实并删除相应内容。

.../> 5、按“启动”按钮启动服务。 </div> <div style="width: 70%; text-align: left;"> <b>二、按本页的“打印”按钮开始打印。</b><br/> </div><br/> <input type="button" id="btnPrint" value="打印" /><br/><br/> <div style="width: 70%; text-align: left; font-size: 12px;"> 由于JavaScript在不同域名下访问会出现由来已久的跨域问题，所以正式部署到服务器使用时，要解决跨域问题。<br/> 对于IE8以上版本浏览器，只需增加一个reponse头：Access-Control-Allow-Origin即可，而对于php、jsp、asp/aspx等动态语言而言，增加一个response头是非常简单的事，例如：<br/> <b>在php：</b><br/><span style="color: red;"> <?php <br/> header("Access-Control-Allow-Origin: ");<br/> ?><br/> </span> <b>在jsp：</b><br/><span style="color: red;"> <% <br/> response.setHeader("Access-Control-Allow-Origin", ""); <br/> %><br/> </span> <b>在asp.net中：</b><br/><span style="color: red;"> Response.AppendHeader("Access-Control-Allow-Origin", ""); </span>，<br/>其他语言里，大家请自行搜索“ajax跨域”。而对于IE8以下的浏览器，大家可以自行搜索“IE6+Ajax+跨域”寻找解决办法吧，也可以联系我们帮助。 </div> </div> </div> <div id="output"></div> </body> <!-- 引入模式2所需的javascript支持库 --> <script type="text/javascript" src="cfprint_mode2.min.js" charset="UTF-8"></script> <!-- 构造报表数据 --> <script type="text/javascript"> var _reportData = '{"template":"waybill.fr3","ver":3, "Tables":[ {"Name":"Table1", "Cols":[{"type":"str","size":255,"name":"HAWB","required":false},{"type":"int","size":0,"name":"NO","required":false},{"type":"float","size":0,"name":"报关公司面单号","required":false},{"type":"integer","size":0,"name":"公司内部单号","required":false},{"type":"str","size":255,"name":"发件人","required":false},{"type":"str","size":255,"name":"发件人地址","required":false},{"type":"str","size":255,"name":"发件人电话","required":false},{"type":"str","size":255,"name":"发货国家","required":false},{"type":"str","size":255,"name":"收件人","required":false},{"type":"str","size":255,"name":"收件人地址","required":false},{"type":"str","size":255,"name":"收件人电话","required":false},{"type":"str","size":255,"name":"收货人证件号码","required":false},{"type":"str","size":255,"name":"收货省份","required":false},{"type":"float","size":0,"name":"总计费重量","required":false},{"type":"int","size":0,"name":"总件数","required":false},{"type":"float","size":0,"name":"申报总价（CNY）","required":false},{"type":"float","size":0,"name":"申报总价（JPY）","required":false},{"type":"int","size":0,"name":"件数1","required":false},{"type":"str","size":255,"name":"品名1","required":false},{"type":"float","size":0,"name":"单价1（JPY）","required":false},{"type":"str","size":255,"name":"单位1","required":false},{"type":"float","size":0,"name":"申报总价1（CNY）","required":false},{"type":"float","size":0,"name":"申报总价1（JPY）","required":false},{"type":"int","size":0,"name":"件数2","required":false},{"type":"str","size":255,"name":"品名2","required":false},{"type":"float","size":0,"name":"单价2（JPY）","required":false},{"type":"str","size":255,"name":"单位2","required":false},{"type":"float","size":0,"name":"申报总价2（CNY）","required":false},{"type":"float","size":0,"name":"申报总价2（JPY）","required":false},{"type":"int","size":0,"name":"件数3","required":false},{"type":"str","size":255,"name":"品名3","required":false},{"type":"float","size":0,"name":"单价3（JPY）","required":false},{"type":"str","size":255,"name":"单位3","required":false},{"type":"float","size":0,"name":"申报总价3（CNY）","required":false},{"type":"float","size":0,"name":"申报总价3（JPY）","required":false},{"type":"int","size":0,"name":"件数4","required":false},{"type":"str","size":255,"name":"品名4","required":false},{"type":"float","size":0,"name":"单价4（JPY）","required":false},{"type":"str","size":255,"name":"单位4","required":false},{"type":"float","size":0,"name":"申报总价4（CNY）","required":false},{"type":"float","size":0,"name":"申报总价4（JPY）","required":false},{"type":"int","size":0,"name":"件数5","required":false},{"type":"str","size":255,"name":"品名5","required":false},{"type":"float","size":0,"name":"单价5（JPY）","required":false},{"type":"str","size":255,"name":"单位5","required":false},{"type":"float","size":0,"name":"申报总价5（CNY）","required":false},{"type":"float","size":0,"name":"申报总价5（JPY）","required":false},{"type":"str","size":255,"name":"参考号","required":false},{"type":"AutoInc","size":0,"name":"ID","required":false}],"Data":[{"公司内部单号":730293,"发货国家":"日本","单价1（JPY）":null,"申报总价2（JPY）":null,"单价4（JPY）":null,"申报总价2（CNY）":null,"申报总价5（JPY）":null,"报关公司面单号":200303900791,"申报总价5（CNY）":null,"收货人证件号码":null,"申报总价1（JPY）":null,"单价3（JPY）":null,"申报总价1（CNY）":null,"申报总价4（JPY）":null,"申报总价4（CNY）":null,"收件人电话":"182-1758-9999","收件人地址":"上海市闵行区虹梅南路1660弄蔷薇八村139号502室","HAWB":"860014010055","发件人电话":"03-3684-9999","发件人地址":" 1-1-13,Kameido,Koto-ku,Tokyo","NO":3,"ID":3,"单价2（JPY）":null,"申报总价3（JPY）":null,"单价5（JPY）":null,"申报总价3（CNY）":null,"收货省份":null,"申报总价（JPY）":null,"申报总价（CNY）":null,"总计费重量":3.20,"收件人":"张三丰2","总件数":13,"品名5":null,"品名4":null,"品名3":null,"品名2":null,"品名1":"纸尿片","参考号":null,"发件人":"NAKAGAWA SUMIRE 2","单位5":null,"单位4":null,"单位3":null,"单位2":null,"单位1":null,"件数5":null,"件数4":null,"件数3":3,"件数2":null,"件数1":10},{"公司内部单号":730291,"发货国家":"日本","单价1（JPY）":null,"申报总价2（JPY）":null,"单价4（JPY）":null,"申报总价2（CNY）":null,"申报总价5（JPY）":null,"报关公司面单号":200303900789,"申报总价5（CNY）":null,"收货人证件号码":null,"申报总价1（JPY）":null,"单价3（JPY）":null,"申报总价1（CNY）":null,"申报总价4（JPY）":null,"申报总价4（CNY）":null,"收件人电话":"182-1758-9999","收件人地址":"上海市闵行区虹梅南路1660弄蔷薇八村139号502室","HAWB":"860014010035","发件人电话":"03-3684-9999","发件人地址":" 1-1-13,Kameido,Koto-ku,Tokyo","NO":1,"ID":1,"单价2（JPY）":null,"申报总价3（JPY）":null,"单价5（JPY）":null,"申报总价3（CNY）":null,"收货省份":null,"申报总价（JPY）":null,"申报总价（CNY）":null,"总计费重量":3.20,"收件人":"张三丰","总件数":13,"品名5":null,"品名4":null,"品名3":null,"品名2":null,"品名1":"纸尿片","参考号":null,"发件人":"NAKAGAWA SUMIRE","单位5":null,"单位4":null,"单位3":null,"单位2":null,"单位1":null,"件数5":null,"件数4":null,"件数3":3,"件数2":null,"件数1":10},{"公司内部单号":730292,"发货国家":"日本","单价1（JPY）":null,"申报总价2（JPY）":null,"单价4（JPY）":null,"申报总价2（CNY）":null,"申报总价5（JPY）":null,"报关公司面单号":200303900790,"申报总价5（CNY）":null,"收货人证件号码":null,"申报总价1（JPY）":null,"单价3（JPY）":null,"申报总价1（CNY）":null,"申报总价4（JPY）":null,"申报总价4（CNY）":null,"收件人电话":"182-1758-9999","收件人地址":"上海市闵行区虹梅南路1660弄蔷薇八村139号502室","HAWB":"860014010045","发件人电话":"03-3684-9999","发件人地址":" 1-1-13,Kameido,Koto-ku,Tokyo","NO":2,"ID":2,"单价2（JPY）":null,"申报总价3（JPY）":null,"单价5（JPY）":null,"申报总价3（CNY）":null,"收货省份":null,"申报总价（JPY）":null,"申报总价（CNY）":null,"总计费重量":3.20,"收件人":"张无忌","总件数":13,"品名5":null,"品名4":null,"品名3":null,"品名2":null,"品名1":"纸尿片","参考号":null,"发件人":"NAKAGAWA SUMIRE 1","单位5":null,"单位4":null,"单位3":null,"单位2":null,"单位1":null,"件数5":null,"件数4":null,"件数3":3,"件数2":null,"件数1":10}]}]}'; if(window.console) console.log("reportData = " + _reportData); </script> <!-- 设置服务器参数 --> <script language="javascript" type="text/javascript"> var cfprint_addr = "127.0.0.1"; //打印服务器监听地址 var cfprint_port = 54321; //打印服务器监听端口 var _url = "http://"+cfprint_addr+":"+cfprint_port; </script> <!-- 编写回调函数用以处理服务器返回的数据 --> <script type="text/javascript"> / 参数： readyState: XMLHttpRequest的状态 httpStatus: 服务端返回的http状态 responseText: 服务端返回的内容 / var callbackSuccess = function(readyState, httpStatus, responseText){ if (httpStatus === 200) { //{"result": 1, "message": "打印完成"} var response = CFPrint.parseJSON(responseText); alert(response.message+", 状态码["+response.result+"]"); }else{ alert('打印失败，HTTP状态代码是：'+httpStatus); } } / 参数： message: 错误信息 / var callbackFailed = function(message){ alert('发送打印任务出错: ' + message); } </script> <!-- 调用发送打印请求功能 --> <script type="text/javascript"> (function(){ document.getElementById("btnPrint").onclick = function() { CFPrint.outputid = "output"; //指定调试信息输出div的id CFPrint.SendRequest(_url, _reportData, callbackSuccess, callbackFailed); //发送打印请求 }; })(); </script> </html> 六、模板设计器（重要！重要！！，好多朋友都找不到设计器入口） 在主界面上，双击右下角的“设计”两个字，即可打开模板设计工具箱，在工具箱有三个按钮和一个大文本框。三个按钮的作用分别是： 设计：以大文本框中的json数据为数据源，打开模板设计器窗口； 预览：以大文本框中的json数据为数据源，预览当前所用模板的打印效果； 打印：以大文本框中的json数据为数据源，向打印机输出当前所用模板生成的报表； 以后将会有详细的模板设计教程发布，如果您遇到紧急的难题，请向作者咨询。 本篇文章为转载内容。原文链接：https://blog.csdn.net/chensongmol/article/details/76087600。 该文由互联网用户投稿提供，文中观点代表作者本人意见，并不代表本站的立场。 作为信息平台，本站仅提供文章转载服务，并不拥有其所有权，也不对文章内容的真实性、准确性和合法性承担责任。 如发现本文存在侵权、违法、违规或事实不符的情况，请及时联系我们，我们将第一时间进行核实并删除相应内容。

...比如一个网站，它可以启动多个线程去各大电商网站，比如淘宝，京东，收集某些或某一个商品的价格 最后，将获取的数据进行整合封装 最终，客户就可以通过此网站，获取某类商品在各网站的价格信息 / 假设你能够提供一个服务 这个服务查询各大电商网站同一类产品的价格并汇总展示 @author 马士兵 http://mashibing.com/import java.io.IOException;import java.util.Random;import java.util.concurrent.CompletableFuture;import java.util.concurrent.ExecutionException;import java.util.concurrent.TimeUnit;public class T06_01_CompletableFuture {public static void main(String[] args) throws ExecutionException, InterruptedException {long start, end;/start = System.currentTimeMillis();priceOfTM();priceOfTB();priceOfJD();end = System.currentTimeMillis();System.out.println("use serial method call! " + (end - start));/start = System.currentTimeMillis();CompletableFuture<Double> futureTM = CompletableFuture.supplyAsync(()->priceOfTM());CompletableFuture<Double> futureTB = CompletableFuture.supplyAsync(()->priceOfTB());CompletableFuture<Double> futureJD = CompletableFuture.supplyAsync(()->priceOfJD());CompletableFuture.allOf(futureTM, futureTB, futureJD).join();//当所有结果集都获取到，才汇总阻塞CompletableFuture.supplyAsync(()->priceOfTM()).thenApply(String::valueOf).thenApply(str-> "price " + str).thenAccept(System.out::println);end = System.currentTimeMillis();System.out.println("use completable future! " + (end - start));try {System.in.read();} catch (IOException e) {e.printStackTrace();} }private static double priceOfTM() {delay();return 1.00;}private static double priceOfTB() {delay();return 2.00;}private static double priceOfJD() {delay();return 3.00;}/private static double priceOfAmazon() {delay();throw new RuntimeException("product not exist!");}/private static void delay() {int time = new Random().nextInt(500);try {TimeUnit.MILLISECONDS.sleep(time);} catch (InterruptedException e) {e.printStackTrace();}System.out.printf("After %s sleep!\n", time);} } 4、TPE型线程池1：ThreadPoolExecutor 原理及其参数 线程池由两个集合组成，一个集合存储线程，一个集合存储任务 存储线程：可以规定大小，最多可以有多少个，以及指定核心线程数量（不会被回收） 任务队列：存储任务 细节：初始线程池没有线程，当有一个任务来，线程池起一个线程，又有一个任务来，再起一个线程，直到达到核心线程数量 核心线程数量达到时，新来的任务将存储到任务队列中等待核心线程处理完成，直到任务队列也满了 当任务队列满了，此时再次启动一个线程（非核心线程，一旦空闲，达到指定时间将会消失），直到达到线程最大数量 当线程容器和任务容器都满了，又来了线程，将会执行拒绝策略 上面的细节涉及的所有步骤内容，均由创建线程池的参数执行 下面是ThreadPoolExecutor构造方法参数的源码注释 / 用给定的初始值，创建一个新的线程池 @param corePoolSize 核心线程数量 @param maximumPoolSize 最大线程数量 @param keepAliveTime 当线程数大于核心线程数量时，空闲的线程可生存的时间 @param unit 时间单位 @param workQueue 任务队列，只能包含由execute提交的Runnable任务 @param threadFactory 工厂，用于创建线程给线程池调度的工厂，可以自定义 @param handler 拒绝策略(可以自定义，JDK默认提供4种)，当线程边界和队列容量已经满了，新来线程被阻塞时使用的处理程序/public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) JDK提供的4种拒绝策略，不常用，一般都是自己定义拒绝策略 Abort：抛异常 Discard：扔掉，不抛异常 DiscardOldest：扔掉排队时间最久的（将队列中排队时间最久的扔掉，然后让新来的进来） CallerRuns：调用者处理任务（谁通过execute方法提交任务，谁处理） ThreadPoolExecutor继承关系 继承关系：ThreadPoolExecutor->AbstractExectorService类->ExectorService接口->Exector接口 Executors（注意这后面有s） 它可以说是线程池工厂类，我们一般通过它创建线程池，并且它为我们封装了线程 看看下面创建线程池，哪里用到了它 使用实例 import java.io.IOException;import java.util.concurrent.;public class T05_00_HelloThreadPool {static class Task implements Runnable {private int i;public Task(int i) {this.i = i;}@Overridepublic void run() {System.out.println(Thread.currentThread().getName() + " Task " + i);try {System.in.read();} catch (IOException e) {e.printStackTrace();} }@Overridepublic String toString() {return "Task{" +"i=" + i +'}';} }public static void main(String[] args) {ThreadPoolExecutor tpe = new ThreadPoolExecutor(2, 4,60, TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(4),Executors.defaultThreadFactory(),new ThreadPoolExecutor.CallerRunsPolicy());//创建线程池，核心2个，最大4个，空闲线程存活时间60s，任务队列容量4，使用默认线程工程，创建线程。拒绝策略是JDK提供的for (int i = 0; i < 8; i++) {tpe.execute(new Task(i));//供提交8次任务}System.out.println(tpe.getQueue());//查看任务队列tpe.execute(new Task(100));//提交新的任务System.out.println(tpe.getQueue());tpe.shutdown();//关闭线程池} } 5、TPE型线程池2：SingleThreadPool 单例线程池(只有一个线程) 为什么有单例线程池 有任务队列，有线程池管理机制 Executors（注意这后面有s） 它可以说是线程池工厂类，我们一般通过它创建线程池，并且它为我们封装了线程 看看下面哪里用到了它 /创建单例线程池，扔5个任务进去，查看输出结果，看看有几个线程执行任务/import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;public class T07_SingleThreadPool {public static void main(String[] args) {ExecutorService service = Executors.newSingleThreadExecutor();for(int i=0; i<5; i++) {final int j = i;service.execute(()->{System.out.println(j + " " + Thread.currentThread().getName());});} }} 6、TPE型线程池3：CachedPool 缓存，存储线程池 此线程池没有核心线程，来一个任务启动一个线程（最多Integer.MaxValue，不会放在任务队列，因为任务队列容量为0），每个线程空闲后，只能活60s 实例 import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;public class T07_SingleThreadPool {public static void main(String[] args) {ExecutorService service = Executors.newSingleThreadExecutor();//通过Executors获取池子for(int i=0; i<5; i++) {final int j = i;service.execute(()->{//提交任务System.out.println(j + " " + Thread.currentThread().getName());});}service.shutdown();} } 7、TPE型线程池4：FixedThreadPool 固定线程池 此线次池，用于创建一个固定线程数量的线程池，不会回收 实例 import java.util.ArrayList;import java.util.List;import java.util.concurrent.Callable;import java.util.concurrent.ExecutionException;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Future;public class T09_FixedThreadPool {public static void main(String[] args) throws InterruptedException, ExecutionException {//并发执行long start = System.currentTimeMillis();getPrime(1, 200000); long end = System.currentTimeMillis();System.out.println(end - start);//输出并发执行耗费时间final int cpuCoreNum = 4;//并行执行ExecutorService service = Executors.newFixedThreadPool(cpuCoreNum);MyTask t1 = new MyTask(1, 80000); //1-5 5-10 10-15 15-20MyTask t2 = new MyTask(80001, 130000);MyTask t3 = new MyTask(130001, 170000);MyTask t4 = new MyTask(170001, 200000);Future<List<Integer>> f1 = service.submit(t1);Future<List<Integer>> f2 = service.submit(t2);Future<List<Integer>> f3 = service.submit(t3);Future<List<Integer>> f4 = service.submit(t4);start = System.currentTimeMillis();f1.get();f2.get();f3.get();f4.get();end = System.currentTimeMillis();System.out.println(end - start);//输出并行耗费时间}static class MyTask implements Callable<List<Integer>> {int startPos, endPos;MyTask(int s, int e) {this.startPos = s;this.endPos = e;}@Overridepublic List<Integer> call() throws Exception {List<Integer> r = getPrime(startPos, endPos);return r;} }static boolean isPrime(int num) {for(int i=2; i<=num/2; i++) {if(num % i == 0) return false;}return true;}static List<Integer> getPrime(int start, int end) {List<Integer> results = new ArrayList<>();for(int i=start; i<=end; i++) {if(isPrime(i)) results.add(i);}return results;} } 8、TPE型线程池5：ScheduledPool 预定，延时线程池 根据延时时间（隔多长时间后运行），排序，哪个线程先执行,用户只需要指定核心线程数量 此线程池返回的池对象，和提交任务方法都不一样，比较涉及到时间 import java.util.Random;import java.util.concurrent.Executors;import java.util.concurrent.ScheduledExecutorService;import java.util.concurrent.TimeUnit;public class T10_ScheduledPool {public static void main(String[] args) {ScheduledExecutorService service = Executors.newScheduledThreadPool(4);service.scheduleAtFixedRate(()->{//提交延时任务try {TimeUnit.MILLISECONDS.sleep(new Random().nextInt(1000));} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName());}, 0, 500, TimeUnit.MILLISECONDS);//指定延时时间和单位，第一个任务延时0毫秒，之后的任务，延时500毫秒} } 9、手写拒绝策略小例子 import java.util.concurrent.;public class T14_MyRejectedHandler {public static void main(String[] args) {ExecutorService service = new ThreadPoolExecutor(4, 4,0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(6),Executors.defaultThreadFactory(),new MyHandler());//将手写拒绝策略传入}static class MyHandler implements RejectedExecutionHandler {//1、继承RejectedExecutionHandler@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {//2、重写方法//log("r rejected")//伪代码，表示通过log4j.log()报一下日志，拒绝的时间，线程名//save r kafka mysql redis//可以尝试保存队列//try 3 times //可以尝试几次，比如3次，重新去抢队列，3次还不行就丢弃if(executor.getQueue().size() < 10000) {//尝试条件，如果size>10000了,就执行拒绝策略//try put again();//如果小于10000，尝试将其放到队列中} }} } 10、ForkJoinPool线程池1：ForkJoinPool 前面我们讲过线程分为两大类，TPE和FJP ForkJoinPool（分解汇总任务(将任务细化，最后汇总结果)，少量线程执行多个任务（子任务，TPE做不到先执行子任务），CPU密集型） 适合将大任务切分成多个小任务运行 两个方法，fork()：分子任务，将子任务分配到线程池中 join()：当前任务的计算结果，如果有子任务，等子任务结果返回后再汇总 下面实例实现，一百万个随机数求和，由两种方法实现，一种ForkJoinPool分任务并行，一种使用单线程做 import java.io.IOException;import java.util.Arrays;import java.util.Random;import java.util.concurrent.ForkJoinPool;import java.util.concurrent.RecursiveAction;import java.util.concurrent.RecursiveTask;public class T12_ForkJoinPool {//1000000个随机数求和static int[] nums = new int[1000000];//一堆数static final int MAX_NUM = 50000;//分任务时，每个任务的操作量不能多于50000个，否则就继续细分static Random r = new Random();//使用随机数将数组初始化static {for(int i=0; i<nums.length; i++) {nums[i] = r.nextInt(100);}System.out.println("---" + Arrays.stream(nums).sum()); //stream api 单线程就这么做，一个一个加}//分任务，需要继承，可以继承RecursiveAction(不需要返回值，一般用在不需要返回值的场景)或//RecursiveTask(需要返回值，我们用这个，因为我们需要最后获取求和结果)两个更好实现的类，//他俩继承与ForkJoinTaskstatic class AddTaskRet extends RecursiveTask<Long> {private static final long serialVersionUID = 1L;int start, end;AddTaskRet(int s, int e) {start = s;end = e;}@Overrideprotected Long compute() {if(end-start <= MAX_NUM) {//如果任务操作数小于规定的最大操作数，就进行运算，long sum = 0L;for(int i=start; i<end; i++) sum += nums[i];return sum;//返回结果} //如果分配的操作数大于规定，就继续细分（简单的重中点分，两半）int middle = start + (end-start)/2;//获取中间值AddTaskRet subTask1 = new AddTaskRet(start, middle);//传入起始值和中间值，表示一个子任务AddTaskRet subTask2 = new AddTaskRet(middle, end);//中间值和结尾值，表示一个子任务subTask1.fork();//分任务subTask2.fork();//分任务return subTask1.join() + subTask2.join();//最后返回结果汇总} }public static void main(String[] args) throws IOException {/ForkJoinPool fjp = new ForkJoinPool();AddTask task = new AddTask(0, nums.length);fjp.execute(task);/ForkJoinPool fjp = new ForkJoinPool();//创建线程池AddTaskRet task = new AddTaskRet(0, nums.length);//创建任务fjp.execute(task);//传入任务long result = task.join();//返回汇总结果System.out.println(result);//System.in.read();} } 11、ForkJoinPool线程池2：WorkStealingPool 任务偷取线程池 原来的线程池，都是有一个任务队列，而这个不同，它给每个线程都分配了一个任务队列 当某一个线程的任务队列没有任务，并且自己空闲，它就去其它线程的任务队列中偷任务，所以叫任务偷取线程池 细节：当线程自己从自己的任务队列拿任务时，不需要加锁，但是偷任务时，因为有两个线程，可能发生同步问题，需要加锁 此线程继承FJP 实例 import java.io.IOException;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.TimeUnit;public class T11_WorkStealingPool {public static void main(String[] args) throws IOException {ExecutorService service = Executors.newWorkStealingPool();System.out.println(Runtime.getRuntime().availableProcessors());service.execute(new R(1000));service.execute(new R(2000));service.execute(new R(2000));service.execute(new R(2000)); //daemonservice.execute(new R(2000));//由于产生的是精灵线程（守护线程、后台线程），主线程不阻塞的话，看不到输出System.in.read(); }static class R implements Runnable {int time;R(int t) {this.time = t;}@Overridepublic void run() {try {TimeUnit.MILLISECONDS.sleep(time);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(time + " " + Thread.currentThread().getName());} }} 12、流式API：ParallelStreamAPI 不懂的请参考：https://blog.csdn.net/grd_java/article/details/110265219 实例 import java.util.ArrayList;import java.util.List;import java.util.Random;public class T13_ParallelStreamAPI {public static void main(String[] args) {List<Integer> nums = new ArrayList<>();Random r = new Random();for(int i=0; i<10000; i++) nums.add(1000000 + r.nextInt(1000000));//System.out.println(nums);long start = System.currentTimeMillis();nums.forEach(v->isPrime(v));long end = System.currentTimeMillis();System.out.println(end - start);//使用parallel stream apistart = System.currentTimeMillis();nums.parallelStream().forEach(T13_ParallelStreamAPI::isPrime);//并行流，将任务切分成子任务执行end = System.currentTimeMillis();System.out.println(end - start);}static boolean isPrime(int num) {for(int i=2; i<=num/2; i++) {if(num % i == 0) return false;}return true;} } 13、总结 总结 Callable相当于一Runnable但是它有返回值 Future：存储执行完产生的结果 FutureTask 相当于Future+Runnable，既可以执行任务，又能获取任务执行的Future结果 CompletableFuture 可以多任务异步，并对多任务控制，整合任务结果，细化完美，比如可以一个任务完成就可以整合结果，也可以所有任务完成才整合结果 4、ThreadPoolExecutor源码解析 依然只讲重点，实际还需要大家按照上篇博客中看源码的方式来看 1、常用变量的解释 // 1. ctl，可以看做一个int类型的数字，高3位表示线程池状态，低29位表示worker数量private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));// 2. COUNT_BITS，Integer.SIZE为32，所以COUNT_BITS为29private static final int COUNT_BITS = Integer.SIZE - 3;// 3. CAPACITY，线程池允许的最大线程数。1左移29位，然后减1，即为 2^29 - 1private static final int CAPACITY = (1 << COUNT_BITS) - 1;// runState is stored in the high-order bits// 4. 线程池有5种状态，按大小排序如下：RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATEDprivate static final int RUNNING = -1 << COUNT_BITS;private static final int SHUTDOWN = 0 << COUNT_BITS;private static final int STOP = 1 << COUNT_BITS;private static final int TIDYING = 2 << COUNT_BITS;private static final int TERMINATED = 3 << COUNT_BITS;// Packing and unpacking ctl// 5. runStateOf()，获取线程池状态，通过按位与操作，低29位将全部变成0private static int runStateOf(int c) { return c & ~CAPACITY; }// 6. workerCountOf()，获取线程池worker数量，通过按位与操作，高3位将全部变成0private static int workerCountOf(int c) { return c & CAPACITY; }// 7. ctlOf()，根据线程池状态和线程池worker数量，生成ctl值private static int ctlOf(int rs, int wc) { return rs | wc; }/ Bit field accessors that don't require unpacking ctl. These depend on the bit layout and on workerCount being never negative./// 8. runStateLessThan()，线程池状态小于xxprivate static boolean runStateLessThan(int c, int s) {return c < s;}// 9. runStateAtLeast()，线程池状态大于等于xxprivate static boolean runStateAtLeast(int c, int s) {return c >= s;} 2、构造方法 public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {// 基本类型参数校验if (corePoolSize < 0 ||maximumPoolSize <= 0 ||maximumPoolSize < corePoolSize ||keepAliveTime < 0)throw new IllegalArgumentException();// 空指针校验if (workQueue == null || threadFactory == null || handler == null)throw new NullPointerException();this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;// 根据传入参数unit和keepAliveTime，将存活时间转换为纳秒存到变量keepAliveTime 中this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;} 3、提交执行task的过程 public void execute(Runnable command) {if (command == null)throw new NullPointerException();/ Proceed in 3 steps: 1. If fewer than corePoolSize threads are running, try to start a new thread with the given command as its first task. The call to addWorker atomically checks runState and workerCount, and so prevents false alarms that would add threads when it shouldn't, by returning false. 2. If a task can be successfully queued, then we still need to double-check whether we should have added a thread (because existing ones died since last checking) or that the pool shut down since entry into this method. So we recheck state and if necessary roll back the enqueuing if stopped, or start a new thread if there are none. 3. If we cannot queue task, then we try to add a new thread. If it fails, we know we are shut down or saturated and so reject the task./int c = ctl.get();// worker数量比核心线程数小，直接创建worker执行任务if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))return;c = ctl.get();}// worker数量超过核心线程数，任务直接进入队列if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();// 线程池状态不是RUNNING状态，说明执行过shutdown命令，需要对新加入的任务执行reject()操作。// 这儿为什么需要recheck，是因为任务入队列前后，线程池的状态可能会发生变化。if (! isRunning(recheck) && remove(command))reject(command);// 这儿为什么需要判断0值，主要是在线程池构造方法中，核心线程数允许为0else if (workerCountOf(recheck) == 0)addWorker(null, false);}// 如果线程池不是运行状态，或者任务进入队列失败，则尝试创建worker执行任务。// 这儿有3点需要注意：// 1. 线程池不是运行状态时，addWorker内部会判断线程池状态// 2. addWorker第2个参数表示是否创建核心线程// 3. addWorker返回false，则说明任务执行失败，需要执行reject操作else if (!addWorker(command, false))reject(command);} 4、addworker源码解析 private boolean addWorker(Runnable firstTask, boolean core) {retry:// 外层自旋for (;;) {int c = ctl.get();int rs = runStateOf(c);// 这个条件写得比较难懂，我对其进行了调整，和下面的条件等价// (rs > SHUTDOWN) || // (rs == SHUTDOWN && firstTask != null) || // (rs == SHUTDOWN && workQueue.isEmpty())// 1. 线程池状态大于SHUTDOWN时，直接返回false// 2. 线程池状态等于SHUTDOWN，且firstTask不为null，直接返回false// 3. 线程池状态等于SHUTDOWN，且队列为空，直接返回false// Check if queue empty only if necessary.if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;// 内层自旋for (;;) {int wc = workerCountOf(c);// worker数量超过容量，直接返回falseif (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;// 使用CAS的方式增加worker数量。// 若增加成功，则直接跳出外层循环进入到第二部分if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get(); // Re-read ctl// 线程池状态发生变化，对外层循环进行自旋if (runStateOf(c) != rs)continue retry;// 其他情况，直接内层循环进行自旋即可// else CAS failed due to workerCount change; retry inner loop} }boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {w = new Worker(firstTask);final Thread t = w.thread;if (t != null) {final ReentrantLock mainLock = this.mainLock;// worker的添加必须是串行的，因此需要加锁mainLock.lock();try {// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.// 这儿需要重新检查线程池状态int rs = runStateOf(ctl.get());if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {// worker已经调用过了start()方法，则不再创建workerif (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();// worker创建并添加到workers成功workers.add(w);// 更新largestPoolSize变量int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;} } finally {mainLock.unlock();}// 启动worker线程if (workerAdded) {t.start();workerStarted = true;} }} finally {// worker线程启动失败，说明线程池状态发生了变化（关闭操作被执行），需要进行shutdown相关操作if (! workerStarted)addWorkerFailed(w);}return workerStarted;} 5、线程池worker任务单元 private final class Workerextends AbstractQueuedSynchronizerimplements Runnable{/ This class will never be serialized, but we provide a serialVersionUID to suppress a javac warning./private static final long serialVersionUID = 6138294804551838833L;/ Thread this worker is running in. Null if factory fails. /final Thread thread;/ Initial task to run. Possibly null. /Runnable firstTask;/ Per-thread task counter /volatile long completedTasks;/ Creates with given first task and thread from ThreadFactory. @param firstTask the first task (null if none)/Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;// 这儿是Worker的关键所在，使用了线程工厂创建了一个线程。传入的参数为当前workerthis.thread = getThreadFactory().newThread(this);}/ Delegates main run loop to outer runWorker /public void run() {runWorker(this);}// 省略代码...} 6、核心线程执行逻辑-runworker final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;// 调用unlock()是为了让外部可以中断w.unlock(); // allow interrupts// 这个变量用于判断是否进入过自旋（while循环）boolean completedAbruptly = true;try {// 这儿是自旋// 1. 如果firstTask不为null，则执行firstTask；// 2. 如果firstTask为null，则调用getTask()从队列获取任务。// 3. 阻塞队列的特性就是：当队列为空时，当前线程会被阻塞等待while (task != null || (task = getTask()) != null) {// 这儿对worker进行加锁，是为了达到下面的目的// 1. 降低锁范围，提升性能// 2. 保证每个worker执行的任务是串行的w.lock();// If pool is stopping, ensure thread is interrupted;// if not, ensure thread is not interrupted. This// requires a recheck in second case to deal with// shutdownNow race while clearing interrupt// 如果线程池正在停止，则对当前线程进行中断操作if ((runStateAtLeast(ctl.get(), STOP) ||(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())wt.interrupt();// 执行任务，且在执行前后通过beforeExecute()和afterExecute()来扩展其功能。// 这两个方法在当前类里面为空实现。try {beforeExecute(wt, task);Throwable thrown = null;try {task.run();} catch (RuntimeException x) {thrown = x; throw x;} catch (Error x) {thrown = x; throw x;} catch (Throwable x) {thrown = x; throw new Error(x);} finally {afterExecute(task, thrown);} } finally {// 帮助gctask = null;// 已完成任务数加一 w.completedTasks++;w.unlock();} }completedAbruptly = false;} finally {// 自旋操作被退出，说明线程池正在结束processWorkerExit(w, completedAbruptly);} } 本篇文章为转载内容。原文链接：https://blog.csdn.net/grd_java/article/details/113116244。 该文由互联网用户投稿提供，文中观点代表作者本人意见，并不代表本站的立场。 作为信息平台，本站仅提供文章转载服务，并不拥有其所有权，也不对文章内容的真实性、准确性和合法性承担责任。 如发现本文存在侵权、违法、违规或事实不符的情况，请及时联系我们，我们将第一时间进行核实并删除相应内容。

优化PostgreSQL的网络连接性能：深入实践与探讨 1. 引言 在当今数据驱动的世界中，数据库作为信息存储和处理的核心组件，其性能直接影响着整个系统的响应速度和服务质量。PostgreSQL，这个牛气哄哄的开源关系型数据库系统，靠的就是它那坚若磐石的可靠性以及琳琅满目的功能，在江湖上赢得了响当当的好口碑，深受大家的喜爱和推崇。不过，当碰上那种用户挤爆服务器、数据量大到离谱的场景时，怎样把PostgreSQL这个数据库网络连接的速度给提上去，就成了我们不得不面对的一项重点挑战。本文将深入探讨这一主题，通过实际操作与代码示例来揭示优化策略。 2. 网络连接性能瓶颈分析 首先，我们需要理解影响PostgreSQL网络连接性能的主要因素，这包括但不限于： - 连接池管理：频繁地创建和销毁数据库连接会消耗大量资源。 - 网络延迟：物理距离、带宽限制以及TCP/IP协议本身的特性都可能导致网络延迟。 - 数据包大小和传输效率：如批量处理能力、压缩设置等。 3. 连接池优化（示例） 为解决连接频繁创建销毁的问题，我们可以借助连接池技术，例如使用PgBouncer或pgpool-II等第三方工具。下面是一个使用PgBouncer配置连接池的例子： ini [databases] mydb = host=127.0.0.1 port=5432 dbname=mydb user=myuser password=mypassword [pgbouncer] pool_mode = transaction max_client_conn = 100 default_pool_size = 20 上述配置中，PgBouncer以事务模式运行，最大允许100个客户端连接，并为每个数据库预设了20个连接池，从而有效地复用了数据库连接，降低了开销。 4. TCP/IP参数调优 PostgreSQL可以通过调整TCP/IP相关参数来改善网络性能。比如说，为了让连接不因为长时间没动静而断开，咱们可以试着调大tcp_keepalives_idle、tcp_keepalives_interval和tcp_keepalives_count这三个参数。这就像是给你的网络连接按个“心跳检测器”，时不时地检查一下，确保连接还活着，即使在传输数据的间隙也不会轻易掉线。修改postgresql.conf文件如下： conf tcp_keepalives_idle = 60 tcp_keepalives_interval = 15 tcp_keepalives_count = 5 这里表示如果60秒内没有数据传输，PostgreSQL将开始发送心跳包，每隔15秒发送一次，最多发送5次尝试维持连接。 5. 数据传输效率提升 5.1 批量处理 尽量减少SQL查询的次数，利用PostgreSQL的批量插入功能提高效率。例如，原来逐行插入的代码： sql INSERT INTO my_table (column1, column2) VALUES ('value1', 'value2'); INSERT INTO my_table (column1, column2) VALUES ('value3', 'value4'); ... 可以改为批量插入： sql INSERT INTO my_table (column1, column2) VALUES ('value1', 'value2'), ('value3', 'value4'), ... 5.2 数据压缩 PostgreSQL支持对客户端/服务器之间的数据进行压缩传输，通过设置client_min_messages和log_statement参数开启日志记录，观察并决定是否启用压缩。若网络带宽有限且数据量较大，可考虑开启压缩： conf client_min_messages = notice log_statement = 'all' Compression = on 6. 结论与思考 优化PostgreSQL的网络连接性能是一项涉及多方面的工作，需要我们根据具体应用场景和问题特点进行细致的分析与实践。要是我们能灵活运用连接池，巧妙调整个网络参数，再把数据传输策略优化得恰到好处，就能让PostgreSQL在网络环境下的表现嗖嗖提升，效果显著得很！在这个过程中，不断尝试、犯错、反思再改进，就像一次次打怪升级，这正是我们在追求超神表现的旅程中寻觅的乐趣源泉。

PostgreSQL系统配置错误：导致性能下降与故障发生的深层解析 1. 引言 PostgreSQL，作为一款功能强大、开源的关系型数据库管理系统，在全球范围内广受赞誉。不过呢，就像老话说的，“好马得配好鞍”，哪怕PostgreSQL这匹“骏马”有着超凡的性能和稳如磐石的稳定性，可一旦咱们给它配上不合适的“鞍子”，也就是配置出岔子或者系统闹点儿小情绪，那很可能就拖了它的后腿，影响性能，严重点儿还可能引发各种意想不到的问题。这篇文章咱们要接地气地聊聊，配置出岔子可能会带来的那些糟心影响，并且我还会手把手地带你瞧瞧实例代码，教你如何把配置调校得恰到好处，让这些问题通通远离咱们。 2. 配置失误对性能的影响 2.1 shared_buffers设置不合理 shared_buffers是PostgreSQL用于缓存数据的重要参数，其大小直接影响到数据库的查询性能。要是你把这数值设得过小，就等于是在让磁盘I/O忙个不停，频繁操作起来，就像个永不停歇的陀螺，会拖累整体性能，让系统跑得像只乌龟。反过来，如果你一不留神把数值调得过大，那就像是在内存里开辟了一大片空地却闲置不用，这就白白浪费了宝贵的内存资源，还会把其他系统进程挤得没地方住，人家也会闹情绪的。 postgresql -- 在postgresql.conf中调整shared_buffers值 shared_buffers = 4GB -- 假设服务器有足够内存支持此设置 2.2 work_mem不足 work_mem定义了每个SQL查询可以使用的内存量，对于复杂的排序、哈希操作等至关重要。过低的work_mem设定可能导致大量临时文件生成，进一步降低性能。 postgresql -- 调整work_mem大小 work_mem = 64MB -- 根据实际业务负载进行合理调整 3. 配置失误导致的故障案例 3.1 max_connections设置过高 max_connections参数限制了PostgreSQL同时接受的最大连接数。如果设置得过高，却没考虑服务器的实际承受能力，就像让一个普通人硬扛大铁锤，早晚得累垮。这样一来，系统资源就会被消耗殆尽，好比车票都被抢光了，新的连接请求就无法挤上这趟“网络列车”。最终，整个系统可能就要“罢工”瘫痪啦。 postgresql -- 不合理的高连接数设置示例 max_connections = 500 -- 若服务器硬件条件不足以支撑如此多的并发连接，则可能引发故障 3.2 日志设置不当造成磁盘空间耗尽 log_line_prefix、log_directory等日志相关参数设置不当，可能导致日志文件迅速增长，占用过多磁盘空间，进而引发数据库服务停止。 postgresql -- 错误的日志设置示例 log_line_prefix = '%t [%p]: ' -- 时间戳和进程ID前缀可能会使日志行变得冗长 log_directory = '/var/log/postgresql' -- 如果不加以定期清理，日志文件可能会撑满整个分区 4. 探讨与建议 面对PostgreSQL的系统配置问题，我们需要深入了解每个参数的含义以及它们在不同场景下的最佳实践。优化配置是一个持续的过程，需要结合业务特性和硬件资源来进行细致调优。 - 理解需求：首先，应了解业务特点，包括数据量大小、查询复杂度、并发访问量等因素。 - 监控分析：借助pg_stat_activity、pg_stat_bgwriter等视图监控数据库运行状态，结合如pgBadger、pg_top等工具分析性能瓶颈。 - 逐步调整：每次只更改一个参数，观察并评估效果，切忌盲目跟从网络上的推荐配置。 总结来说，PostgreSQL的强大性能背后，合理的配置是关键。要让咱们的数据库系统跑得溜又稳，像老黄牛一样可靠，给业务发展扎扎实实当好坚强后盾，那就必须把这些参数整得门儿清，调校得恰到好处才行。

...是理想的网站建设项目启动模板。 点我下载 文件大小：699.03 KB 您将下载一个资源包，该资源包内部文件的目录结构如下： 本网站提供模板下载功能，旨在帮助广大用户在工作学习中提升效率、节约时间。 本网站的下载内容来自于互联网。如您发现任何侵犯您权益的内容，请立即告知我们，我们将迅速响应并删除相关内容。 免责声明：站内所有资源仅供个人学习研究及参考之用，严禁将这些资源应用于商业场景。 若擅自商用导致的一切后果，由使用者承担责任。

...开发者能够轻松配置并启动多个独立且精确的倒计时实例，同时保持代码的简洁与高效。该插件兼顾实用性和视觉体验，使得网页中的倒计时元素既信息明确又富有吸引力，从而有效提升用户交互体验。 点我下载 文件大小：81.05 KB 您将下载一个JQuery插件资源包，该资源包内部文件的目录结构如下： 本网站提供JQuery插件下载功能，旨在帮助广大用户在工作学习中提升效率、节约时间。 本网站的下载内容来自于互联网。如您发现任何侵犯您权益的内容，请立即告知我们，我们将迅速响应并删除相关内容。 免责声明：站内所有资源仅供个人学习研究及参考之用，严禁将这些资源应用于商业场景。 若擅自商用导致的一切后果，由使用者承担责任。

...、限时抢购，还是项目启动等各类场景，都能提供生动直观且富有节奏感的时间提醒，有效提升了用户在网页交互过程中的体验度与参与感。这款插件不仅展示了技术与艺术的完美结合，更是在实用性与观赏性之间找到了绝佳平衡点。 点我下载 文件大小：72.80 KB 您将下载一个JQuery插件资源包，该资源包内部文件的目录结构如下： 本网站提供JQuery插件下载功能，旨在帮助广大用户在工作学习中提升效率、节约时间。 本网站的下载内容来自于互联网。如您发现任何侵犯您权益的内容，请立即告知我们，我们将迅速响应并删除相关内容。 免责声明：站内所有资源仅供个人学习研究及参考之用，严禁将这些资源应用于商业场景。 若擅自商用导致的一切后果，由使用者承担责任。

...误的密码时，该插件将启动其独特的动态反馈效果。首先，密码输入框会呈现生动的抖动动画，以此直观地提醒用户注意其输入可能存在错误；其次，错误消息将以醒目的红色高亮显示，明确指出问题所在。此插件兼顾了实用性和视觉吸引力，使得密码验证过程变得不再单调，有效增强了与用户的交互性。通过融入这样的微交互元素，不仅能够帮助用户快速识别并纠正密码输入错误，同时也为网页整体增添了一抹趣味和专业感，显著提升了网站或应用的整体品质感和用户体验。 点我下载 文件大小：46.05 KB 您将下载一个JQuery插件资源包，该资源包内部文件的目录结构如下： 本网站提供JQuery插件下载功能，旨在帮助广大用户在工作学习中提升效率、节约时间。 本网站的下载内容来自于互联网。如您发现任何侵犯您权益的内容，请立即告知我们，我们将迅速响应并删除相关内容。 免责声明：站内所有资源仅供个人学习研究及参考之用，严禁将这些资源应用于商业场景。 若擅自商用导致的一切后果，由使用者承担责任。

...图标，而是呈现出缓慢启动，逐渐加速，接近顶部时又减速的视觉动画，仿佛真的在模拟电梯升至顶层的过程。这款插件注重细节与用户体验，不仅提供视觉上的愉悦，还考虑到了用户的触感，让返回顶部的操作变得有趣而不突兀。此外，Elevator.js允许用户个性化定制，比如添加背景音乐，使得整个返回顶部的过程更具沉浸式体验。无论是对于网页设计师还是对用户体验有高要求的网站所有者来说，Elevator.js都是一种提升网站互动性和吸引力的巧妙工具。无论是在长篇文章阅读或是产品详情页浏览中，都能有效减少用户的疲劳感，提高导航效率。 点我下载 文件大小：3.46 MB 您将下载一个JQuery插件资源包，该资源包内部文件的目录结构如下： 本网站提供JQuery插件下载功能，旨在帮助广大用户在工作学习中提升效率、节约时间。 本网站的下载内容来自于互联网。如您发现任何侵犯您权益的内容，请立即告知我们，我们将迅速响应并删除相关内容。 免责声明：站内所有资源仅供个人学习研究及参考之用，严禁将这些资源应用于商业场景。 若擅自商用导致的一切后果，由使用者承担责任。

...菜单按钮时，特效即刻启动，菜单以模拟纸张逐页展开的形式展示各个子页面选项。这些子页面以层叠堆叠的方式呈现，每一个子菜单项如同一张独立的卡片，直观且具有吸引力。用户通过点击他们感兴趣的“纸张”，即可实现无缝跳转至相应的内页，不仅增强了页面之间的过渡效果，同时也极大提升了网站的整体互动性和沉浸式浏览感受。此款插件凭借其优雅且高效的动画表现，无疑能够成为现代响应式网页设计中的一大亮点，有效吸引并引导用户在各类设备上进行顺畅无阻的导航操作。 点我下载 文件大小：43.96 KB 您将下载一个JQuery插件资源包，该资源包内部文件的目录结构如下： 本网站提供JQuery插件下载功能，旨在帮助广大用户在工作学习中提升效率、节约时间。 本网站的下载内容来自于互联网。如您发现任何侵犯您权益的内容，请立即告知我们，我们将迅速响应并删除相关内容。 免责声明：站内所有资源仅供个人学习研究及参考之用，严禁将这些资源应用于商业场景。 若擅自商用导致的一切后果，由使用者承担责任。

...轻松控制计时流程——启动、暂停与重新开始。其核心功能包括：1.多级计时：用户可以根据需要设置秒、分钟或小时作为计时单位，满足各种场景需求。2.交互式控制：插件提供了启动、暂停和重新开始计时的功能按钮，使操作直观且便捷。3.美观界面：采用现代设计风格，确保与各种网站主题和谐融合，提升用户体验。4.兼容性：支持主流浏览器，确保在不同环境下稳定运行。5.轻量级：优化代码结构，减少加载时间和资源消耗，适合各类项目应用。Jquery-Stylish-Timer的集成过程简单高效，只需几行代码即可完成初始化配置。它适用于构建倒计时、计时器、任务提醒等多种功能，极大地丰富了网页交互体验。无论是个人项目还是商业应用，此插件都能有效提升网页的实用性和吸引力。 点我下载 文件大小：70.21 KB 您将下载一个JQuery插件资源包，该资源包内部文件的目录结构如下： 本网站提供JQuery插件下载功能，旨在帮助广大用户在工作学习中提升效率、节约时间。 本网站的下载内容来自于互联网。如您发现任何侵犯您权益的内容，请立即告知我们，我们将迅速响应并删除相关内容。 免责声明：站内所有资源仅供个人学习研究及参考之用，严禁将这些资源应用于商业场景。 若擅自商用导致的一切后果，由使用者承担责任。

...或其他用户交互行为来启动。该插件兼容现代浏览器，不依赖于老旧的IE浏览器，确保了在跨平台和设备上的一致性和高性能表现。无论是用于产品展示、内容切换还是提升网站整体互动性，jQuery3D立方体翻转插件都能成为网页设计与开发中的一个亮点元素，有效增强用户的浏览体验。 点我下载 文件大小：45.52 KB 您将下载一个JQuery插件资源包，该资源包内部文件的目录结构如下： 本网站提供JQuery插件下载功能，旨在帮助广大用户在工作学习中提升效率、节约时间。 本网站的下载内容来自于互联网。如您发现任何侵犯您权益的内容，请立即告知我们，我们将迅速响应并删除相关内容。 免责声明：站内所有资源仅供个人学习研究及参考之用，严禁将这些资源应用于商业场景。 若擅自商用导致的一切后果，由使用者承担责任。

...需复杂的配置就能快速启动，为开发者和设计师提供了一种快速构建动态、美观的幻灯片展示界面的方式。osSlider的百叶窗效果在slide过渡时随机生成，为每一张幻灯片的展示添加了动态变化，使得内容呈现更加生动有趣。这种效果不仅能够吸引用户的注意力，还能有效提升网站或应用的交互体验。此外，osSlider兼容主流浏览器，确保了广泛的设备和平台支持，使得其应用范围更加广泛。插件的使用极为简单，只需几个基本的HTML标签和jQuery调用，即可轻松集成到现有项目中。osSlider还支持个性化定制，允许用户根据自己的需求调整过渡速度、动画类型等参数，以适应不同的设计风格和应用场景。总之，osSlider是一款功能强大、操作简便的jQuery幻灯片插件，通过其独特的百叶窗过渡效果，为网页内容展示带来了新颖的视觉体验，是构建现代、互动性强的网站或应用的理想选择。 点我下载 文件大小：231.73 KB 您将下载一个JQuery插件资源包，该资源包内部文件的目录结构如下： 本网站提供JQuery插件下载功能，旨在帮助广大用户在工作学习中提升效率、节约时间。 本网站的下载内容来自于互联网。如您发现任何侵犯您权益的内容，请立即告知我们，我们将迅速响应并删除相关内容。 免责声明：站内所有资源仅供个人学习研究及参考之用，严禁将这些资源应用于商业场景。 若擅自商用导致的一切后果，由使用者承担责任。

...且直观。紧接着，插件启动核心的全屏变形动画效果，将触发点转化为一片与按钮颜色相呼应的全屏遮罩层，整个过程流畅且自然过渡。这种扁平化的设计风格遵循了MaterialDesign的核心原则，即在简洁、清晰的界面中融入富有层次感与动感的元素，从而提升用户体验与感知质量。通过此插件，网页设计师能够轻松为自己的项目添加时尚前卫的互动元素，使页面更具吸引力与沉浸感，不仅强化了品牌一致性，也顺应了现代Web设计趋势，让用户的每一次点击都成为一次独特的视觉享受。 点我下载 文件大小：217.68 KB 您将下载一个JQuery插件资源包，该资源包内部文件的目录结构如下： 本网站提供JQuery插件下载功能，旨在帮助广大用户在工作学习中提升效率、节约时间。 本网站的下载内容来自于互联网。如您发现任何侵犯您权益的内容，请立即告知我们，我们将迅速响应并删除相关内容。 免责声明：站内所有资源仅供个人学习研究及参考之用，严禁将这些资源应用于商业场景。 若擅自商用导致的一切后果，由使用者承担责任。

...页面滚动到特定位置时启动自定义动画或过渡效果，例如淡入淡出、位移变化、尺寸调整等。这不仅有助于提升网站视觉吸引力，还能引导用户的浏览焦点，强化信息层次和故事叙述。只需简单地调用该插件并配置相应的选项，即可在不同滚动位置添加丰富的动态响应，无需深入理解复杂的滚动监听和动画编程逻辑。因此，无论是在单页应用中创建沉浸式滚动体验，还是为多级导航菜单、图片画廊等内容区域增添生动效果，jquery.scrollex都是一个强大且高效的工具选择。 点我下载 文件大小：53.00 KB 您将下载一个JQuery插件资源包，该资源包内部文件的目录结构如下： 本网站提供JQuery插件下载功能，旨在帮助广大用户在工作学习中提升效率、节约时间。 本网站的下载内容来自于互联网。如您发现任何侵犯您权益的内容，请立即告知我们，我们将迅速响应并删除相关内容。 免责声明：站内所有资源仅供个人学习研究及参考之用，严禁将这些资源应用于商业场景。 若擅自商用导致的一切后果，由使用者承担责任。

...现计时器的基本操作：启动、暂停、继续以及完全移除。这意味着你可以在用户交互过程中动态控制计时器的状态，比如根据用户的操作暂停或恢复计时，或是响应某些条件后移除计时器。这不仅提升了用户体验，也让网站或应用的功能更加丰富多样。此外，由于其基于jQuery开发，timer.jquery兼容所有主流浏览器，确保了跨平台的一致性和稳定性。开发者还可以通过自定义CSS样式来调整计时器的外观，使其完美融入各种设计风格的网站。无论你是构建游戏、活动页面还是需要时间管理功能的应用，timer.jquery都是你的理想选择。它简化了计时功能的实现过程，让你有更多精力专注于其他核心功能的开发。 点我下载 文件大小：55.73 KB 您将下载一个JQuery插件资源包，该资源包内部文件的目录结构如下： 本网站提供JQuery插件下载功能，旨在帮助广大用户在工作学习中提升效率、节约时间。 本网站的下载内容来自于互联网。如您发现任何侵犯您权益的内容，请立即告知我们，我们将迅速响应并删除相关内容。 免责声明：站内所有资源仅供个人学习研究及参考之用，严禁将这些资源应用于商业场景。 若擅自商用导致的一切后果，由使用者承担责任。

...的API接口，提供了启动、停止、获取当前时间、设置指定时间等方法，并且支持自定义回调函数，极大地满足了开发者对时钟组件个性化需求和交互体验的要求。总而言之，FlipClock.js是一个功能强大、设计精美的jQuery插件，以其创新的3D翻页动画和灵活多样的定制选项，在提升用户体验的同时，也为网页设计师和前端开发者带来了更多创造性的可能。 点我下载 文件大小：108.21 KB 您将下载一个JQuery插件资源包，该资源包内部文件的目录结构如下： 本网站提供JQuery插件下载功能，旨在帮助广大用户在工作学习中提升效率、节约时间。 本网站的下载内容来自于互联网。如您发现任何侵犯您权益的内容，请立即告知我们，我们将迅速响应并删除相关内容。 免责声明：站内所有资源仅供个人学习研究及参考之用，严禁将这些资源应用于商业场景。 若擅自商用导致的一切后果，由使用者承担责任。

...ding指示器会自动启动，并替换或覆盖原有的按钮内容，从而传达出数据正在处理中的信息。这一特性有助于减少用户的等待焦虑感，同时确保界面交互的一致性和流畅性。此外，由于其高度可定制化的设计，开发者可以根据项目需求和品牌风格自由选择loading动画样式，实现与整体页面视觉效果的完美融合。 点我下载 文件大小：34.51 KB 您将下载一个JQuery插件资源包，该资源包内部文件的目录结构如下： 本网站提供JQuery插件下载功能，旨在帮助广大用户在工作学习中提升效率、节约时间。 本网站的下载内容来自于互联网。如您发现任何侵犯您权益的内容，请立即告知我们，我们将迅速响应并删除相关内容。 免责声明：站内所有资源仅供个人学习研究及参考之用，严禁将这些资源应用于商业场景。 若擅自商用导致的一切后果，由使用者承担责任。

...面前端。这一特性使得排查问题、演示动态数据变化以及与用户交互时的反馈变得更加直观且易于理解。这款插件无需依赖其他库（如jQuery），它简单易用，只需加载脚本并调用初始化方法即可启用。此外，Screenlog.js还提供了灵活的API，允许用户自定义日志显示的位置、格式及行为，并支持在不需要时销毁插件以恢复原始console.log()功能，确保了对原有开发流程的最小干扰。 点我下载 文件大小：11.26 KB 您将下载一个JQuery插件资源包，该资源包内部文件的目录结构如下： 本网站提供JQuery插件下载功能，旨在帮助广大用户在工作学习中提升效率、节约时间。 本网站的下载内容来自于互联网。如您发现任何侵犯您权益的内容，请立即告知我们，我们将迅速响应并删除相关内容。 免责声明：站内所有资源仅供个人学习研究及参考之用，严禁将这些资源应用于商业场景。 若擅自商用导致的一切后果，由使用者承担责任。