130 lines
3.0 KiB
Go
130 lines
3.0 KiB
Go
|
/*
|
|||
|
|
* Copyright (C) THL A29 Limited, a Tencent company. All rights reserved.
|
|||
|
|
*
|
|||
|
|
* SPDX-License-Identifier: Apache-2.0
|
|||
|
|
*
|
|||
|
|
*/
|
|||
|
|
|
|||
|
|
package main
|
|||
|
|
|
|||
|
|
import (
|
|||
|
|
"fmt"
|
|||
|
|
"github.com/bruceshao/lockfree"
|
|||
|
|
"math/rand"
|
|||
|
|
"sync"
|
|||
|
|
"sync/atomic"
|
|||
|
|
"time"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
func main() {
|
|||
|
|
fmt.Println("========== start write by discard ==========")
|
|||
|
|
writeByDiscard()
|
|||
|
|
fmt.Println("========== complete write by discard ==========")
|
|||
|
|
fmt.Println("========== start write by cursor ==========")
|
|||
|
|
writeByCursor()
|
|||
|
|
fmt.Println("========== complete write by cursor ==========")
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func writeByDiscard() {
|
|||
|
|
var counter = uint64(0)
|
|||
|
|
// 写入超时，如何使用
|
|||
|
|
eh := &fixedSleepEventHandler[uint64]{
|
|||
|
|
sm: time.Millisecond * 10,
|
|||
|
|
}
|
|||
|
|
disruptor := lockfree.NewLockfree[uint64](2, eh, lockfree.NewSleepBlockStrategy(time.Microsecond))
|
|||
|
|
disruptor.Start()
|
|||
|
|
producer := disruptor.Producer()
|
|||
|
|
// 假设有100个写g
|
|||
|
|
var wg sync.WaitGroup
|
|||
|
|
for i := 0; i < 10; i++ {
|
|||
|
|
wg.Add(1)
|
|||
|
|
go func() {
|
|||
|
|
for j := 0; j < 10; j++ {
|
|||
|
|
v := atomic.AddUint64(&counter, 1)
|
|||
|
|
ww := producer.WriteWindow()
|
|||
|
|
if ww <= 0 {
|
|||
|
|
// 表示无法写入，丢弃
|
|||
|
|
fmt.Println("discard ", v)
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
// 表示可以写入，写入即可
|
|||
|
|
err := producer.Write(v)
|
|||
|
|
if err != nil {
|
|||
|
|
panic(err)
|
|||
|
|
}
|
|||
|
|
fmt.Println("write ", v)
|
|||
|
|
}
|
|||
|
|
wg.Done()
|
|||
|
|
}()
|
|||
|
|
}
|
|||
|
|
wg.Wait()
|
|||
|
|
time.Sleep(3 * time.Second)
|
|||
|
|
disruptor.Close()
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func writeByCursor() {
|
|||
|
|
var counter = uint64(0)
|
|||
|
|
// 写入超时，如何使用
|
|||
|
|
eh := &randomSleepEventHandler[uint64]{}
|
|||
|
|
disruptor := lockfree.NewLockfree[uint64](2, eh, lockfree.NewSleepBlockStrategy(time.Microsecond))
|
|||
|
|
disruptor.Start()
|
|||
|
|
producer := disruptor.Producer()
|
|||
|
|
// 假设有100个写g
|
|||
|
|
var wg sync.WaitGroup
|
|||
|
|
for i := 0; i < 10; i++ {
|
|||
|
|
wg.Add(1)
|
|||
|
|
go func() {
|
|||
|
|
for j := 0; j < 10; j++ {
|
|||
|
|
v := atomic.AddUint64(&counter, 1)
|
|||
|
|
wc, exist, err := producer.WriteTimeout(v, time.Millisecond)
|
|||
|
|
if err != nil {
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
if !exist {
|
|||
|
|
// 重复写入1次，写入不成功则丢弃重新写其他的
|
|||
|
|
if ok, _ := producer.WriteByCursor(v, wc); ok {
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
fmt.Println("discard ", v)
|
|||
|
|
// 重新生成值，一直等待写入
|
|||
|
|
v = atomic.AddUint64(&counter, 1)
|
|||
|
|
for {
|
|||
|
|
if ok, _ := producer.WriteByCursor(v, wc); ok {
|
|||
|
|
fmt.Println("write ", v, " with x times")
|
|||
|
|
break
|
|||
|
|
}
|
|||
|
|
// 写入不成功则休眠，防止CPU暴增
|
|||
|
|
time.Sleep(100 * time.Microsecond)
|
|||
|
|
}
|
|||
|
|
} else {
|
|||
|
|
fmt.Println("write ", v, " with 1 time")
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
wg.Done()
|
|||
|
|
}()
|
|||
|
|
}
|
|||
|
|
wg.Wait()
|
|||
|
|
time.Sleep(3 * time.Second)
|
|||
|
|
disruptor.Close()
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
type fixedSleepEventHandler[T uint64] struct {
|
|||
|
|
sm time.Duration
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (h *fixedSleepEventHandler[T]) OnEvent(v uint64) {
|
|||
|
|
time.Sleep(h.sm)
|
|||
|
|
fmt.Println("consumer ", v)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type randomSleepEventHandler[T uint64] struct {
|
|||
|
|
count int32
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (h *randomSleepEventHandler[T]) OnEvent(v uint64) {
|
|||
|
|
// 每次处理都会进行随机休眠，可以导致消费端变慢
|
|||
|
|
intn := rand.Intn(1000)
|
|||
|
|
time.Sleep(time.Duration(intn * 1000))
|
|||
|
|
fmt.Println("consumer count ", atomic.AddInt32(&h.count, 1))
|
|||
|
|
}
|