【小沐学Python】Python实现Web服务器（Flask+celery，生产者-消费者）

文章目录

  • 1、简介
  • 2、安装和下载
    • 2.1 flask
    • 2.2 celery
    • 2.3 redis
  • 3、功能开发
    • 3.1 创建异步任务的方法
      • 3.1.1 使用默认的参数
      • 3.1.2 指定相关参数
      • 3.1.3 自定义Task基类
    • 3.2 调用异步任务的方法
      • 3.2.1 app.send_task
      • 3.2.2 Task.delay
      • 3.2.3 Task.apply_async
    • 3.3 获取任务结果和状态
  • 4、入门示例（celery）
    • 4.1 新建任务task
    • 4.2 新建应用app
    • 4.3 执行命令
  • 5、更多示例（celery）
    • 5.1 例子1
    • 5.2 例子2
    • 5.3 例子3
    • 5.4 例子4
  • 6、扩展示例（celery+flask）
    • 6.1 例子1
    • 6.2 例子2
    • 6.3 例子3
  • 结语

1、简介

Celery 是一个简单、灵活且可靠的分布式系统，用于 处理大量消息，同时为操作提供 维护此类系统所需的工具。专注于实时处理的任务队列，同时也 支持任务调度。
在这里插入图片描述

Celery 拥有庞大而多样化的用户和贡献者社区， 你应该来加入我们的IRC或我们的邮件列表。
在这里插入图片描述

Celery 是开源的，并在 BSD 许可证下获得许可。

在这里插入图片描述

  • celery的架构由三部分组成：

    • 消息中间件（broker）:
      任务调度队列，用于存放task，接收任务生产者发来的消息（即任务），将任务tasks存入队列。Celery 本身不提供队列服务，但是可以方便的和第三方提供的消息中间件集成，官方推荐使用 RabbitMQ 和 Redis。

    • 任务执行单元（worker）
      Worker 是执行任务的处理单元，它实时监控消息队列，获取队列中调度的任务，并执行它。

    • 任务执行结果存储（backend）
      Backend 用于存储任务的执行结果，以供查询。同消息中间件一样，存储也可使用 RabbitMQ, Redis 和 MongoDB 等。

在这里插入图片描述
Celery 采用典型生产者和消费者模型。生产者提交任务到任务队列，众多消费者从任务队列中取任务执行。
在这里插入图片描述

2、安装和下载

查看安装的库的信息如下：

pip list

在这里插入图片描述

2.1 flask

Flask 是一款使用 Python 编写的轻量级 Web 应用框架，它基于 Werkzeug WSGI 工具箱和 Jinja2 模板引擎。Flask 由 Armin Ronacher 开发，其目标是提供一个简单、灵活且易于扩展的框架，可以帮助开发人员快速构建 Web 应用程序。

pip install Flask

在这里插入图片描述

2.2 celery

Celery 是一个 基于python开发的分布式异步消息任务队列，通过它可以轻松的实现任务的异步处理。
Celery 在执行任务时需要通过一个消息中间件（Broker）来接收和发送任务消息，以及存储任务结果， 一般使用rabbitMQ or Redis。

pip install celery
#pip install -U Celery
#pip install "celery[librabbitmq]"
#pip install "celery[librabbitmq,redis,auth,msgpack]"

在这里插入图片描述

2.3 redis

https://redis.io/download/

REmote DIctionary Server(Redis) 是一个由 Salvatore Sanfilippo 写的 key-value 存储系统，是跨平台的非关系型数据库。

Redis 是一个开源的使用 ANSI C 语言编写、遵守 BSD 协议、支持网络、可基于内存、分布式、可选持久性的键值对(Key-Value)存储数据库，并提供多种语言的 API。

Redis 通常被称为数据结构服务器，因为值（value）可以是字符串(String)、哈希(Hash)、列表(list)、集合(sets)和有序集合(sorted sets)等类型。
在这里插入图片描述
安装redis库：

pip install redis

在这里插入图片描述

3、功能开发

3.1 创建异步任务的方法

任何被 task 修饰的方法都会被创建一个 Task 对象，变成一个可序列化并发送到远程服务器的任务；

3.1.1 使用默认的参数

@celery.task
def function_name():pass

示例如下：

