ksync

package
v0.4.13 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Apr 14, 2026 License: Apache-2.0 Imports: 4 Imported by: 1

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrChannelNotClosed   = errors.New("reset allowed only in closed state")
	ErrChannelClosed      = errors.New("send on closed channel")
	ErrGenerationOutdated = errors.New("generation outdated, send rejected")
	ErrChannelNil         = errors.New("channel is nil")
	ErrChannelFull        = errors.New("channel is full")
)
View Source 
var (
	ErrRingBufferSize = errors.New("ring buffer size must be a power of 2") // ErrRingBufferSize 无效的缓冲区大小错误
)
View Source 
var ErrWaitTimeout = errors.New("wait timeout")

定义超时错误

Functions

This section is empty.

Types

type CountDownLatch 

type CountDownLatch struct {
	// contains filtered or unexported fields
}

func NewCountDownLatch 

func NewCountDownLatch(delta int) *CountDownLatch

NewCountDownLatch 初始化一个指定计数的倒计时锁

func (*CountDownLatch) CountDown 

func (that *CountDownLatch) CountDown()

CountDown 减少计数器的值

func (*CountDownLatch) GetCount 

func (that *CountDownLatch) GetCount() int

GetCount 获取当前还有多少个任务没完成

func (*CountDownLatch) Wait 

func (that *CountDownLatch) Wait()

Wait 阻塞等待，直到计数器归零

func (*CountDownLatch) WaitWithTimeout 

func (that *CountDownLatch) WaitWithTimeout(timeout time.Duration) error

WaitWithTimeout 阻塞等待，直到计数器归零或超时

type LockedRingBuffer added in v0.4.11 

type LockedRingBuffer[T any] struct {
	// contains filtered or unexported fields
}

LockedRingBuffer 有锁版本的 SPSC Ring Buffer 使用 Mutex + Cond 实现高效阻塞，适合中低吞吐或需要规范阻塞语义的场景

func NewLockedRingBuffer added in v0.4.11 

func NewLockedRingBuffer[T any](size uint64) (*LockedRingBuffer[T], error)

func (*LockedRingBuffer[T]) AsyncDequeue added in v0.4.11 

func (that *LockedRingBuffer[T]) AsyncDequeue() (T, bool)

AsyncDequeue 非阻塞单条读取

func (*LockedRingBuffer[T]) AsyncDequeueBatch added in v0.4.11 

func (that *LockedRingBuffer[T]) AsyncDequeueBatch(max int) ([]T, int)

AsyncDequeueBatch 非阻塞批量读取

  • 参数

  • max: 最大读取数量

  • 返回

  • 读取到的数据

  • 读取到的数量

func (*LockedRingBuffer[T]) AsyncDequeueTo added in v0.4.11 

func (that *LockedRingBuffer[T]) AsyncDequeueTo(dst []T) int

AsyncDequeueTo 非阻塞地将数据拷贝到调用方提供的 dst 切片中 返回实际拷贝的元素数量（0 表示队列为空或 dst 长度为0） 这是非阻塞版本，不会等待

func (*LockedRingBuffer[T]) AsyncEnqueue added in v0.4.11 

func (that *LockedRingBuffer[T]) AsyncEnqueue(item T) bool

AsyncEnqueue 非阻塞单条写入

func (*LockedRingBuffer[T]) AsyncEnqueueBatch added in v0.4.11 

func (that *LockedRingBuffer[T]) AsyncEnqueueBatch(items []T) int

AsyncEnqueueBatch 非阻塞批量写入

func (*LockedRingBuffer[T]) Cap added in v0.4.11 

func (that *LockedRingBuffer[T]) Cap() uint64

func (*LockedRingBuffer[T]) Close added in v0.4.11 

func (that *LockedRingBuffer[T]) Close()

Close 关闭队列并唤醒所有等待者

func (*LockedRingBuffer[T]) Dequeue added in v0.4.11 

func (that *LockedRingBuffer[T]) Dequeue() (T, bool)

Dequeue 阻塞读取单个元素

func (*LockedRingBuffer[T]) DequeueBatch added in v0.4.11 

func (that *LockedRingBuffer[T]) DequeueBatch(max int) ([]T, int)

DequeueBatch 从环形缓冲区中批量取出元素

如果缓冲区中没有数据，会阻塞等待直到有数据可用、超时或缓冲区关闭。 最多读取 max 个元素，实际读取数量取决于缓冲区当前可用元素数量。

参数:

  • max: 最大读取数量，必须大于0

