README
¶
通用golang函数库
go get gitee.com/ivfzhou/gotools/v4@latest
// NewRunner 该函数提供同时最多运行max个协程fn，一旦fn发生error便终止fn运行。
//
// max小于等于0表示不限制协程数。
//
// 朝返回的run函数中添加fn，若block为true表示正在运行的任务数已达到max则会阻塞。
//
// run函数返回error为任务fn返回的第一个error，与wait函数返回的error为同一个。
//
// 注意请在add完所有任务后调用wait。
func NewRunner[T any](ctx context.Context, max int, fn func(context.Context, T) error) (
run func(t T, block bool) error, wait func(fastExit bool) error)
// RunPipeline 将每个jobs依次递给steps函数处理。一旦某个step发生error或者panic，立即返回该error，并及时结束其他协程。
// 除非stopWhenErr为false，则只是终止该job往下一个step投递。
//
// 一个job最多在一个step中运行一次，且一个job一定是依次序递给steps，前一个step处理完毕才会给下一个step处理。
//
// 每个step并发运行jobs。
//
// 等待所有jobs处理结束时会close successCh、errCh，或者ctx被cancel时也将及时结束开启的goroutine后返回。
//
// 从successCh和errCh中获取成功跑完所有step的job和是否发生error。
//
// 若steps中含有nil将会panic。
func RunPipeline[T any](ctx context.Context, jobs []T, stopWhenErr bool, steps ...func(context.Context, T) error) (
successCh <-chan T, errCh <-chan error)
// NewWriteAtReader 获取一个WriterAt和Reader对象，其中WriterAt用于并发写入数据，而与此同时Reader对象同时读取出已经写入好的数据。
//
// WriterAt写入完毕后调用Close，则Reader会全部读取完后结束读取。
//
// WriterAt发生的error会传递给Reader返回。
//
// 该接口是特定为一个目的实现————服务器分片下载数据中转给客户端下载，提高中转数据效率。
func NewWriteAtReader() (WriteAtCloser, io.ReadCloser)
// NewMultiReadCloserToReader 依次从rc中读出数据直到io.EOF则close rc。从r获取rc中读出的数据。
//
// add添加rc，返回error表明读取rc发生错误，可以安全的添加nil。调用endAdd表明不会再有rc添加，当所有数据读完了时，r将返回EOF。
//
// 如果ctx被cancel，将停止读取并返回error。
//
// 所有添加进去的io.ReadCloser都会被close。
func NewMultiReadCloserToReader(ctx context.Context, rc ...io.ReadCloser) (
r io.Reader, add func(rc io.ReadCloser) error, endAdd func())
联系电邮：ivfzhou@126.com
Documentation
¶
Index ¶
- func CloseIO(c io.Closer)
- func Contains[E comparable](arr []E, elem E) bool
- func ConvertMap[K comparable, V, T any](m map[K]V, fn func(K, V) T) []T
- func ConvertSlice[E, T any](sli []E, fn func(E) T) []T
- func ConvertSliceToMap[K comparable, V, E any](sli []E, fn func(E) (K, V)) map[K]V
- func CopyFile(src, dest string) error
- func DistinctSlice[E comparable](sli []E) []E
- func DistinctSliceByHash[E any](sli []E, hash func(E) string) []E
- func DropSliceZero[E any](sli []E) []E
- func FilterMap[K comparable, V any](m map[K]V, fn func(K, V) bool) map[K]V
- func FilterSlice[E any](sli []E, fn func(E) bool) []E
- func ForeachSlice[E any](sli []E, fn func(E))
- func GCD(x, y int) int
- func IPv4ToNum(ip string) uint32
- func IPv4ToStr(ip uint32) string
- func IsIPv4(s string) bool
- func IsIPv6(s string) bool
- func IsIntranet(ipv4 string) bool
- func IsMAC(s string) bool
- func IsZero(v any) bool
- func ListenChan[T any](chans ...<-chan T) (ch <-chan T)
- func Max[T Number](x ...T) T
- func Min[T Number](x ...T) T
- func NewMultiReadCloserToReader(ctx context.Context, rc ...io.ReadCloser) (r io.Reader, add func(rc io.ReadCloser) error, endAdd func())
- func NewMultiReadCloserToWriter(ctx context.Context, writer func(order int, p []byte)) (send func(readSize, order int, reader io.ReadCloser), wait func() error)
- func NewPipelineRunner[T any](ctx context.Context, steps ...func(context.Context, T) bool) (push func(T) bool, successCh <-chan T, endPush func())
- func NewRunner[T any](ctx context.Context, max int, fn func(context.Context, T) error) (run func(t T, block bool) error, wait func(fastExit bool) error)
- func PickMapKey[K comparable, V any](m map[K]V) []K
- func PickMapValue[K comparable, V any](m map[K]V) []V
- func RandomChars(length int) string
- func RandomCharsCaseInsensitive(length int) string
- func RandomString(length int, chars string) string
- func RunCommandAndPrompt(bin string, args []string, prompts ...string) (stdout, stderr []byte, err error)
- func RunConcurrently(ctx context.Context, fn ...func(context.Context) error) (wait func(fastExit bool) error)
- func RunData[T any](ctx context.Context, fn func(context.Context, T) error, fastExit bool, ...) error
- func RunPeriodically(period time.Duration) (run func(fn func()))
- func RunPipeline[T any](ctx context.Context, jobs []T, stopWhenErr bool, ...) (successCh <-chan T, errCh <-chan error)
- func RunSequentially(ctx context.Context, fn ...func(context.Context) error) error
- func StackTrace() string
- func UUIDLike() string
- func UnzipFromBytes(zippedBytes []byte, toDir string) (filePaths []string, err error)
- func UnzipFromFiles(zippedFilePath string, toDir string) (filePaths []string, err error)
- func ZipFiles(toZippedFilePath string, fromDir string) error
- func ZipFilesToBytes(filePaths ...string) ([]byte, error)
- type AxisMarker
- type Command
- type FairLocker
- type Number
- type Queue
- type ReadFirstLocker
- type WriteAtCloser
- type WriteFirstLocker
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Contains ¶
func Contains[E comparable](arr []E, elem E) bool
func ConvertMap ¶
func ConvertMap[K comparable, V, T any](m map[K]V, fn func(K, V) T) []T
func ConvertSlice ¶
func ConvertSlice[E, T any](sli []E, fn func(E) T) []T
func ConvertSliceToMap ¶
func ConvertSliceToMap[K comparable, V, E any](sli []E, fn func(E) (K, V)) map[K]V
func DistinctSlice ¶
func DistinctSlice[E comparable](sli []E) []E
func DistinctSliceByHash ¶ added in v4.1.0
func DropSliceZero ¶
func DropSliceZero[E any](sli []E) []E
func FilterMap ¶
func FilterMap[K comparable, V any](m map[K]V, fn func(K, V) bool) map[K]V
func FilterSlice ¶
func ForeachSlice ¶
func ForeachSlice[E any](sli []E, fn func(E))
func ListenChan ¶
func ListenChan[T any](chans ...<-chan T) (ch <-chan T)
ListenChan 监听chans，一旦有一个chan激活便立即将T发送给ch，并close ch。
若所有chans都未曾激活（chan是nil也认为未激活）且都close了，则ch被close。
若同时多个chans被激活，则随机将一个激活值发送给ch。
func NewMultiReadCloserToReader ¶
func NewMultiReadCloserToReader(ctx context.Context, rc ...io.ReadCloser) ( r io.Reader, add func(rc io.ReadCloser) error, endAdd func())
NewMultiReadCloserToReader 依次从rc中读出数据直到io.EOF则close rc。从r获取rc中读出的数据。
add添加rc，返回error表明读取rc发生错误，可以安全的添加nil。调用endAdd表明不会再有rc添加，当所有数据读完了时，r将返回EOF。
如果ctx被cancel，将停止读取并返回error。
所有添加进去的io.ReadCloser都会被close。
func NewMultiReadCloserToWriter ¶
func NewMultiReadCloserToWriter(ctx context.Context, writer func(order int, p []byte)) ( send func(readSize, order int, reader io.ReadCloser), wait func() error)
NewMultiReadCloserToWriter 依次从reader读出数据并写入writer中，并close reader。
返回send用于添加reader，readSize表示需要从reader读出的字节数，order用于表示记录读取序数并传递给writer，若读取字节数对不上则返回error。
返回wait用于等待所有reader读完，若读取发生error，wait返回该error，并结束读取。
务必等所有reader都已添加给send后再调用wait。
该函数可用于需要非同一时间多个读取流和一个写入流的工作模型。
func NewPipelineRunner ¶
func NewPipelineRunner[T any](ctx context.Context, steps ...func(context.Context, T) bool) ( push func(T) bool, successCh <-chan T, endPush func())
NewPipelineRunner 形同RunPipeline，不同在于使用push推送job。step返回true表示传递给下一个step处理。
func NewRunner ¶
func NewRunner[T any](ctx context.Context, max int, fn func(context.Context, T) error) ( run func(t T, block bool) error, wait func(fastExit bool) error)
NewRunner 该函数提供同时最多运行max个协程fn，一旦fn发生error便终止fn运行。
max小于等于0表示不限制协程数。
朝返回的run函数中添加fn，若block为true表示正在运行的任务数已达到max则会阻塞。
run函数返回error为任务fn返回的第一个error，与wait函数返回的error为同一个。
注意请在add完所有任务后调用wait。
Example ¶
type product struct {
// some stuff
}
ctx := context.Background()
op := func(ctx context.Context, data *product) error {
// do something
return nil
}
add, wait := gotools.NewRunner[*product](ctx, 12, op)
// many products
var projects []*product
for _, v := range projects {
// blocked since number of ops running simultaneously reaches 12
if err := add(v, true); err != nil {
// means having a op return err
}
// no block
if err := add(v, false); err != nil {
// means having a op return err
}
}
// wait all op done and check err
if err := wait(true); err != nil {
// op occur err
}
Output:
func PickMapKey ¶
func PickMapKey[K comparable, V any](m map[K]V) []K
func PickMapValue ¶
func PickMapValue[K comparable, V any](m map[K]V) []V
func RandomCharsCaseInsensitive ¶
RandomCharsCaseInsensitive 生成随机字符串（数字加字母组合）。
func RandomString ¶
RandomString 生成所以字符串，内容限制为chars中存在的字符。
func RunCommandAndPrompt ¶
func RunCommandAndPrompt(bin string, args []string, prompts ...string) (stdout, stderr []byte, err error)
RunCommandAndPrompt 运行命令并响应输入prompts。
func RunConcurrently ¶
func RunConcurrently(ctx context.Context, fn ...func(context.Context) error) (wait func(fastExit bool) error)
RunConcurrently 并发运行fn，一旦有error发生终止运行。
Example ¶
ctx := context.Background()
var order any
work1 := func(ctx context.Context) error {
// op order
order = nil
return nil
}
var stock any
work2 := func(ctx context.Context) error {
// op stock
stock = nil
return nil
}
err := gotools.RunConcurrently(ctx, work1, work2)(false)
// check err
if err != nil {
return
}
// do your want
_ = order
_ = stock
Output:
func RunData ¶
func RunData[T any](ctx context.Context, fn func(context.Context, T) error, fastExit bool, jobs ...T) error
RunData 并发将jobs传递给fn函数运行，一旦发生error便立即返回该error，并结束其它协程。
func RunPeriodically ¶
RunPeriodically 依次运行fn，每个fn之间至少间隔period时间。
func RunPipeline ¶
func RunPipeline[T any](ctx context.Context, jobs []T, stopWhenErr bool, steps ...func(context.Context, T) error) ( successCh <-chan T, errCh <-chan error)
RunPipeline 将每个jobs依次递给steps函数处理。一旦某个step发生error或者panic，立即返回该error，并及时结束其他协程。 除非stopWhenErr为false，则只是终止该job往下一个step投递。
一个job最多在一个step中运行一次，且一个job一定是依次序递给steps，前一个step处理完毕才会给下一个step处理。
每个step并发运行jobs。
等待所有jobs处理结束时会close successCh、errCh，或者ctx被cancel时也将及时结束开启的goroutine后返回。
从successCh和errCh中获取成功跑完所有step的job和是否发生error。
若steps中含有nil将会panic。
Example ¶
type data struct{}
ctx := context.Background()
jobs := []*data{{}, {}}
work1 := func(ctx context.Context, d *data) error { return nil }
work2 := func(ctx context.Context, d *data) error { return nil }
succCh, errCh := gotools.RunPipeline(ctx, jobs, false, work1, work2)
select {
case <-succCh:
case <-errCh:
// return err
}
Output:
func RunSequentially ¶
RunSequentially 依次运行fn，当有error发生时停止后续fn运行。
Example ¶
ctx := context.Background()
first := func(context.Context) error { return nil }
then := func(context.Context) error { return nil }
last := func(context.Context) error { return nil }
err := gotools.RunSequentially(ctx, first, then, last)
if err != nil {
// return err
}
Output:
func StackTrace ¶
func StackTrace() string
func UnzipFromBytes ¶
UnzipFromBytes 将压缩数据解压写入硬盘。
func UnzipFromFiles ¶
UnzipFromFiles 解压并写入硬盘。
func ZipFilesToBytes ¶
ZipFilesToBytes 将文件打成压缩包。
Types ¶
type AxisMarker ¶
type AxisMarker struct {
// contains filtered or unexported fields
}
AxisMarker 坐标轴记录器。
func (*AxisMarker) GetMaxMarkLine ¶
func (m *AxisMarker) GetMaxMarkLine(begin int64) int64
GetMaxMarkLine 获取从某点开始连续被标记的长度。
func (*AxisMarker) String ¶
func (m *AxisMarker) String() string
type FairLocker ¶
type FairLocker struct {
// contains filtered or unexported fields
}
FairLocker 公平读写锁。
先到达的请求先获取处理器，无论是读请求还是写请求。
func (*FairLocker) RLock ¶
func (locker *FairLocker) RLock()
func (*FairLocker) RUnlock ¶
func (locker *FairLocker) RUnlock()
func (*FairLocker) WLock ¶
func (locker *FairLocker) WLock()
func (*FairLocker) WUnlock ¶
func (locker *FairLocker) WUnlock()
type ReadFirstLocker ¶
type ReadFirstLocker struct {
// contains filtered or unexported fields
}
ReadFirstLocker 读优先锁。
读优先意味着，连续地读请求获取处理器的概率高于写请求。
func (*ReadFirstLocker) RLock ¶
func (locker *ReadFirstLocker) RLock()
func (*ReadFirstLocker) RUnlock ¶
func (locker *ReadFirstLocker) RUnlock()
func (*ReadFirstLocker) WLock ¶
func (locker *ReadFirstLocker) WLock()
func (*ReadFirstLocker) WUnlock ¶
func (locker *ReadFirstLocker) WUnlock()
type WriteAtCloser ¶
type WriteAtCloser interface {
io.WriterAt
// Closer
// Deprecated: 使用CloseByError代替
io.Closer
CloseByError(error) error
}
func NewWriteAtReader ¶
func NewWriteAtReader() (WriteAtCloser, io.ReadCloser)
NewWriteAtReader 获取一个WriterAt和Reader对象，其中WriterAt用于并发写入数据，而与此同时Reader对象同时读取出已经写入好的数据。
WriterAt写入完毕后调用Close，则Reader会全部读取完后结束读取。
WriterAt发生的error会传递给Reader返回。
该接口是特定为一个目的实现————服务器分片下载数据中转给客户端下载，提高中转数据效率。
Example ¶
writeAtCloser, readCloser := gotools.NewWriteAtReader()
// 并发写入数据
go func() {
_, _ = writeAtCloser.WriteAt(nil, 0)
// 发生错误时关闭
_ = writeAtCloser.CloseByError(nil)
}()
// 写完close
_ = writeAtCloser.Close()
// 同时读出写入的数据
go func() {
_, _ = readCloser.Read(nil)
}()
// 读完close
_ = readCloser.Close()
Output:
type WriteFirstLocker ¶
type WriteFirstLocker struct {
// contains filtered or unexported fields
}
WriteFirstLocker 写优先锁。
读优先意味着，连续地写请求获取处理器的概率高于读请求。
func (*WriteFirstLocker) RLock ¶
func (locker *WriteFirstLocker) RLock()
func (*WriteFirstLocker) RUnlock ¶
func (locker *WriteFirstLocker) RUnlock()
func (*WriteFirstLocker) WLock ¶
func (locker *WriteFirstLocker) WLock()
func (*WriteFirstLocker) WUnlock ¶
func (locker *WriteFirstLocker) WUnlock()
Source Files
Click to show internal directories.
Click to hide internal directories.