Django - Google cloud - pubsub - celery - kubernetes问题

pip install google-cloud-pubsub

# signals.py
from django.dispatch import receiver
from google.cloud import pubsub_v1
from google.oauth2 import service_account
import os

project_id = os.environ.get('GOOGLE_CLOUD_PROJECT')
subscription_id = os.environ.get('PUBSUB_SUBSCRIPTION_ID')
credentials = service_account.Credentials.from_service_account_file(
    os.environ.get('GOOGLE_APPLICATION_CREDENTIALS'))

subscriber = pubsub_v1.SubscriberClient(credentials=credentials)

@receiver(your_signal)
def callback(message):
    print(f"Received message: {message}")
    # Process the message here

streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
print(f"Listening for messages on {subscription_path}..\n")

# Wrap subscriber in a 'with' block to automatically call close() when done.
with subscriber:
    try:
        # When timeout is not set, result() will block indefinitely,
        # unless an exception is encountered first.
        streaming_pull_future.result()
    except TimeoutError:
        streaming_pull_future.cancel()

pip install celery

# celery.py
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project_name.settings')

app = Celery('your_project_name')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

# __init__.py
from __future__ import absolute_import, unicode_literals

# This will make sure the app is always imported when Django starts so that
# shared_task will use this app.
from .celery import app as celery_app

__all__ = ('celery_app',)

# your_app/tasks.py
from celery import shared_task

@shared_task
def process_message(message):
    # Process the message here
    pass

apiVersion: apps/v1
kind: Deployment
metadata:
  name: django-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: django-app
  template:
    metadata:
      labels:
        app: django-app
    spec:
      containers:
        - name: django-app
          image: your_django_image
          ports:
            - containerPort: 8000
          env:
            - name: GOOGLE_CLOUD_PROJECT
              valueFrom:
                fieldRef:
                  fieldPath: metadata.annotations['google.cloud.project-id']
            - name: PUBSUB_SUBSCRIPTION_ID
              valueFrom:
                fieldRef:
                  fieldPath: metadata.annotations['google.cloud.pubsub.subscription-id']
            - name: GOOGLE_APPLICATION_CREDENTIALS
              value: /app/gcp-credentials.json
          volumeMounts:
            - name: gcp-credentials
              mountPath: /app/gcp-credentials.json
              subPath: gcp-credentials.json
      volumes:
        - name: gcp-credentials
          secret:
            secretName: gcp-credentials

apiVersion: apps/v1
kind: Deployment
metadata:
  name: celery-worker
spec:
  replicas: 3
  selector:
    matchLabels:
      app: celery-worker
  template:
    metadata:
      labels:
        app: celery-worker
    spec:
      containers:
        - name: celery-worker
          image: your_celery_image
          command: ["celery", "-A", "your_project_name", "worker", "--loglevel=info"]
          env:
            - name: DJANGO_SETTINGS_MODULE
              value: "your_project_name.settings"
          volumeMounts:
            - name: gcp-credentials
              mountPath: /app/gcp-credentials.json
              subPath: gcp-credentials.json
      volumes:
        - name: g'dp-credentials
          secret:
            secretName: gcp-credentials

apiVersion: v1
kind: Service
metadata:
  name: django-app-service
spec:
  selector:
    app: django-app
  ports:
    - protocol: TCP
      port: 80
      targetPort: 8000
  type: LoadBalancer

apiVersion: v1
kind: Service
metadata:
  name: celery-worker-service
spec:
  selector:
    app: celery-worker
  ports:
    - protocol: TCP
      port: 5672
      targetPort: 5672

apiVersion: v1
kind: ConfigMap
metadata:
  name: django-settings
data:
  settings.py: |
    # Your Django settings here

kubectl apply -f path/to/your/kubernetes/configurations