搜索引擎优化宝典 pdf(PythonCelery快速开发音乐高潮提取网(1)_)

　　搜索引擎优化宝典 pdf(PythonCelery快速开发音乐高潮提取网(1)_)

　　内容

　　Bamboo-pipeline是蓝鲸智云旗下SaaS标准运维的流程编排引擎。它具有以下特点：

　　

　　本文需要涉及Django和Celery的前置知识。如果你还是不明白这两者，建议你在谷歌或者百度上搜索一下，或者阅读我之前的文章：

　　姜戈：

　　Python Django 快速开发音乐高潮提取网络(1)

　　Python Django 快速开发音乐高潮提取网络(2)

　　Python Django 快速开发音乐高潮提取网络(3)

　　Pycharm+Django 安装配置指南

　　*敏*感*词*的Django+Vue前端分离入门实用教程

　　芹菜：

　　Python celery 异步快速下载*敏*感*词*

　　Django Celery 异步和定时任务实用教程

　　网络IO谁更快？Python 和 Go 请求速度对比

　　什么？Python Celery 也可以调度 Go 工人吗？

　　1.准备

　　在开始之前，您需要确保您的计算机上已成功安装 Python 和 pip。如果没有，请访问这个文章：超详细Python安装指南进行安装。

　　（可选1)如果使用Python进行数据分析，可以直接安装Anaconda：Anaconda，Python数据分析挖掘的好帮手，内置了Python和pip。

　　（可选2) 另外，推荐大家使用VSCode编辑器编写Python小项目：Python编程的最佳搭档——VSCode详解

　　Windows环境下打开Cmd（开始—运行—CMD），苹果系统环境下打开Terminal（command+空格进入Terminal），输入命令安装依赖：

　　pip install bamboo-engine

pip install bamboo-pipeline

pip install django

pip install celery

　　2. 项目初始化

　　（选项1：没有Django项目）如果您没有现成的Django项目，请按照以下步骤进行初始化

　　由于竹管道运行时是基于 Django 实现的，因此需要新建一个 Django 项目：

　　django-admin startproject easy_pipeline

cd easy_pipeline

　　在easy_pipeline.settings.py下添加如下配置：

　　from pipeline.eri.celery.queues import *

from celery import Celery

app = Celery("proj")

app.config_from_object("django.conf:settings")

INSTALLED_APPS = [

...

"pipeline",

"pipeline.engine",

"pipeline.component_framework",

"pipeline.eri",

...

]

　　在easy_pipeline目录下初始化数据库：

　　python manage.py migrate

　　（方案二：有一个Django项目需要使用流程引擎）如果你有现成的PipeLine项目需要使用这个流程引擎，请在项目的settings.py下添加如下配置：

　　from pipeline.eri.celery.queues import *

from celery import Celery

app = Celery("proj")

app.config_from_object("django.conf:settings")

INSTALLED_APPS = [

...

"pipeline",

"pipeline.engine",

"pipeline.component_framework",

"pipeline.eri",

...

]

　　然后重新执行 migrate 生成管道相关的流程模型：

　　python manage.py migrate

　　执行 migrate 后，将如下图所示：

　　

　　由于原项目使用流程引擎，可能存在一些版本不匹配的问题。如果遇到错误，请排查并解决，或者到蓝鲸官网查询。

　　3. 实践中的简单流程编排

　　首先在项目目录下启动celery worker：

　　python manage.py celery worker -Q er_execute,er_schedule --pool=solo -l info

　　启动成功类似于下图：

　　

　　（注）如果你对你原来的Django项目进行改造，可能无法成功启动。这是因为 Pipeline 使用 Django 2.2.24，会有很多版本不兼容。如遇到错误，请排查故障或到蓝鲸官网查询。

　　在以下示例中，我们将创建并执行一个简单的流程：

　　

　　3.1 创建流程应用

　　在bamboo_pipeline 中，一个进程由多个组件组成。官方推荐使用APP来管理和控制组件：

　　python manage.py create_plugins_app big_calculator

　　该命令会在Django项目的根目录下生成一个目录结构如下的APP：

　　big_calculator

├── __init__.py

├── components

│   ├── __init__.py

│   └── collections

│   ├── __init__.py

│   └── plugins.py

├── migrations

│   └── __init__.py

└── static

└── big_calculator

└── plugins.js

　　不要忘记将这个新创建的插件添加到 Django 配置的 INSTALLED_APPS 中：

　　INSTALLED_APPS = (

...

'big_calculator',

...

)

　　3.2 写进程的服务原子

　　组件服务 Service 是组件的核心。Service 定义了调用组件时执行的逻辑。我们来实现一个计算传入参数n的阶乘，并将结果写入输出Service，在big_calculator/components/采集s/中，在plugins.py中输入以下代码：

　　import math

from pipeline.core.flow.activity import Service