返回:

  • []T: 取出的元素数组
  • int: 实际取出的元素数量

func (*LockedRingBuffer[T]) DequeueBatchNoWait added in v0.4.13 

func (that *LockedRingBuffer[T]) DequeueBatchNoWait(max int) ([]T, int)

DequeueBatchNoWait 阻塞批量读取 如果无法获取锁，它会阻塞直到获取锁; 获取锁后，如果队列为空或已关闭，则返回 nil, 0; 否则，返回读取到的数据切片和实际读取的数量

  • 参数

  • max: 最大读取数量

  • 返回

  • 读取到的数据

  • 读取到的数量

func (*LockedRingBuffer[T]) DequeueNoWait added in v0.4.13 

func (that *LockedRingBuffer[T]) DequeueNoWait() (T, bool)

DequeueNoWait 阻塞单条读取 如果无法获取锁，它会阻塞直到获取锁; 获取锁后，如果队列为空或已关闭，则返回 (zero, false); 否则，返回读取到的数据和 true

func (*LockedRingBuffer[T]) DequeueTo added in v0.4.11 

func (that *LockedRingBuffer[T]) DequeueTo(dst []T) int

阻塞 DequeueTo, 将数据读取到 dst 中, 返回实际读取到的元素数量. 若缓冲区为空, 则继续阻塞直到数据可用、超时或缓冲区关闭. 若dst容量大于Buffer容量, 则一次性读取完Buffer中所有数据. 若dst容量小于等于Buffer容量, 则最多读取dst长度个元素.

func (*LockedRingBuffer[T]) DequeueToNoWait added in v0.4.13 

func (that *LockedRingBuffer[T]) DequeueToNoWait(dst []T) int

DequeueToNoWait 阻塞地将数据拷贝到调用方提供的 dst 切片中 如果无法获取锁，它会阻塞直到获取锁; 获取锁后，如果队列为空或已关闭，则返回 0; 否则，返回实际拷贝到 dst 的元素数量

func (*LockedRingBuffer[T]) Enqueue added in v0.4.11 

func (that *LockedRingBuffer[T]) Enqueue(item T) bool

Enqueue 阻塞写入单个元素，直到成功或超时

func (*LockedRingBuffer[T]) EnqueueBatch added in v0.4.11 

func (that *LockedRingBuffer[T]) EnqueueBatch(items []T) int

EnqueueBatch 阻塞批量写入，直到全部写入或超时

func (*LockedRingBuffer[T]) EnqueueBatchNoWait added in v0.4.13 

func (that *LockedRingBuffer[T]) EnqueueBatchNoWait(items []T) int

EnqueueBatchNoWait 阻塞批量写入 如果无法获取锁，它会阻塞直到获取锁; 获取锁后，如果队列已满，立即返回 0; 如果队列有空间，则写入数据并返回 实际写入的数量

func (*LockedRingBuffer[T]) EnqueueNoWait added in v0.4.13 

func (that *LockedRingBuffer[T]) EnqueueNoWait(item T) bool

EnqueueNoWait 阻塞单条写入 如果无法获取锁，它会阻塞直到获取锁; 获取锁后，如果队列已满，立即返回 false; 如果队列有空间，则写入数据并返回 true

func (*LockedRingBuffer[T]) IsClosed added in v0.4.11 

func (that *LockedRingBuffer[T]) IsClosed() bool

func (*LockedRingBuffer[T]) Len added in v0.4.11 

func (that *LockedRingBuffer[T]) Len() uint64

type RingBuffer added in v0.4.11 

type RingBuffer[T any] struct {
	// contains filtered or unexported fields
}

SPSC Ring Buffer - Single Producer Single Consumer

容量必须是 2 的幂次方: 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192, 16384, 32768, 65536

该 RingBuffer 严禁 用于多生产者或多消费者场景。如果有多个 Goroutine 同时调用 Enqueue，会导致 tail 指针竞争导致数据覆盖；若有多个 Goroutine 同时调用 Dequeue，则会导致同一份数据被读取多次。 只要严格遵守 SPSC 的使用约束

func NewRingBuffer added in v0.4.11 

func NewRingBuffer[T any](size uint64) (*RingBuffer[T], error)

NewRingBuffer 创建 RingBuffer，size 必须是 2 的幂

func (*RingBuffer[T]) AsyncDequeue added in v0.4.11 

func (that *RingBuffer[T]) AsyncDequeue() (T, bool)

Dequeue 非阻塞读取单个元素 返回 (item, true) 成功，(zero, false) 队列为空

