Documentation
¶
Index ¶
- type ConcurrentAutoDelSet
- type ConcurrentHashMapSimple
- func (m *ConcurrentHashMapSimple[K, V]) Contains(key K) bool
- func (m *ConcurrentHashMapSimple[K, V]) Delete(key K) bool
- func (m *ConcurrentHashMapSimple[K, V]) Get(key K) (V, bool)
- func (m *ConcurrentHashMapSimple[K, V]) Len() int
- func (m *ConcurrentHashMapSimple[K, V]) Range(f func(key K, value V) bool)
- func (m *ConcurrentHashMapSimple[K, V]) Set(key K, value V)
- func (m *ConcurrentHashMapSimple[K, V]) SetIfAbsent(key K, value V) bool
- type ConcurrentHashSet
- func (s *ConcurrentHashSet) Add(value interface{})
- func (s *ConcurrentHashSet) AddAll(values ...interface{})
- func (s *ConcurrentHashSet) AddIfNotExists(value interface{}) bool
- func (s *ConcurrentHashSet) Clear()
- func (s *ConcurrentHashSet) Contains(value interface{}) bool
- func (s *ConcurrentHashSet) ContainsAny(values []interface{}) bool
- func (s *ConcurrentHashSet) Length() int
- func (s *ConcurrentHashSet) Remove(value interface{})
- func (s *ConcurrentHashSet) ToSlice() []interface{}
- func (s *ConcurrentHashSet) ToString() string
- type FlowMonitor
- type HashSet
- func (s *HashSet) Add(value interface{})
- func (s *HashSet) AddAll(values ...interface{})
- func (s *HashSet) Clear()
- func (s *HashSet) Contains(value interface{}) bool
- func (s *HashSet) ContainsAny(values []interface{}) bool
- func (s *HashSet) Length() int
- func (s *HashSet) Remove(value interface{})
- func (s *HashSet) ToSlice() []interface{}
- func (s *HashSet) ToString() string
- type PercentLimiter
- type SafeChannel
- func (c *SafeChannel[T]) Close()
- func (c *SafeChannel[T]) CloseAndGetRest() []T
- func (c *SafeChannel[T]) CloseAndRelease()
- func (c *SafeChannel[T]) GetCapacity() int
- func (c *SafeChannel[T]) GetSize() int
- func (c *SafeChannel[T]) GetTailElem() T
- func (c *SafeChannel[T]) IsAlive() bool
- func (c *SafeChannel[T]) NonBlockPopup() (T, bool)
- func (c *SafeChannel[T]) Popup() T
- func (c *SafeChannel[T]) PopupWithTimeout(timeoutMs int64) (value T, isValidValue bool)
- func (c *SafeChannel[T]) PopupWithTimeoutAndCancel(timeoutMs int64, isEnd *atomic.Bool, timeoutCallback func(), ...) (value T, isValidValue bool)
- func (c *SafeChannel[T]) Push(value T) bool
- func (c *SafeChannel[T]) Push2Head(value T) bool
- func (c *SafeChannel[T]) PushForce(value T) bool
- func (c *SafeChannel[T]) SignalOnce()
- type SafeWaitGroup
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type ConcurrentAutoDelSet ¶
type ConcurrentAutoDelSet struct {
// contains filtered or unexported fields
}
func NewConcurrentAutoDelSet ¶
func NewConcurrentAutoDelSet() *ConcurrentAutoDelSet
func (*ConcurrentAutoDelSet) Contains ¶
func (s *ConcurrentAutoDelSet) Contains(item interface{}) bool
func (*ConcurrentAutoDelSet) PutAndDelWithDelay ¶
func (s *ConcurrentAutoDelSet) PutAndDelWithDelay(item interface{}, delDurationMs int)
func (*ConcurrentAutoDelSet) Remove ¶
func (s *ConcurrentAutoDelSet) Remove(item interface{})
type ConcurrentHashMapSimple ¶
type ConcurrentHashMapSimple[K comparable, V any] struct { // contains filtered or unexported fields }
func NewConcurrentHashMapSimple ¶
func NewConcurrentHashMapSimple[K comparable, V any]() *ConcurrentHashMapSimple[K, V]
func (*ConcurrentHashMapSimple[K, V]) Contains ¶
func (m *ConcurrentHashMapSimple[K, V]) Contains(key K) bool
func (*ConcurrentHashMapSimple[K, V]) Delete ¶
func (m *ConcurrentHashMapSimple[K, V]) Delete(key K) bool
func (*ConcurrentHashMapSimple[K, V]) Get ¶
func (m *ConcurrentHashMapSimple[K, V]) Get(key K) (V, bool)
func (*ConcurrentHashMapSimple[K, V]) Len ¶
func (m *ConcurrentHashMapSimple[K, V]) Len() int
func (*ConcurrentHashMapSimple[K, V]) Range ¶
func (m *ConcurrentHashMapSimple[K, V]) Range(f func(key K, value V) bool)
func (*ConcurrentHashMapSimple[K, V]) Set ¶
func (m *ConcurrentHashMapSimple[K, V]) Set(key K, value V)
func (*ConcurrentHashMapSimple[K, V]) SetIfAbsent ¶
func (m *ConcurrentHashMapSimple[K, V]) SetIfAbsent(key K, value V) bool
type ConcurrentHashSet ¶
ConcurrentHashSet is a concurrent hash set data structure. Note: when using the Add method, do not use map, slice, or func as element types, since they are not comparable and will cause a runtime panic.
func NewConcurrentHashSet ¶
func NewConcurrentHashSet() *ConcurrentHashSet
NewConcurrentHashSet creates and returns an empty ConcurrentHashSet.
func NewConcurrentHashSetByInit ¶
func NewConcurrentHashSetByInit(data ...interface{}) *ConcurrentHashSet
NewConcurrentHashSetByInit creates a ConcurrentHashSet pre-populated with the given values.
func (*ConcurrentHashSet) Add ¶
func (s *ConcurrentHashSet) Add(value interface{})
Add inserts a value into the set in a thread-safe manner.
func (*ConcurrentHashSet) AddAll ¶
func (s *ConcurrentHashSet) AddAll(values ...interface{})
AddAll batch adds all given values into the set in a thread-safe manner.
func (*ConcurrentHashSet) AddIfNotExists ¶
func (s *ConcurrentHashSet) AddIfNotExists(value interface{}) bool
AddIfNotExists adds the value only if it does not already exist. Returns true if the addition was successful, false otherwise.
func (*ConcurrentHashSet) Clear ¶
func (s *ConcurrentHashSet) Clear()
Clear removes all elements from the set in a thread-safe manner.
func (*ConcurrentHashSet) Contains ¶
func (s *ConcurrentHashSet) Contains(value interface{}) bool
Contains reports whether the set contains the given value in a thread-safe manner.
func (*ConcurrentHashSet) ContainsAny ¶
func (s *ConcurrentHashSet) ContainsAny(values []interface{}) bool
ContainsAny reports whether the set contains any of the given values in a thread-safe manner.
func (*ConcurrentHashSet) Length ¶
func (s *ConcurrentHashSet) Length() int
Length returns the number of elements in the set in a thread-safe manner.
func (*ConcurrentHashSet) Remove ¶
func (s *ConcurrentHashSet) Remove(value interface{})
Remove deletes the given value from the set in a thread-safe manner.
func (*ConcurrentHashSet) ToSlice ¶
func (s *ConcurrentHashSet) ToSlice() []interface{}
ToSlice returns all elements of the set as a slice in a thread-safe manner.
func (*ConcurrentHashSet) ToString ¶
func (s *ConcurrentHashSet) ToString() string
ToString returns a string representation of the set in a thread-safe manner.
type FlowMonitor ¶
type FlowMonitor struct {
// contains filtered or unexported fields
}
FlowMonitor Essentially a rate limiter combined with dynamic traffic monitoring. Rate limiter: when only 20% of the traffic is desired, 80% of the traffic is dropped directly (this limiting is affected solely by the wantedRate value). Dynamic traffic monitoring: tracks the dynamic per-second traffic (average), the previous per-second traffic (lastAverage), and the lifecycle-wide per-second average (allAverage). Note: the granularity is 1 second, and dropping is based on the average.
func NewFlowMonitor ¶
func NewFlowMonitor(errorThreshold int64) *FlowMonitor
NewFlowMonitor creates a new FlowMonitor with the given error threshold. A negative threshold is clamped to the default value of 3.
func (*FlowMonitor) Count ¶
func (m *FlowMonitor) Count()
Count Directly updates: lastCalculateStartTime, variableNum, average, allNum. Indirectly updates: see calculateAverage().
func (*FlowMonitor) GetAllAverage ¶
func (m *FlowMonitor) GetAllAverage() int64
GetAllAverage returns the lifecycle-wide per-second average of counted elements.
func (*FlowMonitor) GetAllNum ¶
func (m *FlowMonitor) GetAllNum() int64
GetAllNum returns the total number of elements counted over the lifecycle.
func (*FlowMonitor) GetAverage ¶
func (m *FlowMonitor) GetAverage() int64
GetAverage returns the current per-second average of counted elements.
func (*FlowMonitor) GetLastAverage ¶
func (m *FlowMonitor) GetLastAverage() int64
GetLastAverage returns the previous per-second average of counted elements.
func (*FlowMonitor) Start ¶
func (m *FlowMonitor) Start()
Start launches the background average calculation and records the start time of the monitor.
func (*FlowMonitor) ToString ¶
func (m *FlowMonitor) ToString() string
ToString Returns a snapshot of the FlowMonitor's current status, using human-readable Chinese field names rather than the raw internal variable names.
type HashSet ¶
type HashSet struct {
// contains filtered or unexported fields
}
HashSet is a hash set data structure. Note: when using the Add method, do not use map, slice, or func as element types, since they are not comparable and will cause a runtime panic.
func NewHashSetByInit ¶
func NewHashSetByInit(data ...interface{}) *HashSet
NewHashSetByInit creates a HashSet pre-populated with the given values.
func (*HashSet) Contains ¶
Contains reports whether the set contains the given value.
func (*HashSet) ContainsAny ¶
ContainsAny reports whether the set contains any of the given values.
func (*HashSet) Length ¶
Length returns the number of elements in the set.
func (*HashSet) Remove ¶
func (s *HashSet) Remove(value interface{})
Remove deletes the given value from the set.
type PercentLimiter ¶
func NewPercentLimiter ¶
func NewPercentLimiter[T any](capacity int, wantedRate float64) *PercentLimiter[T]
func (*PercentLimiter[T]) GetBatchDataByThreshold ¶
func (l *PercentLimiter[T]) GetBatchDataByThreshold(threshold int) ([]T, bool)
func (*PercentLimiter[T]) GetBufferChannel ¶
func (l *PercentLimiter[T]) GetBufferChannel() *SafeChannel[T]
func (*PercentLimiter[T]) GetByTimeout ¶
func (l *PercentLimiter[T]) GetByTimeout(timeoutMs int64) (value T, isValidValue bool)
func (*PercentLimiter[T]) Push ¶
func (l *PercentLimiter[T]) Push(value T) bool
type SafeChannel ¶
SafeChannel is a thread-safe, capacity-bounded FIFO channel with condition-variable-based blocking. It supports both tail and head insertion as well as blocking and non-blocking pop operations.
func NewSafeChannel ¶
func NewSafeChannel[T any](capacity int) *SafeChannel[T]
NewSafeChannel creates and returns a new SafeChannel with the given capacity.
func (*SafeChannel[T]) Close ¶
func (c *SafeChannel[T]) Close()
Close marks the channel as closed and releases all blocked goroutines.
func (*SafeChannel[T]) CloseAndGetRest ¶
func (c *SafeChannel[T]) CloseAndGetRest() []T
CloseAndGetRest closes the channel and returns all remaining elements.
func (*SafeChannel[T]) CloseAndRelease ¶
func (c *SafeChannel[T]) CloseAndRelease()
CloseAndRelease closes the channel and drains all remaining elements, releasing the underlying resources.
func (*SafeChannel[T]) GetCapacity ¶
func (c *SafeChannel[T]) GetCapacity() int
GetCapacity returns the total capacity of the channel.
func (*SafeChannel[T]) GetSize ¶
func (c *SafeChannel[T]) GetSize() int
GetSize returns the current number of elements in the channel.
func (*SafeChannel[T]) GetTailElem ¶
func (c *SafeChannel[T]) GetTailElem() T
func (*SafeChannel[T]) IsAlive ¶
func (c *SafeChannel[T]) IsAlive() bool
IsAlive returns whether the channel is still active (not closed).
func (*SafeChannel[T]) NonBlockPopup ¶
func (c *SafeChannel[T]) NonBlockPopup() (T, bool)
NonBlockPopup get value from channel return zero value if channel is empty
func (*SafeChannel[T]) Popup ¶
func (c *SafeChannel[T]) Popup() T
Popup get value from channel block until channel is not empty
func (*SafeChannel[T]) PopupWithTimeout ¶
func (c *SafeChannel[T]) PopupWithTimeout(timeoutMs int64) (value T, isValidValue bool)
PopupWithTimeout must be used together with SignalOnce. Otherwise, it may block forever if no other operation wakes up the waiting goroutine.
----------------WARNING--------------- If you use PopupWithTimeout, make sure to use it together with SignalOnce. Usage example:
go func(){
time.Sleep(3 * time.Second)
safeChannel.SignalOnce()
}()
v,isOK := safeChannel.PopupWithTimeout(2800)
Note: ideally the "actual expected timeout" is the value used in Sleep, while the value passed to PopupWithTimeout should be slightly smaller than the "actual expected timeout".
func (*SafeChannel[T]) PopupWithTimeoutAndCancel ¶
func (c *SafeChannel[T]) PopupWithTimeoutAndCancel(timeoutMs int64, isEnd *atomic.Bool, timeoutCallback func(), cancelCallback func()) (value T, isValidValue bool)
PopupWithTimeoutAndCancel pops a value within the given timeout, with support for an external cancel signal. It invokes timeoutCallback on timeout and cancelCallback when the isEnd flag is set.
func (*SafeChannel[T]) Push ¶
func (c *SafeChannel[T]) Push(value T) bool
Push put value to queue tail Pushes the value to the tail of the queue by default.
func (*SafeChannel[T]) Push2Head ¶
func (c *SafeChannel[T]) Push2Head(value T) bool
Push2Head put value to queue head Pushes the value to the head of the queue.
func (*SafeChannel[T]) PushForce ¶
func (c *SafeChannel[T]) PushForce(value T) bool
PushForce pushes the value even if the channel is full. If the channel is full, the oldest element (head) is discarded.
func (*SafeChannel[T]) SignalOnce ¶
func (c *SafeChannel[T]) SignalOnce()
SignalOnce wakes up all goroutines blocked on this channel, typically used to unblock PopupWithTimeout.
type SafeWaitGroup ¶
func NewSafeWaitGroup ¶
func NewSafeWaitGroup() *SafeWaitGroup
func (*SafeWaitGroup) Add ¶
func (w *SafeWaitGroup) Add(delta int)
func (*SafeWaitGroup) ClearWait ¶
func (w *SafeWaitGroup) ClearWait()
func (*SafeWaitGroup) Count ¶
func (w *SafeWaitGroup) Count() int
func (*SafeWaitGroup) Done ¶
func (w *SafeWaitGroup) Done()
func (*SafeWaitGroup) Wait ¶
func (w *SafeWaitGroup) Wait()
Source Files
¶
- concurrent_auto_delete_set.go
- concurrent_hash_map.go
- concurrent_hashmap_simple.go
- hashset.go
- limiter.go
- monitor.go
- safe_channel.go
- safe_wait_group.go