app = Celery('tasks', backend='redis://localhost', broker='pyamqp://')@app.task
def add(x, y):return x+y

或

from celery.utils.log import get_task_logger
logger = get_task_logger(__name__)@app.task
def add(x, y):logger.info('Adding {0} + {1}'.format(x, y))return x + y

3.1.2 指定相关参数

@celery.task(bind=True, name='name')
def function_name():pass

示例如下：

name : 可以显式指定任务的名字；默认是模块的命名空间中本函数的名字。
bind : 一个bool值，设置是否绑定一个task的实例，如果绑定，task实例会作为参数传递到任务方法中，可以访问task实例的所有的属性，即前面反序列化中那些属性

# 当bind=True时，add函数第一个参数是self，指的是task实例
@task(bind=True)  # 第一个参数是self，使用self.request访问相关的属性
def add(self, x, y):try:logger.info(self.request.id)except:self.retry() # 当任务失败则进行重试，也可以通过max_retries属性来指定最大重试次数

或

logger = get_task_logger(__name__)@app.task(bind=True)
def add(self, x, y):logger.info(self.request.id)

3.1.3 自定义Task基类

import celeryclass MyTask(celery.Task):# 任务失败时执行def on_failure(self, exc, task_id, args, kwargs, einfo):print('{0!r} failed: {1!r}'.format(task_id, exc))# 任务成功时执行def on_success(self, retval, task_id, args, kwargs):pass# 任务重试时执行def on_retry(self, exc, task_id, args, kwargs, einfo):pass@task(base=MyTask)
def add(x, y):raise KeyError()

或

import celeryclass MyTask(celery.Task):def on_failure(self, exc, task_id, args, kwargs, einfo):print('{0!r} failed: {1!r}'.format(task_id, exc))@app.task(base=MyTask)
def add(x, y):raise KeyError()

3.2 调用异步任务的方法

3.2.1 app.send_task

send_task 在发送的时候是不会检查 tasks.add 函数是否存在的，即使为空也会发送成功，所以 celery 执行是可能找不到该函数报错；

# tasks.py
from celery import Celery
app = Celery()def add(x, y):return x+y# app.py
app.send_task('tasks.add',args=[3,4])  # 参数基本和apply_async函数一样

3.2.2 Task.delay

delay 方法是 apply_async 方法的简化版，不支持执行选项，只能传递任务的参数。

# tasks.py
from celery import Celery
app = Celery()@app.task
def add(x, y, z=0):return x + y# app.py
add.delay(30, 40, z=5)	# 包括位置参数和关键字参数

3.2.3 Task.apply_async

apply_async 支持执行选项，它会覆盖全局的默认参数和定义该任务时指定的执行选项，本质上还是调用了 send_task 方法；

# tasks.py
from celery import Celery
app = Celery()@app.task
def add(x, y, z=0):return x + y# app.py
add.apply_async(args=[30,40], kwargs={'z':5})

3.3 获取任务结果和状态

由于 celery 发送的都是去其他进程执行的任务，如果需要在客户端监控任务的状态，有如下方法：

r = task.apply_async()
r.ready()     # 查看任务状态，返回布尔值,  任务执行完成, 返回 True, 否则返回 False.
r.wait()      # 会阻塞等待任务完成, 返回任务执行结果，很少使用；
r.get(timeout=1)       # 获取任务执行结果，可以设置等待时间，如果超时但任务未完成返回None；
r.result      # 任务执行结果，未完成返回None；
r.state       # PENDING, START, SUCCESS，任务当前的状态
r.status      # PENDING, START, SUCCESS，任务当前的状态
r.successful  # 任务成功返回true
r.traceback  # 如果任务抛出了一个异常，可以获取原始的回溯信息

在这里插入图片描述

4、入门示例（celery）

4.1 新建任务task

  • tasks.py
from celery import Celery#消息中间件（使用的redis）
broker = 'redis://localhost:6379/1'
#结果存储（使用的redis）
backend = 'redis://localhost:6379/2'
#实例化Celery对象
app = Celery('celeryDemo',broker=broker,backend=backend
)@app.task()
def add(x,y):print('task: add')return x+y

4.2 新建应用app

  • app.py
from tasks import addif __name__ == '__main__':print('task start....')result = add.delay(2,3)print('task end....')print(result)

或者

