优化MinIO list_objects_v2 操作的性能瓶颈与最佳实践

import boto3
import os

# 假设MinIO配置已通过环境变量或直接传入
# s3_client = boto3.client(
#     's3',
#     endpoint_url='http://localhost:9000',
#     aws_access_key_id='minioadmin',
#     aws_secret_access_key='minioadmin'
# )

# 示例：从环境变量获取配置
s3_client = boto3.client(
    's3',
    endpoint_url=os.getenv('MINIO_ENDPOINT', 'http://localhost:9000'),
    aws_access_key_id=os.getenv('MINIO_ACCESS_KEY', 'minioadmin'),
    aws_secret_access_key=os.getenv('MINIO_SECRET_KEY', 'minioadmin')
)

bucket_name = "my-large-bucket" # 假设此桶有40万对象

try:
    paginator = s3_client.get_paginator('list_objects_v2')
    page_iterator = paginator.paginate(Bucket=bucket_name)

    object_keys_count = 0
    print(f"开始列出存储桶 '{bucket_name}' 中的对象...")
    for page in page_iterator:
        keys = [obj['Key'] for obj in page.get('Contents', [])]
        object_keys_count += len(keys)
        # 在对象数量庞大时，每次迭代都可能非常慢，耗时数秒甚至数十秒
        print(f"已处理 {object_keys_count} 个对象...")
    print(f"总共列出对象: {object_keys_count} 个")

except Exception as e:
    print(f"列出对象时发生错误: {e}")

CREATE TABLE minio_objects (
    id SERIAL PRIMARY KEY,
    bucket_name VARCHAR(255) NOT NULL,
    object_key VARCHAR(1024) NOT NULL,
    size BIGINT,
    last_modified TIMESTAMP WITH TIME ZONE,
    etag VARCHAR(255),
    content_type VARCHAR(255),
    -- 可以添加其他自定义元数据字段
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    UNIQUE (bucket_name, object_key) -- 确保每个桶内的对象键唯一
);

-- 为常用查询字段创建索引以提高性能
CREATE INDEX idx_minio_objects_bucket_name ON minio_objects (bucket_name);
CREATE INDEX idx_minio_objects_object_key ON minio_objects (object_key);
CREATE INDEX idx_minio_objects_bucket_key ON minio_objects (bucket_name, object_key); -- 复合索引
CREATE INDEX idx_minio_objects_last_modified ON minio_objects (last_modified);

import boto3
import psycopg2 # 假设使用PostgreSQL
from datetime import datetime
import json
import os

# MinIO客户端配置
s3_client = boto3.client(
    's3',
    endpoint_url=os.getenv('MINIO_ENDPOINT', 'http://localhost:9000'),
    aws_access_key_id=os.getenv('MINIO_ACCESS_KEY', 'minioadmin'),
    aws_secret_access_key=os.getenv('MINIO_SECRET_KEY', 'minioadmin')
)

# 数据库客户端配置（示例，实际应用中应使用连接池）
def get_db_connection():
    return psycopg2.connect(
        host=os.getenv('DB_HOST', 'localhost'),
        database=os.getenv('DB_NAME', 'minio_metadata_db'),
        user=os.getenv('DB_USER', 'dbuser'),
        password=os.getenv('DB_PASSWORD', 'dbpassword')
    )

def upload_object_and_record_metadata(bucket, key, data, content_type='application/octet-stream', user_metadata=None):
    """上传对象到MinIO并记录元数据到数据库"""
    try:
        # 1. 上传到MinIO
        response = s3_client.put_object(
            Bucket=bucket,
            Key=key,
            Body=data,
            ContentType=content_type,
            Metadata=user_metadata or {}
        )

        # 2. 记录或更新到数据库
        with get_db_connection() as conn:
            with conn.cursor() as cur:
                cur.execute(
                    """
                    INSERT INTO minio_objects 
                    (bucket_name, object_key, size, last_modified, etag, content_type, updated_at)
                    VALUES (%s, %s, %s, %s, %s, %s, %s)
                    ON CONFLICT (bucket_name, object_key) DO UPDATE
                    SET size = EXCLUDED.size, 
                        last_modified = EXCLUDED.last_modified, 
                        etag = EXCLUDED.etag,
                        content_type = EXCLUDED.content_type,
                        updated_at = EXCLUDED.updated_at;
                    """,
                    (bucket, key, len(data), datetime.now(), response.get('ETag', '').strip('"'), content_type, datetime.now())
                )
            conn.commit()
        print(f"对象 '{key}' 已上传并元数据已记录。")
        return True
    except Exception as e:
        print(f"上传或记录元数据失败: {e}")
        return False

def delete_object_and_metadata(bucket, key):
    """从MinIO删除对象并从数据库移除元数据"""
    try:
        # 1. 从MinIO删除
        s3_client.delete_object(Bucket=bucket, Key=key)

        # 2. 从数据库删除
        with get_db_connection() as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "DELETE FROM minio_objects WHERE bucket_name = %s AND object_key = %s;",
                    (bucket, key)
                )
            conn.commit()
        print(f"对象 '{key}' 已删除。")
        return True
    except Exception as e:
        print(f"删除对象或元数据失败: {e}")
        return False

def get_all_object_keys_from_db(bucket):
    """从数据库高效获取指定桶的所有对象键"""
    keys = []
    try:
        with get_db_connection() as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT object_key FROM minio_objects WHERE bucket_name = %s ORDER BY object_key;",
                    (bucket,)
                )
                for row in cur.fetchall():
                    keys.append(row[0])
        print(f"从数据库获取到 {len(keys)} 个对象键。")
        return keys
    except Exception as e:
        print(f"从数据库获取对象键失败: {e}")
        return []

# 示例使用
if __name__ == "__main__":
    test_bucket = "my-tutorial-bucket"
    test_key_1 = "document/report_2023.pdf"
    test_key_2 = "image/logo.png"

    # 确保桶存在
    try:
        s3_client.create_bucket(Bucket=test_bucket)
        print(f"桶 '{test_bucket}' 已创建或已存在。")
    except Exception as e:
        if 'BucketAlreadyOwnedByYou' not in str(e):
            print(f"创建桶失败: {e}")

    # 上传对象并同步元数据
    upload_object_and_record_metadata(test_bucket, test_key_1, b"This is a test report content.", "application/pdf")
    upload_object_and_record_metadata(test_bucket, test_key_2, b"Image data here.", "image/png")

    # 从数据库获取对象键（高效操作）
    all_keys = get_all_object_keys_from_db(test_bucket)
    print("当前桶中的所有对象键 (从DB):", all_keys)

    # 删除对象并同步元数据
    delete_object_and_metadata(test_bucket, test_key_1)

    # 再次从数据库获取对象键
    all_keys_after_delete = get_all_object_keys_from_db(test_bucket)
    print("删除后桶中的所有对象键 (从DB):", all_keys_after_delete)