pools

Documentation

Overview

Example

package main

import (
	"fmt"
	"time"

	"github.com/xuender/kit/los"
	"github.com/xuender/kit/pools"
)

func main() {
	pool := pools.New(10, func(value, num int) string {
		time.Sleep(time.Millisecond)

		return fmt.Sprintf("%d: %d*2=%d", num, value, value*2)
	})
	defer pool.Close()

	outputs := pool.Post(los.Range[int](100))

	fmt.Println(len(outputs))

}

Output:

100

Index

• type Native
  • func NewNative[I any, O any](size uint, yield func(context.Context, I) O) *Native[I, O]
  • func (p *Native[I, O]) Close()
  • func (p *Native[I, O]) Post(ctx context.Context, elems ...I) []O
  • func (p *Native[I, O]) Wait()
• type Pool
  • func New[I, O any](size int, yield func(I, int) O) *Pool[I, O]
  • func (p *Pool[I, O]) Close()
  • func (p *Pool[I, O]) Post(inputs []I) []O
  • func (p *Pool[I, O]) Run(input I) O
• type Simple
  • func NewSimple[T any](size int, yield func(T, int)) *Simple[T]
  • func (p *Simple[T]) Close()
  • func (p *Simple[T]) Post(elems ...T)
  • func (p *Simple[T]) Wait()
• type SyncPool
  • func NewSyncPool[T any](newFunc func() T, initFunc func(T)) *SyncPool[T]
  • func (p *SyncPool[T]) Get() T
  • func (p *SyncPool[T]) Put(elem T)

Examples

• Package
• NewSyncPool
• Pool (Context)
• Pool (Error)
• Pool.Run
• Simple
• Simple.Wait

Types

type Native

type Native[I, O any] struct {
	// contains filtered or unexported fields
}

Native 原生的协程控制.

func NewNative

func NewNative[I any, O any](size uint, yield func(context.Context, I) O) *Native[I, O]

NewNative 新建原生的协程控制.

func (*Native[I, O]) Close

func (p *Native[I, O]) Close()

Close 关闭协程池.

func (*Native[I, O]) Post

func (p *Native[I, O]) Post(ctx context.Context, elems ...I) []O

Post 发送数据.

func (*Native[I, O]) Wait

func (p *Native[I, O]) Wait()

Wait 等待完毕.

type Pool

type Pool[I, O any] struct {
	// contains filtered or unexported fields
}

Pool Goroutine 池.

Example (Context)

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/samber/lo"
	"github.com/xuender/kit/los"
	"github.com/xuender/kit/pools"
)

func main() {
	pool := pools.New(10, func(input lo.Tuple2[context.Context, int], _ int) int {
		time.Sleep(time.Millisecond)

		return input.B * input.B
	})
	defer pool.Close()

	inputs := los.Map(los.Range[int](100), func(num int) lo.Tuple2[context.Context, int] {
		return lo.T2(context.Background(), num)
	})
	outputs := pool.Post(inputs)

	fmt.Println(len(outputs))

}

Output:

100

Example (Error)

package main

import (
	"errors"
	"fmt"
	"time"

	"github.com/samber/lo"
	"github.com/xuender/kit/los"
	"github.com/xuender/kit/pools"
)

func main() {
	pool := pools.New(10, func(value, _ int) lo.Tuple2[int, error] {
		time.Sleep(time.Millisecond)

		if value == 0 {
			// nolint
			return lo.T2(0, errors.New("divide by zero"))
		}

		return lo.T2[int, error](100/value, nil)
	})
	defer pool.Close()

	outputs := pool.Post(los.Range[int](100))

	fmt.Println(len(outputs))

}

Output:

100

func New

func New[I, O any](size int, yield func(I, int) O) *Pool[I, O]

New 新建 Goroutine 池.

func (*Pool[I, O]) Close

func (p *Pool[I, O]) Close()

Close 关闭协程池.

func (*Pool[I, O]) Post

func (p *Pool[I, O]) Post(inputs []I) []O

Post 批量任务处理.

func (*Pool[I, O]) Run

func (p *Pool[I, O]) Run(input I) O

Run 执行单个任务.

Example

package main

import (
	"fmt"
	"time"

	"github.com/xuender/kit/pools"
)

func main() {
	pool := pools.New(10, func(value, _ int) string {
		time.Sleep(time.Millisecond)

		return fmt.Sprintf("%d*2=%d", value, value*2)
	})
	defer pool.Close()

	fmt.Println(pool.Run(3))

}

Output:

3*2=6

type Simple

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

Simle 简单协程池.

Example

ExampleSimple is an example function.

package main

import (
	"fmt"
	"time"

	"github.com/xuender/kit/los"
	"github.com/xuender/kit/pools"
)

func main() {
	pool := pools.NewSimple(2, func(_, _ int) {
		fmt.Println("a")
	})
	defer pool.Close()

	pool.Post(los.Range[int](3)...)
	time.Sleep(time.Millisecond)

}

Output:

a
a
a

func NewSimple

func NewSimple[T any](size int, yield func(T, int)) *Simple[T]

NewSimle 新建简单协程池.

func (*Simple[T]) Close

func (p *Simple[T]) Close()

Close 关闭协程池.

func (*Simple[T]) Post

func (p *Simple[T]) Post(elems ...T)

Post 发送数据.

func (*Simple[T]) Wait

func (p *Simple[T]) Wait()

Wait 等待完毕.

Example

package main

import (
	"fmt"
	"time"

	"github.com/xuender/kit/pools"
)

func main() {
	pool := pools.NewSimple(2, func(_, _ int) {
		fmt.Println("a")
	})
	defer pool.Close()

	pool.Post(1)
	pool.Post(2, 3)
	pool.Wait()
	time.Sleep(time.Millisecond)

}

Output:

a
a
a

type SyncPool

type SyncPool[T any] struct {
	sync.Pool
	// contains filtered or unexported fields
}

func NewSyncPool

func NewSyncPool[T any](newFunc func() T, initFunc func(T)) *SyncPool[T]

Example

ExampleNewSyncPool is an example function.

package main

import (
	"fmt"

	"github.com/xuender/kit/pools"
)

type data struct {
	data int
}

// ExampleNewSyncPool is an example function.
func main() {
	pool := pools.NewSyncPool(func() *data { return &data{1} }, func(d *data) { d.data = 1 })
	d1 := pool.Get()
	d1.data = 3
	pool.Put(d1)

	d2 := pool.Get()
	fmt.Println(d2.data)

}

Output:

1

func (*SyncPool[T]) Get

func (p *SyncPool[T]) Get() T

nolint: forcetypeassert

func (*SyncPool[T]) Put

func (p *SyncPool[T]) Put(elem T)