# 創(chuàng)建虛擬環(huán)境
python3 -m venv env
source env/bin/activate

# 安裝依賴
pip install django celery redis django-celery-beat aliyun-python-sdk-core-v3 aliyun-python-sdk-mq mysqlclient

# 創(chuàng)建項(xiàng)目
django-admin startproject rocketmq_manager
cd rocketmq_manager

# 創(chuàng)建應(yīng)用
python manage.py startapp rocketmq

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME': 'rocketmq_manager',      # 數(shù)據(jù)庫(kù)名
        'USER': 'your_username',         # 用戶名
        'PASSWORD': 'your_password',     # 密碼
        'HOST': 'localhost',             # 主機(jī)
        'PORT': '3306',                  # 端口
        'OPTIONS': {
            'init_command': "SET sql_mode='STRICT_TRANS_TABLES'",
        },
    }
}

INSTALLED_APPS = [
    # ...
    'django_celery_beat',
    'django_celery_results',
    'rocketmq',
]

# Celery配置
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'django-db'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai'

# 阿里云配置（從環(huán)境變量獲取）
ALIYUN_ACCESS_KEY_ID = os.environ.get('ALIYUN_ACCESS_KEY_ID')
ALIYUN_ACCESS_KEY_SECRET = os.environ.get('ALIYUN_ACCESS_KEY_SECRET')
ALIYUN_REGION_ID = os.environ.get('ALIYUN_REGION_ID', 'cn-hangzhou')

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

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

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

