利用Go语言构建高效分布式数据处理管道框架的实践与思考

package networkchannel

import (
    "encoding/gob"
    "fmt"
    "net"
    "sync"
    "time"
)

// NetWriter represents the writing end of a networked channel.
type NetWriter[T any] struct {
    conn net.Conn
    enc  *gob.Encoder
    mu   sync.Mutex // Protects conn and enc
}

// NewNetWriter creates a new NetWriter connected to a remote address.
func NewNetWriter[T any](addr string) (*NetWriter[T], error) {
    conn, err := net.Dial("tcp", addr)
    if err != nil {
        return nil, fmt.Errorf("failed to connect to %s: %w", addr, err)
    }
    return &NetWriter[T]{
        conn: conn,
        enc:  gob.NewEncoder(conn),
    }, nil
}

// Send sends data over the networked channel.
func (nw *NetWriter[T]) Send(data T) error {
    nw.mu.Lock()
    defer nw.mu.Unlock()
    return nw.enc.Encode(data)
}

// Close closes the network connection.
func (nw *NetWriter[T]) Close() error {
    nw.mu.Lock()
    defer nw.mu.Unlock()
    if nw.conn != nil {
        return nw.conn.Close()
    }
    return nil
}

// NetReader represents the reading end of a networked channel.
type NetReader[T any] struct {
    listener net.Listener
    incoming chan T
    mu       sync.Mutex // Protects listener
}

// NewNetReader creates a new NetReader listening on a given address.
func NewNetReader[T any](addr string, bufferSize int) (*NetReader[T], error) {
    listener, err := net.Listen("tcp", addr)
    if err != nil {
        return nil, fmt.Errorf("failed to listen on %s: %w", addr, err)
    }

    nr := &NetReader[T]{
        listener: listener,
        incoming: make(chan T, bufferSize),
    }

    go nr.acceptConnections()
    return nr, nil
}

// Recv returns the incoming channel for reading data.
func (nr *NetReader[T]) Recv() <-chan T {
    return nr.incoming
}

// Close closes the listener and incoming channel.
func (nr *NetReader[T]) Close() error {
    nr.mu.Lock()
    defer nr.mu.Unlock()
    if nr.listener != nil {
        err := nr.listener.Close()
        close(nr.incoming) // Close the channel when the reader is closed
        return err
    }
    return nil
}

func (nr *NetReader[T]) acceptConnections() {
    for {
        conn, err := nr.listener.Accept()
        if err != nil {
            // Check if listener was closed
            select {
            case <-time.After(10 * time.Millisecond): // Small delay to avoid busy-loop
                if nr.listener == nil { // Check again after a small delay
                    return // Listener closed, exit goroutine
                }
            default:
            }
            fmt.Printf("Error accepting connection: %v\n", err)
            continue
        }
        go nr.handleConnection(conn)
    }
}

func (nr *NetReader[T]) handleConnection(conn net.Conn) {
    defer conn.Close()
    dec := gob.NewDecoder(conn)
    for {
        var data T
        if err := dec.Decode(&data); err != nil {
            fmt.Printf("Error decoding data from %s: %v\n", conn.RemoteAddr(), err)
            return
        }
        nr.incoming <- data
    }
}

// --- Usage Example ---
func main() {
    // Start a reader (consumer)
    readerAddr := ":8080"
    reader, err := NewNetReader[string](readerAddr, 10)
    if err != nil {
        fmt.Printf("Failed to create reader: %v\n", err)
        return
    }
    defer reader.Close()

    go func() {
        for msg := range reader.Recv() {
            fmt.Printf("Consumer received: %s\n", msg)
        }
    }()
    fmt.Printf("Consumer listening on %s\n", readerAddr)

    // Give reader a moment to start
    time.Sleep(100 * time.Millisecond)

    // Start a writer (producer)
    writer, err := NewNetWriter[string](readerAddr)
    if err != nil {
        fmt.Printf("Failed to create writer: %v\n", err)
        return
    }
    defer writer.Close()

    fmt.Println("Producer sending messages...")
    writer.Send("Hello from Producer 1")
    writer.Send("Another message")

    // Simulate another producer
    writer2, err := NewNetWriter[string](readerAddr)
    if err != nil {
        fmt.Printf("Failed to create writer 2: %v\n", err)
        return
    }
    defer writer2.Close()
    writer2.Send("Message from Producer 2")

    time.Sleep(500 * time.Millisecond) // Wait for messages to be processed
    fmt.Println("Done.")
}