1. Nguyên lý cơ bản của tệp nhị phân

Khác với tệp văn bản — nơi dữ liệu được mã hóa theo chuẩn UTF-8 hoặc ASCII và đọc theo dòng — tệp nhị phân lưu trữ luồng byte thô, không qua giải mã. Việc xử lý chúng đòi hỏi thao tác trực tiếp trên từng byte hoặc cấu trúc nhị phân có định dạng rõ ràng.

2. Đọc/ghi cơ bản và tối ưu cho tệp lớn

Để tránh tràn bộ nhớ khi làm việc với tệp dung lượng cao (ví dụ >100 MB), nên sử dụng cơ chế đọc từng khối (chunk) thay vì nạp toàn bộ vào RAM:

# Ghi dữ liệu nhị phân với kích thước khối tùy chỉnh
OUTPUT_FILE = "output.bin"
BLOCK_SIZE = 8192  # 8 KB

with open(OUTPUT_FILE, "wb") as out_f:
    for i in range(1000):
        # Tạo một khối dữ liệu mẫu (4 byte số nguyên + 4 byte ký tự)
        payload = i.to_bytes(4, "big") + b"DATA"
        out_f.write(payload)

# Đọc tuần tự từng khối
with open(OUTPUT_FILE, "rb") as in_f:
    while True:
        block = in_f.read(BLOCK_SIZE)
        if not block:
            break
        # Xử lý block ở đây — ví dụ: kiểm tra checksum, nén, phân tích...
        handle_binary_block(block)

3. Truy cập ngẫu nhiên bằng mmap

Khi cần sửa đổi hoặc truy vấn các vùng cụ thể trong tệp rất lớn (ví dụ: cơ sở dữ liệu nhúng, tệp log có index), mmap cung cấp khả năng ánh xạ bộ nhớ — cho phép truy cập như mảng byte mà không tải toàn bộ tệp:

import mmap

def patch_at_offset(file_path: str, offset: int, new_bytes: bytes) -> None:
    with open(file_path, "r+b") as f:
        with mmap.mmap(f.fileno(), 0) as mm:
            mm[offset : offset + len(new_bytes)] = new_bytes

# Ví dụ: ghi đè 6 byte tại vị trí thứ 1000
patch_at_offset("archive.bin", 1000, b"\x01\x02\x03\x04\x05\x06")

4. Xử lý cấu trúc nhị phân có định dạng

Mô-đun struct cho phép đóng gói/giải nén dữ liệu theo định dạng nhị phân tiêu chuẩn (ví dụ: header mạng, giao thức tùy chỉnh). Dưới đây là hàm tiện ích để phân tích packet có header cố định:

import struct

# Định nghĩa layout: 1 byte version + 2 byte type + 4 byte length + 16 byte payload ID
HEADER_LAYOUT = ">BHI16s"
HEADER_SIZE = struct.calcsize(HEADER_LAYOUT)

def decode_packet(raw_bytes: bytes) -> dict:
    if len(raw_bytes) < HEADER_SIZE:
        raise ValueError("Dữ liệu không đủ độ dài header")
    
    version, pkt_type, pkt_len, payload_id = struct.unpack_from(HEADER_LAYOUT, raw_bytes)
    payload = raw_bytes[HEADER_SIZE : HEADER_SIZE + pkt_len]
    
    return {
        "version": version,
        "type": pkt_type,
        "length": pkt_len,
        "id": payload_id.strip(b"\x00").decode("ascii", errors="ignore"),
        "payload": payload
    }

# Sử dụng memoryview để tránh sao chép bộ nhớ khi xử lý tệp lớn
with open("packets.dat", "rb") as f:
    view = memoryview(f.read())
    packet_info = decode_packet(view[:512])  # Chỉ xử lý 512 byte đầu

5. Đảm bảo tính toàn vẹn và khôi phục tiến trình

Với các tác vụ kéo dài (ví dụ: xử lý tệp 50 GB), việc lưu trạng thái trung gian giúp tránh chạy lại từ đầu khi xảy ra lỗi hoặc gián đoạn:

import json
import os
from pathlib import Path

class ResumableBinaryProcessor:
    def __init__(self, source_path: str):
        self.source = Path(source_path)
        self.checkpoint = self.source.with_suffix(self.source.suffix + ".resume")
        self.offset = self._read_offset() or 0

    def _read_offset(self) -> int | None:
        if self.checkpoint.exists():
            try:
                return int(json.loads(self.checkpoint.read_text()).get("offset", 0))
            except (ValueError, OSError, KeyError):
                pass
        return 0

    def save_state(self, current_offset: int) -> None:
        self.checkpoint.write_text(json.dumps({"offset": current_offset}))

    def process_in_chunks(self, chunk_size: int = 65536) -> None:
        with self.source.open("rb") as f:
            f.seek(self.offset)
            while True:
                chunk = f.read(chunk_size)
                if not chunk:
                    break
                self._process_single_chunk(chunk)
                self.offset += len(chunk)
                self.save_state(self.offset)

    def _process_single_chunk(self, data: bytes) -> None:
        # Thay thế bằng logic xử lý thực tế: mã hóa, nén, phân tích...
        pass

# Khởi tạo và chạy
processor = ResumableBinaryProcessor("huge_payload.bin")
processor.process_in_chunks()

6. Tối ưu hiệu năng

Một số kỹ thuật nâng cao giúp cải thiện tốc độ I/O và giảm áp lực bộ nhớ:

• Bộ đệm tùy chỉnh: Thiết lập buffering lớn hơn mặc định (8192) để giảm số lần gọi hệ thống.
• Sử dụng io.BytesIO để xây dựng dữ liệu nhị phân trong bộ nhớ trước khi ghi đĩa.
• Tránh sao chép không cần thiết: Dùng memoryview, array.array hoặc numpy.ndarray với dtype=uint8.

import io
import array

# Xây dựng dữ liệu nhị phân trong bộ nhớ trước khi ghi
buffer = io.BytesIO()
for value in [100, 200, 300]:
    buffer.write(value.to_bytes(2, "little"))

# Chuyển thành bytes để ghi file
final_bytes = buffer.getvalue()
with open("packed.bin", "wb") as f:
    f.write(final_bytes)

7. Đo lường hiệu năng

Dưới đây là bảng so sánh thời gian và bộ nhớ tiêu thụ khi xử lý tệp 1 GB trên hệ thống tiêu chuẩn:

Các chiến lược được khuyến nghị: luôn dùng with để đảm bảo đóng tệp; ưu tiên đọc khối nhỏ cho tệp lớn; kết hợp struct hoặc numpy cho dữ liệu có cấu trúc; thêm kiểm tra checksum nếu yêu cầu độ tin cậy cao; và sử dụng cProfile để xác định điểm nghẽn khi cần tối ưu sâu.