yyc-pipeline/fn-rabbitmq/src/main/java/net/yaoyi/pipeline/rabbitmq/RabbitConsumer.java

package net.yaoyi.pipeline.rabbitmq;

import com.aliyun.fc.runtime.Context;
import com.aliyun.fc.runtime.PojoRequestHandler;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import lombok.extern.slf4j.Slf4j;
import okhttp3.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * RabbitMQ 消费者 - 阿里云函数计算实现
 */
@Slf4j
public class RabbitConsumer implements PojoRequestHandler<Map<String, Object>, Void> {

    private static final String QUEUE_NAME = "vector.search.task.img.dul.check.queue";

    private static final String PIPELINE_API = "http://1861991840278303.eventbridge.cn-beijing.aliyuncs.com/webhook/putEvents?token=62ed79329bb74d0d8ede5cc5d0487a45897a6ff99f5d4d61946daf9f6e9c7fbc766b9d55ef2241edb2c2cb3aee651ac101409d2783f341aa8e8d27fdb9869d86";

    private static final OkHttpClient httpClient = new OkHttpClient.Builder()
            .connectTimeout(10, TimeUnit.SECONDS)
            .readTimeout(30, TimeUnit.SECONDS)
            .writeTimeout(30, TimeUnit.SECONDS)
            .build();
    private static final MediaType JSON = MediaType.get("application/json; charset=utf-8");

    @Override
    public Void handleRequest(Map<String, Object> input, Context context) {
        log.info("函数计算启动, Request ID: {}", context.getRequestId());

        try (Connection connection = RabbitMQHelper.createConnection();
             Channel channel = RabbitMQHelper.createChannel(connection, QUEUE_NAME)) {

            startConsumer(channel);
            log.info("RabbitMQ 消费者启动成功");

            Object val = input.get("waitSeconds");
            if (val instanceof Integer) {
                Thread.sleep(((int) val) * 1000);
            } else {
                Thread.sleep(2 * 60 * 1000);
            }
        } catch (Exception e) {
            log.error("启动 RabbitMQ 消费者失败", e);
            throw new RuntimeException("启动失败: " + e.getMessage(), e);
        }
        return null;
    }

    /**
     * 启动消费者
     */
    private void startConsumer(Channel channel) throws IOException {
        channel.basicQos(1);

        log.info("开始消费队列: {}", QUEUE_NAME);
        channel.basicConsume(QUEUE_NAME, false, (consumerTag, delivery) -> {
                    String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                    log.info("收到消息: {}", message);

                    try {
                        processMessage(message);
                        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                        log.info("消息处理成功并已确认");
                    } catch (Exception e) {
                        log.error("处理消息失败", e);
                        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
                        log.warn("消息已重新入队");
                    }
                },
                consumerTag -> log.info("消费者被取消: {}", consumerTag));
    }

    private void processMessage(String message) {
        RequestBody body = RequestBody.create(message, JSON);
        Request request = new Request.Builder().url(PIPELINE_API).put(body).build();

        try (Response response = httpClient.newCall(request).execute()) {
            if (response.isSuccessful()) {
                String responseBody = response.body() != null ? response.body().string() : "";
                log.info("Pipline API 请求成功:{}", responseBody);
            } else {
                log.info("Pipline API 请求失败:{}, body:{}", response.code(), response.body() != null ? response.body().string() : "");
            }
        } catch (Exception e) {
            log.error("Pipline API 请求异常", e);
        }
    }
}