#coding=utf-8
from tasks import addif __name__ == '__main__':# delay与apply_async生成的都是AsyncResult对象res = add.apply_async((123, 100))if res.successful():result = res.get()print(result)elif res.failed():print('任务失败')elif res.status == 'PENDING':print('任务等待中被执行')elif res.status == 'RETRY':print('任务异常后正在重试')elif res.status == 'STARTED':print('任务已经开始被执行')

4.3 执行命令

（1）运行redis服务器：
命令行执行如下程序。

redis-server.exe

在这里插入图片描述
在这里插入图片描述
（2）运行worker端:

##常规启动Worker 
#celery -A tasks worker --loglevel=INFO ##Windows下启动Worker 
#pip install eventlet 
#celery -A tasks worker --loglevel=INFO -P eventlet 
#celery -A tasks worker -l info -P eventlet  -c 10celery -A tasks worker -l info -P threads

在这里插入图片描述

（3）运行 app.py文件：

python app.py

在这里插入图片描述
在这里插入图片描述

5、更多示例（celery）

5.1 例子1

  • config.py
# config.py
# 设置配置
BROKER_URL =  'amqp://username:password@localhost:5672/yourvhost'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_TASK_SERIALIZER = 'msgpack'
CELERY_RESULT_SERIALIZER = 'msgpack'
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24
CELERY_ACCEPT_CONTENT = ["msgpack"]
CELERY_DEFAULT_QUEUE = "default"   
CELERY_QUEUES = {"default": { # 这是上面指定的默认队列"exchange": "default","exchange_type": "direct","routing_key": "default"}
}
  • tasks.py
# tasks.py
from app import celery@celery.task
def add(x, y):return x + y@celery.task(name="sub")
def sub(x, y):return x - y
  • app.py
# app.py --- 初始化celery对象 
from celery import Celery
import config
from task import add, subcelery = Celery(__name__, include=["task"]) # 设置需要导入的模块
# 引入配置文件
celery.config_from_object(config)if __name__ == '__main__':add.apply_async((2,2), routing_key='default',priority=0,exchange='default')

5.2 例子2

  • config.py
# coding: utf-8
from celery import Celerycelery_broker = 'amqp://guest@127.0.0.1//'
celery_backend = 'amqp://guest@127.0.0.1//'# Add tasks here
CELERY_IMPORTS = ('tasks',
)app = Celery('celery', broker=celery_broker, 
backend=celery_backend, include=CELERY_IMPORTS)app.conf.update(CELERY_ACKS_LATE=True,  # 允许重试CELERY_ACCEPT_CONTENT=['pickle', 'json'],CELERY_TASK_SERIALIZER='json',CELERY_RESULT_SERIALIZER='json',# 设置并发worker数量CELERYD_CONCURRENCY=4, # 每个worker最多执行500个任务被销毁，可以防止内存泄漏CELERYD_MAX_TASKS_PER_CHILD=500, BROKER_HEARTBEAT=0,  # 心跳CELERYD_TASK_TIME_LIMIT=12 * 30,  # 超时时间
)# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = True
# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {'sub': {'task': 'tasks.sub','schedule': timedelta(seconds=3),# 每周一早八点# 'schedule': crontab(hour=8, day_of_week=1), 'args': (300, 150),}
}
  • tasks.py
#coding=utf-8
from config import app
from celery.signals import worker_process_init, worker_process_shutdown@worker_process_init.connect
def init_worker(*args, **kwargs):# 初始化资源pass@worker_process_shutdown.connect
def release_worker(*args, **kwargs):# 释放资源pass# 普通函数装饰为 celery task
@app.task
def add(x, y):return x + y@app.task
def sub(x, y):return x - y
  • main.py
#coding=utf-8import time
from tasks import addif __name__ == '__main__':a = time.time()# delay与apply_async生成的都是AsyncResult对象async = add.apply_async((1, 100))if async.successful():result = async.get()print(result)elif async.failed():print('任务失败')elif async.status == 'PENDING':print('任务等待中被执行')elif async.status == 'RETRY':print('任务异常后正在重试')elif async.status == 'STARTED':print('任务已经开始被执行')

5.3 例子3

  • task1.py