@app.task(bind=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

from __future__ import absolute_import, unicode_literals
from .celery import app as celery_app

__all__ = ('celery_app',)

from django.db import models
from django.utils import timezone

class RocketMQInstance(models.Model):
    instance_id = models.CharField('實(shí)例ID', max_length=100, unique=True)
    instance_name = models.CharField('實(shí)例名稱', max_length=200, blank=True, null=True)
    instance_type = models.CharField('實(shí)例類型', max_length=50, blank=True, null=True)
    region_id = models.CharField('區(qū)域ID', max_length=50)
    status = models.CharField('狀態(tài)', max_length=50, blank=True, null=True)
    create_time = models.DateTimeField('創(chuàng)建時(shí)間', blank=True, null=True)
    expire_time = models.DateTimeField('過(guò)期時(shí)間', blank=True, null=True)
    tags = models.JSONField('標(biāo)簽', blank=True, null=True)
    last_updated = models.DateTimeField('最后更新時(shí)間', auto_now=True)
    
    def __str__(self):
        return f"{self.instance_name} ({self.instance_id})"
    
    class Meta:
        verbose_name = 'RocketMQ實(shí)例'
        verbose_name_plural = 'RocketMQ實(shí)例列表'
        indexes = [
            models.Index(fields=['instance_id', 'region_id']),
        ]

class InstanceSyncLog(models.Model):
    sync_time = models.DateTimeField('同步時(shí)間', auto_now_add=True)
    instance_count = models.IntegerField('實(shí)例數(shù)量', default=0)
    success = models.BooleanField('是否成功', default=True)
    error_message = models.TextField('錯(cuò)誤信息', blank=True, null=True)
    execution_time = models.FloatField('執(zhí)行時(shí)間(秒)', blank=True, null=True)
    
    def __str__(self):
        return f"同步記錄 - {self.sync_time}"
    
    class Meta:
        verbose_name = '實(shí)例同步日志'
        verbose_name_plural = '實(shí)例同步日志列表'
        ordering = ['-sync_time']

python manage.py makemigrations
python manage.py migrate

import os
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.acs_exception.exceptions import ClientException
from aliyunsdkcore.acs_exception.exceptions import ServerException
from aliyunsdkmq.model.v20190513 import DescribeInstancesRequest
import json
import time

class AliyunRocketMQClient:
    def __init__(self):
        self.access_key_id = os.environ.get('ALIYUN_ACCESS_KEY_ID')
        self.access_key_secret = os.environ.get('ALIYUN_ACCESS_KEY_SECRET')
        self.region_id = os.environ.get('ALIYUN_REGION_ID', 'cn-hangzhou')
        self.client = AcsClient(self.access_key_id, self.access_key_secret, self.region_id)
    
    def get_instances(self):
        try:
            request = DescribeInstancesRequest.DescribeInstancesRequest()
            request.set_accept_format('json')
            
            # 添加重試機(jī)制
            max_retries = 3
            for attempt in range(max_retries):
                try:
                    response = self.client.do_action_with_exception(request)
                    return json.loads(response)
                except (ClientException, ServerException) as e:
                    if attempt < max_retries - 1:
                        wait_time = (attempt + 1) * 2
                        print(f"請(qǐng)求失敗，{wait_time}秒后重試: {str(e)}")
                        time.sleep(wait_time)
                    else:
                        raise
            
        except Exception as e:
            print(f"獲取實(shí)例信息失敗: {str(e)}")
            raise

from celery import shared_task
from .models import RocketMQInstance, InstanceSyncLog
from .aliyun_client import AliyunRocketMQClient
import logging
from datetime import datetime
import time

logger = logging.getLogger(__name__)

@shared_task(bind=True, autoretry_for=(Exception,), retry_backoff=5, retry_kwargs={'max_retries': 3})
def sync_rocketmq_instances(self):
    start_time = time.time()
    try:
        client = AliyunRocketMQClient()
        response = client.get_instances()
        
        # 處理響應(yīng)數(shù)據(jù)
        instance_list = []
        if 'Data' in response and 'InstanceDoList' in response['Data']:
            for item in response['Data']['InstanceDoList']:
                instance = {
                    'instance_id': item.get('InstanceId', ''),
                    'instance_name': item.get('InstanceName', ''),
                    'instance_type': item.get('InstanceType', ''),
                    'region_id': item.get('RegionId', ''),
                    'status': item.get('InstanceStatus', ''),
                    'create_time': datetime.fromtimestamp(item.get('CreateTime', 0) / 1000) if item.get('CreateTime') else None,
                    'expire_time': datetime.fromtimestamp(item.get('ExpireTime', 0) / 1000) if item.get('ExpireTime') else None,
                    'tags': item.get('Tags', {})
                }
                instance_list.append(instance)
        
        # 使用事務(wù)批量更新數(shù)據(jù)庫(kù)
        from django.db import transaction
        with transaction.atomic():
            # 先刪除不存在的實(shí)例（可選）
            # existing_ids = [item['instance_id'] for item in instance_list]
            # RocketMQInstance.objects.exclude(instance_id__in=existing_ids).delete()
            
            # 批量更新或創(chuàng)建實(shí)例
            for instance_data in instance_list:
                RocketMQInstance.objects.update_or_create(
                    instance_id=instance_data['instance_id'],
                    defaults=instance_data
                )
        
        execution_time = time.time() - start_time
        
        # 記錄同步日志
        log = InstanceSyncLog.objects.create(
            instance_count=len(instance_list),
            success=True,
            execution_time=execution_time
        )
        
        logger.info(f"成功同步 {len(instance_list)} 個(gè)RocketMQ實(shí)例，耗時(shí): {execution_time:.2f}秒")
        return f"同步完成，共 {len(instance_list)} 個(gè)實(shí)例，耗時(shí): {execution_time:.2f}秒"
        
    except Exception as e:
        execution_time = time.time() - start_time
        
        # 記錄錯(cuò)誤日志
        InstanceSyncLog.objects.create(
            success=False,
            error_message=str(e),
            execution_time=execution_time
        )
        
        logger.error(f"同步RocketMQ實(shí)例失敗: {str(e)}，耗時(shí): {execution_time:.2f}秒")
        raise

from rest_framework import serializers
from .models import RocketMQInstance, InstanceSyncLog

class RocketMQInstanceSerializer(serializers.ModelSerializer):
    class Meta:
        model = RocketMQInstance
        fields = '__all__'
        read_only_fields = ['last_updated']

class InstanceSyncLogSerializer(serializers.ModelSerializer):
    class Meta:
        model = InstanceSyncLog
        fields = '__all__'
        read_only_fields = ['sync_time', 'instance_count', 'success', 'error_message', 'execution_time']

from rest_framework import viewsets, status
from rest_framework.response import Response
from .models import RocketMQInstance, InstanceSyncLog
from .serializers import RocketMQInstanceSerializer, InstanceSyncLogSerializer
from .tasks import sync_rocketmq_instances
from rest_framework.decorators import action
from rest_framework.permissions import IsAuthenticated
from rest_framework.authentication import TokenAuthentication

class RocketMQInstanceViewSet(viewsets.ModelViewSet):
    queryset = RocketMQInstance.objects.all()
    serializer_class = RocketMQInstanceSerializer
    authentication_classes = [TokenAuthentication]
    permission_classes = [IsAuthenticated]
    
    @action(detail=False, methods=['post'])
    def sync_now(self, request):
        """立即觸發(fā)實(shí)例同步"""
        task = sync_rocketmq_instances.delay()
        return Response({'task_id': task.id, 'message': '同步任務(wù)已啟動(dòng)'}, status=status.HTTP_202_ACCEPTED)
    
    @action(detail=False, methods=['get'])
    def regions(self, request):
        """獲取所有區(qū)域列表"""
        regions = RocketMQInstance.objects.values_list('region_id', flat=True).distinct()
        return Response(regions, status=status.HTTP_200_OK)

class InstanceSyncLogViewSet(viewsets.ReadOnlyModelViewSet):
    queryset = InstanceSyncLog.objects.all().order_by('-sync_time')
    serializer_class = InstanceSyncLogSerializer
    authentication_classes = [TokenAuthentication]
    permission_classes = [IsAuthenticated]

from django.urls import include, path
from rest_framework import routers
from .views import RocketMQInstanceViewSet, InstanceSyncLogViewSet

router = routers.DefaultRouter()
router.register(r'instances', RocketMQInstanceViewSet)
router.register(r'sync-logs', InstanceSyncLogViewSet)

urlpatterns = [
    path('', include(router.urls)),
]

from django.contrib import admin
from django.urls import path, include
from rest_framework.authtoken.views import obtain_auth_token

urlpatterns = [
    path('admin/', admin.site.urls),
    path('api/', include('rocketmq.urls')),
    path('api/token/', obtain_auth_token, name='api_token_auth'),  # 獲取認(rèn)證令牌
]

CELERY_BEAT_SCHEDULE = {
    'sync-rocketmq-instances': {
        'task': 'rocketmq.tasks.sync_rocketmq_instances',
        'schedule': 3600.0,  # 每小時(shí)執(zhí)行一次
        'args': ()
    },
}

export ALIYUN_ACCESS_KEY_ID=your_access_key_id
export ALIYUN_ACCESS_KEY_SECRET=your_access_key_secret
export ALIYUN_REGION_ID=cn-hangzhou  # 根據(jù)實(shí)際情況修改

redis-server

celery -A rocketmq_manager worker --loglevel=info --pool=prefork --concurrency=4

celery -A rocketmq_manager beat --loglevel=info --scheduler django_celery_beat.schedulers:DatabaseScheduler

python manage.py runserver

curl -X POST -d "username=your_username&password=your_password" http://localhost:8000/api/token/

curl -H "Authorization: Token your_token_here" http://localhost:8000/api/instances/

curl -H "Authorization: Token your_token_here" http://localhost:8000/api/sync-logs/

curl -X POST -H "Authorization: Token your_token_here" http://localhost:8000/api/instances/sync_now/

rocketmq_manager/
├── rocketmq_manager/
│   ├── __init__.py
│   ├── celery.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
├── rocketmq/
│   ├── migrations/
│   ├── __init__.py
│   ├── admin.py
│   ├── apps.py
│   ├── aliyun_client.py
│   ├── models.py
│   ├── serializers.py
│   ├── tasks.py
│   ├── urls.py
│   └── views.py
├── manage.py
└── db.sqlite3