Cloud Run 服务稳定性深度解析：理解实例重启与调度任务的最佳实践

// 假设这是你的调度服务的一部分，负责将任务推送到Pub/Sub
import com.google.api.core.ApiFuture;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.ProjectTopicName;
import com.google.pubsub.v1.PubsubMessage;

import java.io.IOException;
import java.util.concurrent.ExecutionException;

public class FootballMatchTaskPublisher {

    private final String projectId;
    private final String topicId;
    private Publisher publisher;

    public FootballMatchTaskPublisher(String projectId, String topicId) {
        this.projectId = projectId;
        this.topicId = topicId;
        try {
            this.publisher = Publisher.newBuilder(ProjectTopicName.of(projectId, topicId)).build();
        } catch (IOException e) {
            System.err.println("Failed to create Pub/Sub publisher: " + e.getMessage());
            // Handle error appropriately
        }
    }

    public void publishMonitoringTask(String matchId, String monitorConfigJson) {
        if (publisher == null) {
            System.err.println("Publisher not initialized.");
            return;
        }
        try {
            ByteString data = ByteString.copyFromUtf8(monitorConfigJson);
            PubsubMessage pubsubMessage = PubsubMessage.newBuilder()
                    .setData(data)
                    .putAttributes("match_id", matchId)
                    .build();

            ApiFuture<String> messageIdFuture = publisher.publish(pubsubMessage);
            String messageId = messageIdFuture.get(); // Blocks until message is published
            System.out.println("Published monitoring task for match " + matchId + " with ID: " + messageId);
        } catch (InterruptedException | ExecutionException e) {
            System.err.println("Error publishing message for match " + matchId + ": " + e.getMessage());
            Thread.currentThread().interrupt();
        }
    }

    public void shutdown() {
        if (publisher != null) {
            try {
                publisher.shutdown();
                publisher.awaitTermination(1, java.util.concurrent.TimeUnit.MINUTES);
            } catch (InterruptedException e) {
                System.err.println("Publisher shutdown interrupted: " + e.getMessage());
                Thread.currentThread().interrupt();
            }
        }
    }
}

// 部署到Cloud Run的服务，接收Pub/Sub推送的消息
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

import java.util.Base64;
import java.util.Map;

@SpringBootApplication
@RestController
public class FootballMatchMonitorService {

    public static void main(String[] args) {
        SpringApplication.run(FootballMatchMonitorService.class, args);
    }

    @PostMapping("/pubsub/push")
    public ResponseEntity<String> receiveMessage(@RequestBody PubSubMessageWrapper messageWrapper) {
        try {
            String data = new String(Base64.getDecoder().decode(messageWrapper.getMessage().getData()));
            Map<String, String> attributes = messageWrapper.getMessage().getAttributes();
            String matchId = attributes.get("match_id");

            System.out.println("Received monitoring task for match ID: " + matchId + ", config: " + data);

            // TODO: 在这里实现具体的足球比赛监控逻辑
            // 例如：调用第三方API获取比赛数据，分析，存储结果等
            performMonitoring(matchId, data);

            // 成功处理消息，返回200 OK，Pub/Sub将确认消息
            return ResponseEntity.ok("Message processed successfully");
        } catch (Exception e) {
            System.err.println("Error processing Pub/Sub message: " + e.getMessage());
            // 返回非2xx状态码，Pub/Sub将重试消息
            return ResponseEntity.status(500).body("Error processing message");
        }
    }

    private void performMonitoring(String matchId, String monitorConfig) {
        System.out.println("Executing monitoring for match: " + matchId + " with config: " + monitorConfig);
        // 模拟监控工作
        try {
            Thread.sleep(2000); // 模拟耗时操作
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println("Monitoring for match " + matchId + " completed.");
    }

    // Helper classes for Pub/Sub push message structure
    // (实际结构可能因Pub/Sub版本和Spring版本略有不同，但这是常见模式)
    public static class PubSubMessageWrapper {
        private PubSubMessage message;
        // 其他字段如 subscription 可以根据需要添加

        public PubSubMessage getMessage() { return message; }
        public void setMessage(PubSubMessage message) { this.message = message; }
    }

    public static class PubSubMessage {
        private String data; // Base64 encoded
        private Map<String, String> attributes;
        private String messageId;
        private String publishTime;

        public String getData() { return data; }
        public void setData(String data) { this.data = data; }
        public Map<String, String> getAttributes() { return attributes; }
        public void setAttributes(Map<String, String> attributes) { this.attributes = attributes; }
        public String getMessageId() { return messageId; }
        public void setMessageId(String messageId) { this.messageId = messageId; }
        public String getPublishTime() { return publishTime; }
        public void setPublishTime(String publishTime) { this.publishTime = publishTime; }
    }
}

gcloud run deploy football-monitor-worker \
  --image gcr.io/your-project-id/football-monitor-worker \
  --region us-central1 \
  --allow-unauthenticated # 或者配置适当的认证