class FactorialCalculateService(Service):

def execute(self, data, parent_data):

"""

组件被调用时的执行逻辑

:param data: 当前节点的数据对象

:param parent_data: 该节点所属流程的数据对象

:return:

"""

n = data.get_one_of_inputs('n')

if not isinstance(n, int):

data.outputs.ex_data = 'n must be a integer!'

return False

data.outputs.factorial_of_n = math.factorial(n)

return True

def inputs_format(self):

"""

组件所需的输入字段，每个字段都包含字段名、字段键、字段类型及是否必填的说明。

:return:必须返回一个 InputItem 的数组，返回的这些信息能够用于确认该组件需要获取什么样的输入数据。

"""

return [

Service.InputItem(name='integer n', key='n', type='int', required=True)

]

def outputs_format(self):

"""

组件执行成功时输出的字段，每个字段都包含字段名、字段键及字段类型的说明

:return: 必须返回一个 OutputItem 的数组, 便于在流程上下文或后续节点中进行引用

"""

return [

Service.OutputItem(name='factorial of n', key='factorial_of_n', type='int')

]

　　首先，我们继承了Service 基类并实现了execute() 和outputs_format() 方法。它们的功能如下：

　　我们来看看execute()方法内部执行的逻辑。首先，我们尝试从当前节点数据对象的输出中获取输入参数 n。如果获取的参数不是int实例，那么我们将异常信息写入当前节点输出的ex_data字段中，该字段是引擎内部的保留字段，节点执行失败时产生的异常信息应该写入此字段。然后我们返回 False 表示这次组件失败了，然后节点就会进入失败状态：

　　n = data.get_one_of_inputs('n')

if not isinstance(n, int):

data.outputs.ex_data = 'n must be a integer!'

return False

　　如果得到的 n 是一个普通的 int，我们调用 math.factorial() 函数来计算 n 的阶乘。计算完成后，我们将结果写入输出的 factorial_of_n 字段，供流程中其他节点使用：

　　data.outputs.factorial_of_n = math.factorial(n)

return True

　　3.3 编写流程组件并绑定Service原子

　　在完成Service的准备工作后，我们需要将其绑定到一个Component上，才能在组件库中注册。在 big_calculator\components\__init__.py 文件下添加以下代码：

　　import logging

from pipeline.component_framework.component import Component

from big_calculator.components.collections.plugins import FactorialCalculateService

logger = logging.getLogger('celery')

class FactorialCalculateComponent(Component):

name = 'FactorialCalculateComponent'

code = 'fac_cal_comp'

bound_service = FactorialCalculateService

　　我们定义了一个继承自基类 Component 的 FactorialCalculateComponent 类，它具有以下属性：

　　这样，我们就完成了一个流程原子的开发。

　　3.4 生成流程，测试刚刚写的组件

　　在 big_calculator\test.py 中写入以下内容，生成测试刚刚编写的组件的流程：

　　# Python 实用宝典

# 2021/06/20

import time

from bamboo_engine.builder import *

from big_calculator.components import FactorialCalculateComponent

from pipeline.eri.runtime import BambooDjangoRuntime

from bamboo_engine import api

from bamboo_engine import builder

def bamboo_playground():

"""

测试流程引擎

"""

# 使用 builder 构造出流程描述结构

start = EmptyStartEvent()

# 这里使用 我们刚创建好的n阶乘组件

act = ServiceActivity(component_code=FactorialCalculateComponent.code)

# 传入参数

act.component.inputs.n = Var(type=Var.PLAIN, value=4)

end = EmptyEndEvent()

start.extend(act).extend(end)

pipeline = builder.build_tree(start)

api.run_pipeline(runtime=BambooDjangoRuntime(), pipeline=pipeline)

# 等待 1s 后获取流程执行结果

time.sleep(1)

result = api.get_execution_data_outputs(BambooDjangoRuntime(), act.id).data

print(result)

　　然后在命令行输入

　　python manage.py shell

　　打开django控制台，输入以下命令执行这个过程：

　　from big_calculator.test import bamboo_playground

bamboo_playground()

　　流程结束后，获取节点的执行结果，可以看到节点输出factorial_of_n，值为24(4 * 3 * 2 *1)，正是我们需要的效果：

　　{'_loop': 0, '_result': True, 'factorial_of_n': 24}

　　恭喜，您已经成功创建了一个进程并运行了它！在此期间，您可能会遇到很多坑。建议先尝试自己解决。如果解决不了，可以到标准运维仓库提出问题，或者到蓝鲸智云官网提问。

　　我们的文章到此结束。如果您喜欢今天的Python教程，请继续关注Python实用指南。

　　如有疑问，可在公众号后台回复：加群，回答相应验证信息，进入互助群查询。

　　原创不容易，希望大家点赞支持我继续创作，谢谢！

　　Python 实用书 ()

　　不只是一本书