JAVA 通过rabbitMQ调用celery 任务的代码示例

以下是通过 RabbitMQ 调用 Celery 任务的 Java 代码示例：

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

public class RabbitMQClient {

    private final static String QUEUE_NAME = "celery";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(QUEUE_NAME, true, consumer);

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" [x] Received '" + message + "'");
            // 调用 Celery 任务
            CeleryClient.runTask(message);
        }
    }
}

在这个示例中，我们使用 RabbitMQ 的 Java 客户端库创建了一个消费者，等待来自名为 "celery" 的队列的消息。当收到消息时，我们将其传递给 Celery 客户端的 runTask() 方法进行处理。

以下是 Celery 客户端的示例代码：

import io.github.cdimascio.dotenv.Dotenv;
import io.github.cdimascio.dotenv.DotenvBuilder;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class CeleryClient {

    private static final Logger logger = LoggerFactory.getLogger(CeleryClient.class);
    private static final String TASK_QUEUE_NAME = "celery";
    private static final String APP_NAME = "myapp";
    private static final String DEFAULT_EXCHANGE = "celery";
    private static final String DEFAULT_ROUTING_KEY = "celery";
    private static final String DEFAULT_BACKEND = "rpc://";
    private static final String DEFAULT_RESULT_EXPIRES = "3600";
    private static final String DEFAULT_TASK_TRACK_STARTED = "true";
    private static final String DEFAULT_TASK_TIME_LIMIT = "600";
    private static final String DEFAULT_TASK_SOFT_TIME_LIMIT = "600";

    private static final Dotenv dotenv = new DotenvBuilder().ignoreIfMissing().load();
    private static final String CELERY_BROKER_URL = dotenv.get("CELERY_BROKER_URL");
    private static final String CELERY_RESULT_BACKEND = dotenv.get("CELERY_RESULT_BACKEND");

    public static void runTask(String message) {
        try {
            JSONObject json = new JSONObject(message);
            String taskName = json.getString("task");
            JSONObject args = json.getJSONObject("args");

            ConnectionFactory factory = new ConnectionFactory();
            factory.setUri(CELERY_BROKER_URL);

            try (Connection connection = factory.newConnection()) {
                try (Channel channel = connection.createChannel()) {
                    String taskId = UUID.randomUUID().toString();

                    Map<String, Object> headers = new HashMap<>();
                    headers.put("task", taskName);
                    headers.put("id", taskId);
                    headers.put("app", APP_NAME);
                    headers.put("lang", "java");
                    headers.put("taskset", "celery");
                    headers.put("shadow", null);
                    headers.put("eta", null);
                    headers.put("expires", DEFAULT_RESULT_EXPIRES);
                    headers.put("group", null);
                    headers.put("retries", null);
                    headers.put("timelimit", new int[]{Integer.parseInt(DEFAULT_TASK_TIME_LIMIT), Integer.parseInt(DEFAULT_TASK_SOFT_TIME_LIMIT)});
                    headers.put("callbacks", null);
                    headers.put("errbacks", null);
                    headers.put("reply_to", null);
                    headers.put("correlation_id", null);
                    headers.put("delivery_info", null);
                    headers.put("kwargsrepr", "{}");
                    headers.put("utc", true);

                    channel.basicPublish(DEFAULT_EXCHANGE, DEFAULT_ROUTING_KEY, null, args.toString().getBytes("UTF-8"), headers);
                    logger.info("Task sent: " + taskName + " - " + taskId);
                }
            }
        } catch (Exception e) {
            logger.error("Error while sending task to Celery", e);
        }
    }
}

在这个示例中，我们使用 Celery Java 客户端库创建了一个方法，接收 RabbitMQ 消息并将其作为 Celery 任务发送。我们使用 dotenv 库从环境变量中读取 Celery 的 URL 和结果后端 URL，并使用这些值创建 Celery 连接。

注意，这里的 runTask() 方法只是将消息发送到 Celery，而实际任务的执行是在 Celery 服务器上完成的。因此，需要在 Celery 服务器上配置任务并启动 Celery 工作人员。