# -*- coding: utf-8 -*-# 使用celery
import time
from celery import Celery
import redis# 创建一个Celery类的实例对象
app = Celery('celery_demo', broker='redis://127.0.0.1:6379/15')@app.task
def add(a, b):count = a + bprint('任务函数正在执行....')time.sleep(1)return count
  • app1.py
import time
from task1 import adddef notity(a, b):result = add.delay(a, b)return resultif __name__ == '__main__':for i in range(5):time.sleep(1)result = notity(i, 100)print(result)

先执行celery命令：

celery -A task1 worker -l info -P eventlet

在这里插入图片描述
再执行worker命令：

python app1.py

在这里插入图片描述

5.4 例子4

  • celery_config.py
#-*-coding=utf-8-*-
from __future__ import absolute_importfrom celery.schedules import crontab
# 中间件
BROKER_URL = 'redis://localhost:6379/6'# 结果存储CELERY_RESULT_BACKEND = 'redis://:127.0.0.1:6379/5' 
# 默认worker队列
CELERY_DEFAULT_QUEUE = 'default'
# 异步任务
CELERY_IMPORTS = ("tasks"
)from datetime import timedelta
# celery beat
CELERYBEAT_SCHEDULE = {'add':{'task':'tasks.add','schedule':timedelta(seconds=10),'args':(1,12)},# 每10s执行一次'task1': {'task': 'tasks.add','schedule': timedelta(seconds=10),'args': (2, 8)},# 每天15:00执行'task2': {'task': 'tasks.add','schedule': crontab(hour=15, minute=0),'args': (9, 9)}
}
  • celery_app.py
from __future__ import absolute_import
from celery import Celeryapp = Celery('celery_app')
app.config_from_object('celery_config')
  • tasks.py
from celery_app import app@app.task(queue='default')
def add(x, y):return x + y@app.task(queue='default')
def sub(x, y):return x - y
  • app.py
import sys, os# sys.path.append(os.path.abspath('.'))
sys.path.append(os.path.abspath('..'))
from tasks import adddef add_loop():ret = add.apply_async((1, 2), queue='default')print(type(ret))return retif __name__ == '__main__':ret = add_loop()print(ret.get())print(ret.status)

  • 运行命令
    （1）终端输入：celery -A celery_app worker -Q default --loglevel=info
    （2）终端输入：celery -A celery_app beat
    （3）终端执行：python app.py

  • 启动worker

celery -A celery_app worker --loglevel=info
#celery -A celery_app worker --loglevel=info --concurrency=10
  • 启动定时任务配置：
celery -A celery_app beat --loglevel=info
  • 同时启动worker和beat：
celery -A celery_app worker --beat --loglevel=info

6、扩展示例（celery+flask）

flask是一个阻塞式的框架。这里的“阻塞”是指flask处理请求的时候，一次只能处理一个，当多个requests过来，flask会说，大家不要急，一个一个来。

6.1 例子1

  • mycelery.py
