Skip to content

Commit ba3ae3f

Browse files
flycashhookokokoflyhigher139longyue0521WeiJiadong
authored
准备 Release 0.0.5 (#125)
* sqlx 加密列 key长度校验 (#102) * sqlx 加密列 key长度校验 * sqlx 加密列 key长度校验 补单元测试 * 修改加密列key长度错误提示 * atomicx: 泛型封装 atomic.Value (#101) * atomicx: 泛型封装 atomic.Value * 添加 CHANGELOG * syncx/atomicx: 增加 Swap 和 CAS 的泛型包装 * 添加 swap nil 的测试 * 添加更加多的 benchmark 测试,同时保证 NewValue 和 NewValueOf 的语义在 nil 上一致 * 优化单元测试 * queue: API 定义 (#109) * queue: API 定义 * 补充 API 说明 * 实现优先级队列和并发安全优先级队列 (#110) 基于小顶堆和切片的实现 * queue: 延时队列 (#115) * 延迟队列: 优化唤醒入队元素逻辑 (#117) * 修改CHANGELOG链接;添加测试用例修复bug Signed-off-by: longyue0521 <[email protected]> * 修改cond的SignalCh为signalCh;理清注释 Signed-off-by: longyue0521 <[email protected]> Signed-off-by: longyue0521 <[email protected]> * value: AnyValue 设计 (#120) (#121) * value: AnyValue 设计 (#120) * 修复ci检测问题 * 1.fix cr问题 2.add changelog对该pr的引用 3.add license 头部 * 1.修改ChangeLog,加入新特性描述 2.挪出value包,放在根目录 3.统一error格式打印 * 断言方式.Name改为.String Co-authored-by: vividwei <[email protected]> * queue: 基于切片的并发阻塞队列和基于 CAS 的并发队列设计 (#119) * queue:使用list包中的LinkedList实现并发阻塞链式队列 (#122) * queue:增加链式并发阻塞队列 Co-authored-by: kangdan <[email protected]> Co-authored-by: dan.kang <[email protected]> * ConcurrentLinkBlockingQueue 改成ConcurrentLinkedBlockingQueue (#123) * ConcurrentLinkBlockingQueue 改成ConcurrentLinkedBlockingQueue * modify .CHANGELOG.md * modify .CHANGELOG.md Co-authored-by: kangdan <[email protected]> Co-authored-by: dan.kang <[email protected]> * queue: ConcurrentLinkedQueue增加超时控制逻辑 (#124) Co-authored-by: kangdan <[email protected]> Co-authored-by: dan.kang <[email protected]> * queue: 添加例子 - 添加队列例子 - 去除 ConcurrentLinkedQueue 的超时机制 * queue: 添加例子 (#126) Signed-off-by: longyue0521 <[email protected]> Co-authored-by: hookokoko <[email protected]> Co-authored-by: Gevin <[email protected]> Co-authored-by: Longyue Li <[email protected]> Co-authored-by: 韦佳栋 <[email protected]> Co-authored-by: vividwei <[email protected]> Co-authored-by: kangdan666 <[email protected]> Co-authored-by: kangdan <[email protected]> Co-authored-by: dan.kang <[email protected]>
1 parent a953461 commit ba3ae3f

26 files changed

+4538
-17
lines changed

.CHANGELOG.md

+8
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,12 @@
11
# 开发中
2+
- [atomicx: 泛型封装 atomic.Value](https://github.com/gotomicro/ekit/pull/101)
3+
- [queue: API 定义](https://github.com/gotomicro/ekit/pull/109)
4+
- [queue: 基于堆和切片的优先级队列](https://github.com/gotomicro/ekit/pull/110)
5+
- [queue: 延时队列](https://github.com/gotomicro/ekit/pull/115)
6+
- [ekit: AnyValue 设计](https://github.com/gotomicro/ekit/pull/121)
7+
- [queue: 基于切片的并发阻塞队列和基于 CAS 的并发队列设计](https://github.com/gotomicro/ekit/pull/119)
8+
- [queue: 基于链表实现的有界/无界阻塞队列](https://github.com/gotomicro/ekit/pull/122)
9+
- [queue: ConcurrentLinkBlockingQueue重命名为ConcurrentLinkedBlockingQueue](https://github.com/gotomicro/ekit/pull/123)
210

311
# v0.0.4
412
- [slice: 重构 index 和 contains 的方法,直接调用对应Func 版本](https://github.com/gotomicro/ekit/pull/87)

internal/errs/error.go

+8-1
Original file line numberDiff line numberDiff line change
@@ -14,9 +14,16 @@
1414

1515
package errs
1616

17-
import "fmt"
17+
import (
18+
"fmt"
19+
)
1820

1921
// NewErrIndexOutOfRange 创建一个代表下标超出范围的错误
2022
func NewErrIndexOutOfRange(length int, index int) error {
2123
return fmt.Errorf("ekit: 下标超出范围,长度 %d, 下标 %d", length, index)
2224
}
25+
26+
// NewErrInvalidType 创建一个代表类型转换失败的错误
27+
func NewErrInvalidType(want, got string) error {
28+
return fmt.Errorf("ekit: 类型转换失败,want:%s, got:%s", want, got)
29+
}

internal/queue/priority_queue.go

+136
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,136 @@
1+
// Copyright 2021 gotomicro
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package queue
16+
17+
import (
18+
"errors"
19+
20+
"github.com/gotomicro/ekit/internal/slice"
21+
22+
"github.com/gotomicro/ekit"
23+
)
24+
25+
var (
26+
ErrOutOfCapacity = errors.New("ekit: 超出最大容量限制")
27+
ErrEmptyQueue = errors.New("ekit: 队列为空")
28+
)
29+
30+
// PriorityQueue 是一个基于小顶堆的优先队列
31+
// 当capacity <= 0时,为无界队列,切片容量会动态扩缩容
32+
// 当capacity > 0 时,为有界队列,初始化后就固定容量,不会扩缩容
33+
type PriorityQueue[T any] struct {
34+
// 用于比较前一个元素是否小于后一个元素
35+
compare ekit.Comparator[T]
36+
// 队列容量
37+
capacity int
38+
// 队列中的元素,为便于计算父子节点的index,0位置留空,根节点从1开始
39+
data []T
40+
}
41+
42+
func (p *PriorityQueue[T]) Len() int {
43+
return len(p.data) - 1
44+
}
45+
46+
// Cap 无界队列返回0,有界队列返回创建队列时设置的值
47+
func (p *PriorityQueue[T]) Cap() int {
48+
return p.capacity
49+
}
50+
51+
func (p *PriorityQueue[T]) IsBoundless() bool {
52+
return p.capacity <= 0
53+
}
54+
55+
func (p *PriorityQueue[T]) isFull() bool {
56+
return p.capacity > 0 && len(p.data)-1 == p.capacity
57+
}
58+
59+
func (p *PriorityQueue[T]) isEmpty() bool {
60+
return len(p.data) < 2
61+
}
62+
63+
func (p *PriorityQueue[T]) Peek() (T, error) {
64+
if p.isEmpty() {
65+
var t T
66+
return t, ErrEmptyQueue
67+
}
68+
return p.data[1], nil
69+
}
70+
71+
func (p *PriorityQueue[T]) Enqueue(t T) error {
72+
if p.isFull() {
73+
return ErrOutOfCapacity
74+
}
75+
76+
p.data = append(p.data, t)
77+
node, parent := len(p.data)-1, (len(p.data)-1)/2
78+
for parent > 0 && p.compare(p.data[node], p.data[parent]) < 0 {
79+
p.data[parent], p.data[node] = p.data[node], p.data[parent]
80+
node = parent
81+
parent = parent / 2
82+
}
83+
84+
return nil
85+
}
86+
87+
func (p *PriorityQueue[T]) Dequeue() (T, error) {
88+
if p.isEmpty() {
89+
var t T
90+
return t, ErrEmptyQueue
91+
}
92+
93+
pop := p.data[1]
94+
p.data[1] = p.data[len(p.data)-1]
95+
p.data = p.data[:len(p.data)-1]
96+
p.shrinkIfNecessary()
97+
p.heapify(p.data, len(p.data)-1, 1)
98+
return pop, nil
99+
}
100+
101+
func (p *PriorityQueue[T]) shrinkIfNecessary() {
102+
if p.IsBoundless() {
103+
p.data = slice.Shrink[T](p.data)
104+
}
105+
}
106+
107+
func (p *PriorityQueue[T]) heapify(data []T, n, i int) {
108+
minPos := i
109+
for {
110+
if left := i * 2; left <= n && p.compare(data[left], data[minPos]) < 0 {
111+
minPos = left
112+
}
113+
if right := i*2 + 1; right <= n && p.compare(data[right], data[minPos]) < 0 {
114+
minPos = right
115+
}
116+
if minPos == i {
117+
break
118+
}
119+
data[i], data[minPos] = data[minPos], data[i]
120+
i = minPos
121+
}
122+
}
123+
124+
// NewPriorityQueue 创建优先队列 capacity <= 0 时,为无界队列,否则有有界队列
125+
func NewPriorityQueue[T any](capacity int, compare ekit.Comparator[T]) *PriorityQueue[T] {
126+
sliceCap := capacity + 1
127+
if capacity < 1 {
128+
capacity = 0
129+
sliceCap = 64
130+
}
131+
return &PriorityQueue[T]{
132+
capacity: capacity,
133+
data: make([]T, 1, sliceCap),
134+
compare: compare,
135+
}
136+
}

0 commit comments

Comments
 (0)