栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 软件开发 > 后端开发 > Java

celery的使用

Java 更新时间: 发布时间: IT归档 最新发布 模块sitemap 名妆网 法律咨询 聚返吧 英语巴士网 伯小乐 网商动力

celery的使用

文章目录

安装及配置redis

将redis服务设置为外网可访问设置密码使用密码连接重启redis服务 安装celery用celery启动异步任务

任务执行脚本(消费者)任务调用脚本(生产者)结果查询脚本

安装及配置redis
# 安装
apt-get install redis-server -y

# 后台运行
redis-server /etc/redis/redis.conf

测试是否安装成功

redis-cli
将redis服务设置为外网可访问

注释掉bind 127.0.0.1,将protected修改为no。

这里使用的kali,开放一下6379端口

# 打开6379端口
ufw allow 6379/tcp
# 重启防火墙
ufw reload
设置密码

1.修改配置文件

vim /etc/redis/redis.conf

取消requirepass的注释,后面就是密码

2.使用命令修改密码

进入redis,执行

config set requirepass 123456

注:修改配置文件为永久修改密码,命令修改为临时修改,重启失效。

使用密码连接
./redis-cli -a 123456
或者
./redis-cli
127.0.0.1:6379> auth 123456
重启redis服务
# 1、查看redis进程是否存在
ps -ef | grep redis

# 2、关闭redis
# 找到自己redis服务中的redis-cli,
redis-cli shutdown

#3、启动redis 【加&表示以后台程序方式运行,不加也可以】
redis-server &
# 使用指定配置文件启动redis
redis-server /etc/redis/redis.conf

安装celery

安装celery==4.4.1版本

pip3 install -i https://pypi.tuna.tsinghua.edu.cn/simple celery==4.4.1

安装redis,否则会缺少python和redis进行连接的模块

pip3 install redis
用celery启动异步任务 任务执行脚本(消费者)

创建异步任务执行文件celery_task.py,模拟发邮件,花费5秒钟

# celery_task
import celery
import time

backend = 'redis://yu123@192.168.170.129:6379/1'
broker = 'redis://yu123@192.168.170.129:6379/2'
cel = celery.Celery('test', backend=backend, broker=broker)

@cel.task
def send_email(name):
    print("向%s发送邮件..." % name)
    time.sleep(5)
    print("向%s发送邮件完成" % name)
    return "ok"

使用celery执行上面这个脚本,会启动多个worker监听

celery worker -A celery_task -l info

任务调用脚本(生产者)

创建执行任务文件,produce_task.py:

# produce_task.py
from celery_task import send_email
result = send_email.delay("yuan")
print(result.id)
result2 = send_email.delay("alex")
print(result2.id) 

运行这个脚本
查看日志可以发现,两个任务同时完成,只花费了5秒

同时获得了两个id

结果查询脚本

查看任务执行结果

# result.py
from celery.result import AsyncResult
from celery_task import cel

async_result = AsyncResult(id="6c9b2715-aa32-48ce-b0b0-1de5c364c85e", app=cel)

if async_result.successful():
    result = async_result.get()
    print(result)
    # result.forget() # 将结果删除
elif async_result.failed():
    print('执行失败')
elif async_result.status == 'PENDING':
    print('任务等待中被执行')
elif async_result.status == 'RETRY':
    print('任务异常后正在重试')
elif async_result.status == 'STARTED':
    print('任务已经开始被执行')

获取到了执行后return的结果ok

转载请注明:文章转载自 www.mshxw.com
本文地址:https://www.mshxw.com/it/777960.html
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号