如何在Golang中构建异步文件IO 使用os.File.SetDeadline实现

os.File.SetDeadline

os.File.SetDeadline

package main

import (
    "fmt"
    "io"
    "os"
    "time"
)

// AsyncReadResult 异步读取的结果
type AsyncReadResult struct {
    Data  []byte
    Error error
}

// AsyncRead 异步读取文件
func AsyncRead(file *os.File, buffer []byte, timeout time.Duration) <-chan AsyncReadResult {
    resultChan := make(chan AsyncReadResult, 1)

    go func() {
        deadline := time.Now().Add(timeout)
        err := file.SetReadDeadline(deadline)
        if err != nil {
            resultChan <- AsyncReadResult{Error: fmt.Errorf("设置读取截止时间失败: %w", err)}
            return
        }

        n, err := file.Read(buffer)
        if err != nil && err != io.EOF {
            resultChan <- AsyncReadResult{Error: fmt.Errorf("读取文件失败: %w", err)}
            return
        }

        resultChan <- AsyncReadResult{Data: buffer[:n], Error: nil}
    }()

    return resultChan
}

func main() {
    // 创建一个临时文件
    tmpFile, err := os.CreateTemp("", "example")
    if err != nil {
        panic(err)
    }
    defer os.Remove(tmpFile.Name()) // 清理临时文件
    defer tmpFile.Close()

    // 写入一些数据
    _, err = tmpFile.WriteString("Hello, Asynchronous World!")
    if err != nil {
        panic(err)
    }

    // 重置文件指针到文件开头
    _, err = tmpFile.Seek(0, io.SeekStart)
    if err != nil {
        panic(err)
    }

    buffer := make([]byte, 32)
    timeout := 100 * time.Millisecond

    resultChan := AsyncRead(tmpFile, buffer, timeout)

    select {
    case result := <-resultChan:
        if result.Error != nil {
            fmt.Println("异步读取错误:", result.Error)
        } else {
            fmt.Println("异步读取数据:", string(result.Data))
        }
    case <-time.After(timeout * 2): // 增加一个额外的超时时间，以防goroutine阻塞
        fmt.Println("异步读取超时")
    }
}

os.File.SetDeadline

select

// 检查是否是超时错误
import (
    "net"
    "errors"
)

func isTimeoutError(err error) bool {
    if errors.Is(err, os.ErrDeadlineExceeded) {
        return true
    }

    // 兼容 net 包的超时错误
    netErr, ok := err.(net.Error)
    return ok && netErr.Timeout()
}

sync.Mutex

sync.RWMutex

package main

import (
    "fmt"
    "io"
    "os"
    "sync"
)

var (
    file *os.File
    mutex sync.Mutex
)

func init() {
    var err error
    file, err = os.OpenFile("data.txt", os.O_RDWR|os.O_CREATE, 0644)
    if err != nil {
        panic(err)
    }
}

func ReadFromFile(buffer []byte) (int, error) {
    mutex.Lock()
    defer mutex.Unlock()

    n, err := file.Read(buffer)
    return n, err
}

func WriteToFile(data []byte) (int, error) {
    mutex.Lock()
    defer mutex.Unlock()

    n, err := file.Write(data)
    return n, err
}

func main() {
    defer file.Close()

    // 启动多个 goroutine 来读取和写入文件
    var wg sync.WaitGroup

    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func(i int) {
            defer wg.Done()

            // 写入数据
            data := fmt.Sprintf("Goroutine %d: Hello!\n", i)
            _, err := WriteToFile([]byte(data))
            if err != nil {
                fmt.Printf("Goroutine %d: 写入错误: %v\n", i, err)
                return
            }

            // 读取数据
            buffer := make([]byte, 100)
            n, err := ReadFromFile(buffer)
            if err != nil && err != io.EOF {
                fmt.Printf("Goroutine %d: 读取错误: %v\n", i, err)
                return
            }
            fmt.Printf("Goroutine %d: 读取内容: %s", i, buffer[:n])
        }(i)
    }

    wg.Wait()
}