from celery import Celerydef make_celery(app):celery = Celery(      #实例化Celery'tasks',broker='redis://localhost:6379/1',      #使用redis为中间人backend='redis://localhost:6379/2'      #结果存储)class ContextTask(celery.Task):    #创建ContextTask类并继承Celery.Task子类def __call__(self, *args, **kwargs): with app.app_context():     #和Flask中的app建立关系return self.run(*args, **kwargs) #返回任务celery.Task = ContextTask     #异步任务实例化ContextTaskreturn celery        #返回celery对象
  • tasks.py
import time
from app import celery@celery.task   #使用异步任务装饰器task
def add(x, y):time.sleep(5)  #休眠5秒return x + y
  • app.py
from flask import Flask
import tasks
from mycelery import make_celeryapp = Flask(__name__)
celery = make_celery(app)    #调用make_celery方法并传入app使celery和app进行关联@app.route('/')
def hello():tasks.add.delay(1,2)    #调用tasks文件中的add()异步任务方法return '请求正在后台处理中，您可以去处理其他事情'if __name__ == '__main__':app.run(debug=True)

在终端执行如下代码运行Celery命令：

celery -A tasks worker -l info -P eventlet  -c 10

启动Flask程序，访问http://127.0.0.1:5000/后在终端查Worker服务：

http://127.0.0.1:5000

6.2 例子2

  • main.py
from flask import Flask
from celery import Celery
from celery.result import AsyncResult
import timeapp = Flask(__name__)
# 用以储存消息队列
app.config['CELERY_BROKER_URL'] = 'redis://127.0.0.1:6379/11'
# 用以储存处理结果
app.config['CELERY_RESULT_BACKEND'] = 'redis://127.0.0.1:6379/12'celery_ = Celery(app.name, broker=app.config['CELERY_BROKER_URL'], backend=app.config['CELERY_RESULT_BACKEND'])
celery_.conf.update(app.config)@celery_.task
def task_add(arg1, arg2):# 两数相加time.sleep(10)return arg1+arg2@app.route("/sum/<arg1>/<arg2>")
def route_sum(arg1,arg2):# 发送任务到celery,并返回任务ID,后续可以根据此任务ID获取任务结果result = task_add.delay(int(arg1),int(arg2))return result.id@app.route("/get_result/<result_id>")
def route_result(result_id):# 根据任务ID获取任务结果result = celery_.AsyncResult(id=result_id)return str(result.get())if __name__ == "__main__":app.run(debug=True)

在终端执行如下代码运行Celery命令：

celery -A main.celery_ worker -l info -P eventlet  -c 10

启动Flask程序，访问http://127.0.0.1:5000/后在终端查Worker服务：

http://127.0.0.1:5000

在这里插入图片描述
通过链接发送任务,并获得任务ID:

http://127.0.0.1:5000/sum/1/2

在这里插入图片描述

通过下面这个链接可以获得任务结果:

http://127.0.0.1:5000/get_result/105fb6d4-67ae-46ab-8c84-77381821a43c

在这里插入图片描述

另外可以通过flower模块来监控celery的任务:

flower --basic_auth=admin:admin --broker=redis://127.0.0.1:6379/0 --address=0.0.0.0 --port=5556
# celery flower --broker=redis://localhost:6379/6
http://127.0.0.1:5556

6.3 例子3

  • main.py
from flask import Flask, jsonify
from celery import create_app, shared_taskapp = Flask(__name__)
app.config.update(CELERY_BROKER_URL='redis://localhost:6379/0',CELERY_RESULT_BACKEND='redis://localhost:6379/0'
)celery = create_app(app)@shared_task(bind=True)
def add(self, x, y):return x + y@app.route('/')
def index():return 'Hello, World!'@app.route('/add/<int:x>/<int:y>')
def add_route(x, y):task = add.delay(x, y)return jsonify({'task_id': task.id})@app.route('/result/<task_id>')
def get_result_route(task_id):task = add.AsyncResult(task_id)if task.state == 'SUCCESS':return jsonify({'result': task.result})else:return jsonify({'status': task.state})if __name__ == '__main__':app.run(debug=True)

结语

如果您觉得该方法或代码有一点点用处，可以给作者点个赞，或打赏杯咖啡；╮(￣▽￣)╭
如果您感觉方法或代码不咋地//(ㄒoㄒ)//，就在评论处留言，作者继续改进；o_O???
如果您需要相关功能的代码定制化开发，可以留言私信作者；(✿◡‿◡)
感谢各位大佬童鞋们的支持！( ´ ▽´ )ﾉ ( ´ ▽´)っ！！！
在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/194929.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

用CHAT如何写教研室工作总结？

用CHAT如何写教研室工作总结？

问CHAT&#xff1a;写一份教研室工作总结 CHAT回复&#xff1a;以下是一个教研室工作总结的大纳&#xff0c;具体内容需要根据你们教研室的实际情况进行填充和修改。 教研室XXXX年度工作总结 1. 引言&#xff1a;简要介绍本年度工作总结的目的和主题。 2. 教育教学工作&…
阅读更多...
Java核心知识点整理大全24-笔记

Java核心知识点整理大全24-笔记

目录 22. 数据结构 22.1.1. 栈&#xff08;stack&#xff09; 22.1.2. 队列&#xff08;queue&#xff09; 22.1.3. 链表&#xff08;Link&#xff09; 22.1.4. 散列表&#xff08;Hash Table&#xff09; 22.1.5. 排序二叉树 22.1.5.1. 插入操作 22.1.5.2. 删除操作 2…
阅读更多...
知识点滴 - 什么是半透膜和渗透压

知识点滴 - 什么是半透膜和渗透压

半透膜和渗透作用 1748年的一天&#xff0c;法国物理学家诺勒为了改进酒的制作水平&#xff0c;设计了这样一个试验&#xff1a;在一个玻璃圆筒中装满酒精&#xff0c;用猪膀胱封住&#xff0c;然后把圆筒全部浸在水中。当他正要做下一步的工作时&#xff0c;突然发现&#xff…
阅读更多...
DDOS攻击为何永不过时？

DDOS攻击为何永不过时？

文章目录 一、DOS二、DDOS三、如何防范DDOS 1.可以过滤IP地址2.增加设备3.在骨干节点配置防火墙4.开启过滤5.配置DNS抗攻击6.白帽团队 四、白帽子 为什么二十年前中国红客们就在用的DDOS攻击直到现在还依然是黑客们最爱的攻击方法&#xff1f;二十年前的攻击技术为什么还不过…
阅读更多...
基于Java SSM框架+Vue实现教学视频点播网站项目【项目源码+论文说明】计算机毕业设计

基于Java SSM框架+Vue实现教学视频点播网站项目【项目源码+论文说明】计算机毕业设计

基于java的SSM框架Vue实现教学视频点播网站演示 摘要 随着现在网络的快速发展&#xff0c;网上管理系统也逐渐快速发展起来&#xff0c;网上管理模式很快融入到了许多学院的之中&#xff0c;随之就产生了“视频点播系统”&#xff0c;这样就让视频点播系统更加方便简单。 对于…
阅读更多...
Java 学习之多态

Java 学习之多态

多态的概念 多态 晚绑定。 所谓多态&#xff0c;就是父类型的引用可以指向子类型的对象&#xff0c;或者接口类型的引用可以指向实现该接口的类的实例。 不要把函数重载理解为多态。因为多态是一种运行期的行为&#xff0c;不是编译期的行为。 多态&#xff1a;父类型的引用可…
阅读更多...
java源码-Java方法的定义和使用详解

java源码-Java方法的定义和使用详解

1、 方法定义 如果我们想定义一个方法&#xff0c;基本语法如下&#xff1a; 修饰符&#xff1a;方法的修饰符是可选的&#xff0c;用于定义该方法的访问类型&#xff0c;可用的修饰符包括public/private/protected/默认的。 返回值&#xff1a;方法可以有返回值&#xff0c;…
阅读更多...
线程中出现异常的处理

线程中出现异常的处理

目录 前言 正文 1.线程出现异常的默认行为 2.使用 setUncaughtExceptionHandler() 方法进行异常处理 3.使用 setDefaultUncaughtExceptionHandler() 方法进行异常处理 4.线程组内处理异常 5.线程异常处理的优先性 总结 前言 在紧密交织的多线程环境中&#xff0c;异…
阅读更多...
解决：ValueError: binary mode doesn‘t take an encoding argument

解决：ValueError: binary mode doesn‘t take an encoding argument

解决&#xff1a;ValueError: binary mode doesn‘t take an encoding argument 文章目录 解决&#xff1a;ValueError: binary mode doesn‘t take an encoding argument背景报错问题报错翻译报错位置代码报错原因解决方法方法一方法二今天的分享就到此结束了 背景 在使用之前…
阅读更多...
USART的PAL库编程

USART的PAL库编程

USART驱动的工作原理 总结一下我们之前使用中断的方式来进行数据的发送和接收 如果收到数据数据在RDR寄存器中 RXNE标志位就从0到1触发中断 进入中断服务函数 把数据缓存在队列中 然后在到进程函数中断接收数据函数中进行出队处理 发送数据就是把中断关闭&#xff08;标志位TXE…
阅读更多...
日志模块Loguru

日志模块Loguru

安装 Loguru 仅支持 Python 3.5 及以上的版本&#xff0c;使用 pip 安装即可&#xff1a; pip install loguru开箱即用 Loguru 的主要概念是只有一个&#xff1a;logger from loguru import loggerlogger.info("This is log info!") logger.warning("This i…
阅读更多...
食物相关的深度学习数据集合集—食物、饮料、肉类、餐具等数据集

食物相关的深度学习数据集合集—食物、饮料、肉类、餐具等数据集

最近收集了一大波与食物酒水相关的数据集&#xff0c;包含食物、饮料、肉类、餐具等不同等类型的数据集&#xff0c;废话不多说&#xff0c;给大家逐一介绍&#xff01;&#xff01; 1、自制啤酒配方数据库 超过20万自制啤酒配方数据库&#xff0c;数据集包含不同精酿啤酒的名…
阅读更多...
Docker Image（镜像）——5

Docker Image（镜像）——5

目录&#xff1a; Docker 镜像是什么镜像生活案例镜像分层生活案例为什么需要镜像镜像命令详解 镜像命令清单docker imagesdocker tagdocker pulldocker pushdocker rmidocker savedocker loaddocker historydocker importdocker image prunedocker build镜像操作案例 查找镜像…
阅读更多...
etlbox.3.1.0 for NET 轻量级 ETL数据集成库 Crack

etlbox.3.1.0 for NET 轻量级 ETL数据集成库 Crack

适用于 .NET 的轻量级 ETL&#xff08;提取、转换、加载&#xff09;工具箱和数据集成库 高度可定制 厌倦了使用几乎不可能实现复杂需求的用户界面&#xff1f;使用 ETLBox&#xff0c;可以轻松编写适合您独特需求的代码。插入您自己的逻辑或修改现有行为以满足您的特定要求。 …
阅读更多...
ScyllaDB 基础入门

ScyllaDB 基础入门

简介 ScyllaDB 是一种开源的 NoSQL 数据库&#xff0c;它提供了高性能、低延迟的数据处理能力&#xff0c;同时保持了与 Apache Cassandra 高度的兼容性。ScyllaDB 使用了一种名为 “Seastar” 的高效并行编程框架&#xff0c;并采用了 C 进行开发&#xff0c;因此它能够充分利…
阅读更多...
基于ssm Vue的戒烟网站源码和论文

基于ssm Vue的戒烟网站源码和论文

基于ssm Vue的戒烟网站源码和论文734 开发工具&#xff1a;idea 数据库mysql5.7 数据库链接工具&#xff1a;navcat,小海豚等 环境&#xff1a; jdk8 tomcat8.5 开发技术 ssm 摘要 随着互联网的高速发展&#xff0c;线上管理成为当代人们管理事物的重要手段之一&#xff…
阅读更多...
numpy知识库：基于numpy绘制灰度直方图

numpy知识库：基于numpy绘制灰度直方图

前言 对于灰度图像而言&#xff0c;灰度直方图可以统计灰度图像内各个灰度级出现的次数。 灰度直方图的横坐标是灰度图像中各像素点的灰度级。灰度的数值范围为[0, 255]。因此&#xff0c;如果将图像分为256个灰度级&#xff0c;那么每个灰度级唯一对应一个灰度&#xff1b;如…
阅读更多...
流媒体方案之Nginx——实现物联网视频监控项目

流媒体方案之Nginx——实现物联网视频监控项目

目录 前言 一、Nginx是什么 二、Nginx在流媒体方案中的位置​编辑 三、软硬件准备 四、移植编译Nginx 五、运行Ngnix 六、测试流媒体方案 七、浏览器播放 前言 最近想做一个安防相关的项目&#xff0c;所以跟着韦东山老师的视频来学习视频监控方案的相关知识&#xff0…
阅读更多...
lv11 嵌入式开发 ADC 16

lv11 嵌入式开发 ADC 16

目录 1 ADC 简介 2 Exynos4412下的ADC控制器 2.1 总览 2.2 特征 2.3 ADC转换时间 2.4 IO口 ​编辑3 ADC寄存器详解 3.1 寄存器介绍 3.2 ADCCON控制寄存器 3.3 ADCDAT 3.4 CLRINTADC 3.5 ADCMUX ​编辑 4 ADC编程 1 ADC 简介 ADC(Analog to Digital Converter)即…
阅读更多...
Android开发，JNI开发项目创建

Android开发，JNI开发项目创建

文章目录 Android开发&#xff0c;JNI开发项目创建1.jni是什么 Android开发&#xff0c;JNI开发项目创建 创建工程 1.jni是什么 使得java可以访问底层c语言&#xff0c;java本地化接口&#xff0c;是桥梁。 运行下我们的项目 出现这个就是我们的JNI开发环境已经配置好了 是…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023