func (*RingBuffer[T]) AsyncDequeueBatch added in v0.4.11 

func (that *RingBuffer[T]) AsyncDequeueBatch(max int) ([]T, int)

DequeueBatch 非阻塞批量读取 返回实际读取的数量（最多不超过 len(dst)）

func (*RingBuffer[T]) AsyncDequeueTo added in v0.4.11 

func (that *RingBuffer[T]) AsyncDequeueTo(dst []T) int

AsyncDequeueTo 非阻塞读取数据到提供的 dst 切片中 返回实际读取的数量

func (*RingBuffer[T]) AsyncEnqueue added in v0.4.11 

func (that *RingBuffer[T]) AsyncEnqueue(item T) bool

AsyncEnqueue 非阻塞写入单个元素 返回 true 表示成功，false 表示队列已满

func (*RingBuffer[T]) AsyncEnqueueBatch added in v0.4.11 

func (that *RingBuffer[T]) AsyncEnqueueBatch(items []T) int

AsyncEnqueueBatch 非阻塞批量写入 返回实际成功写入的数量（可能小于 len(items)）

func (*RingBuffer[T]) Cap added in v0.4.11 

func (that *RingBuffer[T]) Cap() uint64

func (*RingBuffer[T]) Dequeue added in v0.4.11 

func (that *RingBuffer[T]) Dequeue(timeout time.Duration) (T, bool)

DequeueBlocking 阻塞读取单个元素，直到成功或超时

func (*RingBuffer[T]) DequeueBatch added in v0.4.11 

func (that *RingBuffer[T]) DequeueBatch(max int, timeout time.Duration) ([]T, int)

DequeueBatchBlocking 阻塞批量读取，直到读到至少1个或超时 返回实际读取的数量

func (*RingBuffer[T]) DequeueTo added in v0.4.11 

func (that *RingBuffer[T]) DequeueTo(dst []T, timeout time.Duration) int

DequeueTo 阻塞读取数据到 dst 中，直到读到至少一个数据或超时

func (*RingBuffer[T]) Enqueue added in v0.4.11 

func (that *RingBuffer[T]) Enqueue(timeout time.Duration, item T) bool

Enqueue 阻塞写入单个元素，直到成功或超时 timeout <= 0 表示永久阻塞

func (*RingBuffer[T]) EnqueueBatch added in v0.4.11 

func (that *RingBuffer[T]) EnqueueBatch(timeout time.Duration, items []T) int

EnqueueBatchBlocking 阻塞批量写入，直到全部写入或超时 返回实际写入数量（超时情况下可能 < len(items)）

func (*RingBuffer[T]) IsEmpty added in v0.4.11 

func (that *RingBuffer[T]) IsEmpty() bool

func (*RingBuffer[T]) IsFull added in v0.4.11 

func (that *RingBuffer[T]) IsFull() bool

func (*RingBuffer[T]) Len added in v0.4.11 

func (that *RingBuffer[T]) Len() uint64

type SafeChannel added in v0.4.11 

type SafeChannel[T any] struct {
	// contains filtered or unexported fields
}

func NewSafeChannel added in v0.4.11 

func NewSafeChannel[T any](size int) *SafeChannel[T]

NewAtomicSafeChannel 创建高性能安全通道

func (*SafeChannel[T]) AsyncSend added in v0.4.11 

func (that *SafeChannel[T]) AsyncSend(data T) (err error)

AsyncSend 保持极致的原子性能 (无锁)，但为非阻塞, 可能返回 ErrChannelFull 错误。

func (*SafeChannel[T]) Ch added in v0.4.11 

func (that *SafeChannel[T]) Ch() (<-chan T, uint64)

Ch 获取原生 channel 和当前代数用于 select 或 range

返回: 原生 channel 和当前代数,代数在每次 Close 时递增

func (*SafeChannel[T]) Close added in v0.4.11 

func (that *SafeChannel[T]) Close() error

安全的Close函数, 确保只关闭一次，且是原子操作

func (*SafeChannel[T]) Reset added in v0.4.11 

func (that *SafeChannel[T]) Reset() error

Reset 仅在已关闭状态下工作 (无锁)

func (*SafeChannel[T]) Send added in v0.4.11 

func (that *SafeChannel[T]) Send(data T) (err error)

Send 保持极致的原子性能 (无锁)

func (*SafeChannel[T]) Status added in v0.4.11 

func (that *SafeChannel[T]) Status() bool

Status 检查是否已关闭 (原子操作，无锁)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.