使用java程序作为celery的工作节点

Posted 2020-12-25 堕落门徒

celery是python实现的分布式调度框架，有时候想用celery去调用java服务，正好有一个celery-java的库可以使用，能达到这个效果，记录一下:

先添加依赖:

        <dependency>
            <groupId>com.geneea.celery</groupId>
            <artifactId>celery-java</artifactId>
            <version>1.10.4</version>
        </dependency>

先定义任务类TestTask.java

import com.geneea.celery.CeleryTask;

@CeleryTask
public class TestTask {

    public int sum(int x, int y){
        return x + y;
    }
}

再定义工作入口类MyWorker.java

import com.geneea.celery.CeleryWorkerCLI;

public class MyWorker {
    public static void main(String[] args) throws Exception {
        //注意broker的url的末尾/必须用转义符%2F
        args = new String[]{"--queue","celery",
                            "--concurrency","2",
                            "--broker","amqp://guest:[email protected]:5672/%2F"};
        CeleryWorkerCLI.main(args);
    }
}

运行MyWorker即可启动java版的celery工作节点:

Started consuming tasks from queue celery.
Known tasks:
  - cn.main.celery.TestTask

然后定义celery的任务调度程序test.py:

import celery

app = celery.Celery(broker="amqp://guest:[email protected]:5672//", backend="rpc://guest:[email protected]:5672//")

#调度任务格式为pacakage.class#func 
result = app.signature("cn.main.celery.TestTask#sum", [1, 2]).delay().get()

print("result is %s" % result)

运行test.py程序反应如下:

C:ProgramDataAnaconda3python.exe D:/Test/python/celerys/javatest.py
result is 3
Process finished with exit code 0

19:06:43.685 INFO  [pool-1-thread-1] c.g.c.CeleryWorkerCore:126 - CeleryTask cn.main.celery.TestTask#sum[2c1476a0-0d64-47c8-afce-6ddc99915ea9] succeeded in 338.9 μs

参考链接:

https://bitbucket.org/geneea/celery-java