comag

并发处理功能

基本使用

消息发送器 - ProcessSender

对应模型类： ProcessSender

基本使用

sender := NewProcessSender[*tests.Person]()

go func(ctx context.Context) {
    for {
        select {
        case d := <-sender.Ch():
            t.Logf("[%d-%s]%v", ctx.Value(ContextRunId), ctx.Value(ContextRunName), d)
        case <-ctx.Done():
            return
        }
    }
}

for i := 1; i <= 100; i++ {
    sender.Send(tests.NewPerson())
    time.Sleep(time.Millisecond * 200)
}

并发处理器 - ConProcessModelActuator

对应模型类： ConProcessModelActuator

基本使用

actuator := NewConProcessModelActuator()
f := func(id int) CoFunc {
    return func(ctx context.Context) {
        defer fmt.Printf("xxx-%d DONE\n", id)
        ti := time.NewTicker(time.Second * time.Duration(id%4))
        for {
            select {
            case <-ctx.Done():
                time.Sleep(time.Second * time.Duration(id%4))
                return
            case <-ti.C:
                fmt.Printf("DOING-%d\n", id)
                if id == 3 {
                    a.Stop()
                    return
                }
            }
        }
    }
}
actuator.Add(f(1))
actuator.Add(f(2))
actuator.Add(f(3))
actuator.Start()

停止执行（阻塞等待结束）

time.Sleep(time.Second * 3)
actuator.Stop()
fmt.Printf("TO END\n")

actuator.Wait()

a.Stop()
fmt.Printf("TO END\n")

配合ProcessSender（并发消息处理）

可配合 ProcessSender 进行并发消息的分配处理

sender := NewProcessSender[int]()
actuator := NewConProcessModelActuator()

f := func(ctx context.Context) {
		for {
			select {
			case d := <-sender.Ch():
				t.Logf("[%d-%s]%d", ctx.Value(ContextRunId), ctx.Value(ContextRunName), d)
			case <-ctx.Done():
				return
			}
		}
}
actuator.AddRepeat(f, 10)
actuator.Start()
	
for i := 1; i <= 100; i++ {
		sender.Send(i)
		time.Sleep(time.Millisecond * 200)
}

发布订阅处理器 - PubSubActuator

对应模型类： PubSubActuator

初始化订阅

bus := NewPubSubActuator[int](v1log.Out)

countA := 0
common.ErrPanic(bus.SubscribeByFn("order.created", func(index int, msg int) {
    countA++
}))

common.ErrPanic(bus.Start())

手动启动

common.ErrPanic(bus.Start())

发送消息

common.ErrPanic(bus.Publish("order.created", 1))
common.ErrPanic(bus.Publish("order.created", 2))

手动停止（阻塞等待结束）

common.ErrPanic(bus.Stop())

支持泛型消息

指定int类型作为消息数据

bus := NewPubSubActuator[int](v1log.Out)

使用自定义结构体作为消息数据

bus := NewPubSubActuator[*tests.Person](v1log.Out)

EventBus（扩展功能）

支持事件总线，固定消息数据为 Message 类型

bus := NewEventBus(v1log.Out)
err = bus.SubscribeByFn("order.created", func(index int, msg *extendparams.Message[interface{}]) {
	countA++
	t.Logf("[GET.created-%d]%s %d %+v", index, msg.ExtendParams.Get("topic"), countA, msg.PayloadData)
})
common.ErrPanic(err)

common.ErrPanic(bus.Start())

msg := extendparams.NewMessage[interface{}]()
msg.PayloadData = tests.NewPerson()
msg.ExtendParams.Set("topic", "order.payed")
msg.ExtendParams.Set("index", fmt.Sprint(i))
common.ErrPanic(bus.Publish("order.payed", msg))