MinIO 大规模对象存储 list_objects_v2 性能瓶颈与解决方案

import boto3

# 假设 s3_client 已经初始化，并连接到 MinIO
# s3_client = boto3.client('s3',
#                          endpoint_url='http://your-minio-server:9000',
#                          aws_access_key_id='minioadmin',
#                          aws_secret_access_key='minioadmin')

bucket_name = "my-large-bucket"

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

    total_keys = 0
    for page in page_iterator:
        keys = [obj['Key'] for obj in page.get('Contents', [])]
        total_keys += len(keys)
        # 在这里处理获取到的 keys
        print(f"Processed {len(keys)} keys in this page. Total: {total_keys}")
        # ... 进一步处理 keys ...

    print(f"Finished listing all {total_keys} objects.")

except Exception as e:
    print(f"An error occurred: {e}")

import psycopg2 # 假设使用PostgreSQL作为外部数据库
from datetime import datetime

class ObjectMetadataManager:
    def __init__(self, db_config):
        self.conn = psycopg2.connect(**db_config)
        self._create_table_if_not_exists()

    def _create_table_if_not_exists(self):
        with self.conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS object_metadata (
                    id SERIAL PRIMARY KEY,
                    bucket_name VARCHAR(255) NOT NULL,
                    object_key VARCHAR(1024) NOT NULL,
                    size BIGINT,
                    last_modified TIMESTAMP,
                    etag VARCHAR(255),
                    UNIQUE (bucket_name, object_key)
                );
                CREATE INDEX IF NOT EXISTS idx_bucket_key ON object_metadata (bucket_name, object_key);
            """)
            self.conn.commit()

    def add_object_metadata(self, bucket_name, object_key, size, last_modified, etag):
        with self.conn.cursor() as cur:
            cur.execute("""
                INSERT INTO object_metadata (bucket_name, object_key, size, last_modified, etag)
                VALUES (%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;
            """, (bucket_name, object_key, size, last_modified, etag))
            self.conn.commit()

    def remove_object_metadata(self, bucket_name, object_key):
        with self.conn.cursor() as cur:
            cur.execute("""
                DELETE FROM object_metadata WHERE bucket_name = %s AND object_key = %s;
            """, (bucket_name, object_key))
            self.conn.commit()

    def get_object_keys(self, bucket_name, prefix=None, limit=1000, offset=0):
        """
        从数据库中查询指定桶和前缀下的对象键
        """
        query = "SELECT object_key FROM object_metadata WHERE bucket_name = %s"
        params = [bucket_name]

        if prefix:
            query += " AND object_key LIKE %s"
            params.append(f"{prefix}%")

        query += " ORDER BY object_key LIMIT %s OFFSET %s;"
        params.extend([limit, offset])

        with self.conn.cursor() as cur:
            cur.execute(query, params)
            return [row[0] for row in cur.fetchall()]

# 示例使用
db_config = {
    "host": "localhost",
    "database": "minio_metadata",
    "user": "your_user",
    "password": "your_password"
}

# 初始化元数据管理器
metadata_manager = ObjectMetadataManager(db_config)

# 模拟对象上传时更新元数据
# metadata_manager.add_object_metadata("my-large-bucket", "data/file1.csv", 1024, datetime.now(), "etag123")
# metadata_manager.add_object_metadata("my-large-bucket", "data/file2.csv", 2048, datetime.now(), "etag456")
# metadata_manager.add_object_metadata("my-large-bucket", "images/pic1.jpg", 5000, datetime.now(), "etag789")

# 从数据库获取对象键，支持分页和前缀
bucket_name = "my-large-bucket"
keys_page_1 = metadata_manager.get_object_keys(bucket_name, prefix="data/", limit=100, offset=0)
print(f"Retrieved {len(keys_page_1)} keys from DB (page 1, prefix 'data/'): {keys_page_1}")

all_keys_from_db = metadata_manager.get_object_keys(bucket_name, limit=1000000) # 获取所有键
print(f"Retrieved total {len(all_keys_from_db)} keys from external DB.")