GO 协程池的浅探

最新推荐文章于 2024-09-15 08:30:00 发布
最新推荐文章于 2024-09-15 08:30:00 发布
阅读量494 收藏
点赞数
分类专栏： Go 文章标签： go语言 线程池 后端 协程
版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://blog.csdn.net/heshushun/article/details/119065302
版权

为啥需要协程池

go语言支持多协程并发，具有Goroutine调度的特性。当处理并发问题时 可以开辟多个Goroutine去解决。

虽然go语言在调度Goroutine已经优化的非常完善，开启一个Goroutine的代价非常小。但是，如果无休止的开辟Goroutine会出现高频率的调度Goroutine，那么依然会浪费很多上下文切换的资源。并且每个Goroutine都有一定内存的开销，当量足够大的时候系统内存会承担不住 服务会被自动OOM。

先聊聊并发问题

上半部分是串行，下半部分是并行。明显并发可以

  • - 同一时刻处理多个事务
  • - 节省时间，提高性能效率

并发的条件：

  • - 事务独立
  • - 没有前后结果依赖关系

 go是如何实现并发？

串行


package main

import (
   "fmt"
   "time"
)

func doTask1() {
   time.Sleep(1 * time.Second)
   fmt.Println("test1")
}

func doTask2() {
   time.Sleep(1 * time.Second)
   fmt.Println("test2")
}

func main() {
   beforeTime := time.Now()
   doTask1()
   doTask2()
   runTime := time.Since(beforeTime)
   fmt.Println(runTime)
}

结果

```

test1
test2
2.0007157s


```

并行  task加上go

```go

    ...
    go doTask1()
    doTask2()
    ...


```

结果

```

test1
test2
1.0003288s


```

go协程之间是如何通信的？

使用go语言的另外一个特性channel

```go

package main

import (
    "fmt"
    "time"
)

func doTask1(c chan int) {
    for {
        select {
        case num := <-c:
            fmt.Printf("get %d \n", num)
        }
    }
}

func doTask2(c chan int) {
    for i := 1; i < 10; i++ {
        time.Sleep(1 * time.Second)
        fmt.Printf("put %d \n", i)
        c <- i
    }
}

func main() {
    //创建一个channel
    c := make(chan int, 10)
    go doTask1(c)
    go doTask2(c)
    time.Sleep(12 * time.Second)
}

```

协程池的设计

初步设计上 一个pool 由若干个worker和一个JobsChannel组成。

  • - worker 负责接收来自JobsChannel的task，并且工作执行。
  • - JobsChannel 负责接收进来的每个task然后随机分配给每个worker来处理。
  • -  task 由func执行函数和所需的参数组成。

简单实现

workpool.go```go

import (
    "context"
    "fmt"
    "sync"
)

type TaskFunc func(args ...interface{}) error

type Task struct {
    f    TaskFunc
    args []interface{}
}

func NewTask(f TaskFunc, args ...interface{}) *Task {
    t := Task{
        f:    f,
        args: args,
    }
    return &t
}

func (t *Task) Execute() error {
    err := t.f(t.args...)
    return err
}

type Pool struct {
    //协程池worker数量 (Goroutine个数)
    workerNum int

    //协程池任务队列
    JobsChannel chan *Task

    stopCtx        context.Context
    stopCancelFunc context.CancelFunc
    wg             sync.WaitGroup
}

// 创建一个协程池
func NewPool(cap int, poolLen int) *Pool {
    p := Pool{
        workerNum:   cap,
        JobsChannel: make(chan *Task, poolLen),
    }
    return &p
}


```

workpool_test.go```go

import (
    "fmt"
    "testing"
    "time"
)

func calcAttrTaskFunc() TaskFunc {
    taskFunc := TaskFunc(func(args ...interface{}) error {
        // TODO do someting
        return nil
    })
    return taskFunc
}

func TestCalculator_Workpool(t *testing.T) {

    //创建一个协程池
    p := NewPool(3, 100)

    //启动协程池p
    p.Run()

    //开一个协程 不断的向 Pool 输送task任务
    go func() {
        for i := 1; i < 10; i++ {
            task := NewTask(calcAttrTaskFunc(), i)
            p.JobsChannel <- task
        }
    }()

    time.Sleep(1 * time.Second)
}


```

以上只是简单实现 <u>线程池的并发处理</u> 原理， 那 如果任务执行失败了呢，是不是要重塞重试？再优化下。

workpool.go```go

...
// 加一个任务
func (p *Pool) pushTask(t *Task) {
    p.JobsChannel <- t
}

// 协程池创建一个worker并且开始工作
func (p *Pool) worker(workID int) {
    for {
        select {
        case <-p.stopCtx.Done():
            p.wg.Done()
            fmt.Println("协程池Pool停止")
            return
        case task := <-p.JobsChannel:
            err := task.Execute()
            if err != nil {
                p.pushTask(task)
                continue
            }
            fmt.Println("worker ID ", workID, " 执行 task ID", task.args[0].(int), "任务完毕")
        }
    }
}
...


```

再优化代码更加健壮些，如果worker 15s内没有收到任何任务，则当作timeout回收。

```go

// 协程池创建一个worker并且开始工作
func (p *Pool) worker(workID int) {
    ticker := time.NewTicker(time.Second * 15)
    defer ticker.Stop()
    for {
        select {
        case <-p.stopCtx.Done():
            p.wg.Done()
            fmt.Println("协程池Pool停止")
            return
        case task := <-p.JobsChannel:
            err := task.Execute()
            if err != nil {
                p.pushTask(task)
                continue
            }
            fmt.Println("worker ID ", workID, " 执行 task ID", task.args[0].(int), "任务完毕")
            // 重新设置等待15s
            ticker.Reset(time.Second * 15)
        case <-ticker.C: // 15s 都收不到任何处理就回收
            fmt.Println("worker ID ", workID, " 超时回收")
            return
        }
    }
}


```

但还是会有问题，回收完了 再有任务就处理不了。所以至少保留一个worker，并且当有新任务且达到足够的量时需要重新开启被关闭的worker。```go

type WorkerRecover struct {
    recoverList []int
    lock        sync.RWMutex
}

type Pool struct {
    //协程池worker数量 (Goroutine个数)
    workerNum int

    //协程池任务队列
    JobsChannel chan *Task

    workerRecover *WorkerRecover

    stopCtx        context.Context
    stopCancelFunc context.CancelFunc
    wg             sync.WaitGroup
}

...

// 协程池创建一个worker并且开始工作
func (p *Pool) worker(workID int) {
    ticker := time.NewTicker(time.Second * 15)
    defer ticker.Stop()
    for {
        select {
        case <-p.stopCtx.Done():
            p.wg.Done()
            fmt.Println("协程池Pool停止")
            return
        case task := <-p.JobsChannel:
            err := task.Execute()
            if err != nil {
                p.pushTask(task)
                continue
            }
            fmt.Println("worker ID ", workID, " 执行 task ID", task.args[0].(int), "任务完毕")
            // 重新设置等待15s
            ticker.Reset(time.Second * 15)
        case <-ticker.C: // 15s 都收不到任何处理就回收
            p.workerRecover.lock.Lock()
            if p.workerNum-len(p.workerRecover.recoverList) > 1 {
                fmt.Println("worker ID ", workID, " 超时回收")
                p.workerRecover.recoverList = append(p.workerRecover.recoverList, workID)
            } else {
                ticker.Reset(time.Second * 15)
            }
            p.workerRecover.lock.Unlock()
        }
    }
}

```

不能处理上新加了一个 workerRecover 用于存放回收的worker，至少保留一个worker在工作。

但是呢，如果后面task又越来越多的话，那是不是又会处理不过来了，JobsChannel是不是会阻塞了，所以想着那应该把回收的worker给重新启动。```go

// 加一个任务
func (p *Pool) pushTask(t *Task) {
    select {
    case p.JobsChannel <- t:
    default: // 阻塞了，重新启动worker
        p.Restart()
        time.Sleep(10 * time.Millisecond)
        p.JobsChannel <- t
    }
}

...

// 重启回收的worker
func (p *Pool) Restart() {
    p.workerRecover.lock.Lock()
    defer p.workerRecover.lock.Unlock()
    for _, workID := range p.workerRecover.recoverList {
        fmt.Println("worker ID ", workID, " 重开")
        go p.worker(workID)
    }
}


```

修改下tset，再测试一遍：```go

func TestCalculator_Workpool(t *testing.T) {

    //创建一个协程池
    p := NewPool(3, 10)

    //启动协程池p
    p.Run()

    //var wg sync.WaitGroup
    //wg.Add(10)
    //开一个协程 不断的向 Pool 输送task任务
    go func() {
        for i := 1; i < 10; i++ {
            task := NewTask(calcAttrTaskFunc(), i)
            p.pushTask(task)
            //wg.Done()
        }
    }()
    //wg.Wait()
    time.Sleep(20 * time.Second)
    task := NewTask(calcAttrTaskFunc(), 10)
    p.pushTask(task)
    time.Sleep(1 * time.Second)
    go func() {
        for i := 11; i <= 30; i++ {
            task := NewTask(calcAttrTaskFunc(), i)
            p.pushTask(task)
        }
    }()
    time.Sleep(300 * time.Second)
}


```

测试结果：

```

=== RUN   TestCalculator_Workpool
worker ID  2  执行 task ID 1 任务完毕
worker ID  2  执行 task ID 3 任务完毕
worker ID  2  执行 task ID 4 任务完毕
worker ID  2  执行 task ID 5 任务完毕
worker ID  2  执行 task ID 6 任务完毕
worker ID  2  执行 task ID 7 任务完毕
worker ID  2  执行 task ID 8 任务完毕
worker ID  2  执行 task ID 9 任务完毕
worker ID  1  执行 task ID 2 任务完毕
worker ID  1  超时回收
worker ID  0  超时回收
worker ID  1  执行 task ID 10 任务完毕
worker ID  1  重开
worker ID  0  重开
worker ID  0  执行 task ID 11 任务完毕
worker ID  0  执行 task ID 14 任务完毕
worker ID  0  执行 task ID 15 任务完毕
worker ID  0  执行 task ID 16 任务完毕
worker ID  0  执行 task ID 19 任务完毕
worker ID  0  执行 task ID 20 任务完毕
worker ID  0  执行 task ID 21 任务完毕
worker ID  0  执行 task ID 22 任务完毕
worker ID  0  执行 task ID 23 任务完毕
worker ID  1  执行 task ID 13 任务完毕
worker ID  0  执行 task ID 17 任务完毕
worker ID  1  执行 task ID 18 任务完毕
worker ID  2  执行 task ID 12 任务完毕
worker ID  0  执行 task ID 24 任务完毕
worker ID  0  执行 task ID 29 任务完毕
worker ID  0  执行 task ID 30 任务完毕
worker ID  0  执行 task ID 26 任务完毕
worker ID  1  执行 task ID 27 任务完毕
worker ID  2  执行 task ID 28 任务完毕
worker ID  1  执行 task ID 25 任务完毕


```

后续有空

待尝试 加入管理worker的管理者manager。

以及尝试实践在 游戏中的不同系统属性加成 并发体验

web.config加密解密工具(三个软件集合)
08-01
这里提到的"web.config加密解密工具(三个软件集合)"显然是一组可以帮助开发者处理这种加密与解密问题的工具。 首先，让我们详细了解一下`web.config`加密的原理。ASP.NET提供了一种名为“DPAPI”（Data Protection ...
详解C#App.config和Web.config加密
08-27
本文将详细介绍C#中App.config和Web.config加密的相关知识以及加密的具体步骤。 首先，需要明确Web.config适用于***应用程序的配置文件，而App.config则用于Windows窗体或控制台应用程序。加密过程涉及到.NET框架...
C# App.config和Web.config加密
u012949335的专栏
08-07 1680
使用ASP.NET提供的命令工具aspnet_regiis来创建加密命令。
不改代码，实现web.config或app.config的连接字符串加密解密
fddqfddq的专栏
07-02 736
例如，以下命令创建一个名为 NetFrameworkConfigurationKey_components1 的 RSA 密钥容器，该容器是机器级密钥容器并且可以导出。在IIS下创建walkthrough.aspx，内容如下，运行后可见解密内容。step3:更新web.config（由app.config改名而来）文件。step5:改回app.config(由web.config改回）文件。web.config部分截图如下，生成好的加密字符串。好处：不用修改代码，微软自带功能，自动解密。
C# Web.Config 加密与解密
路漫漫 修远兮
04-27 6442
用后台编码的形式对web.config数据库连接字符串或者某个节点进行加密 这里提供俩种方式： DataProtectionConfigurationProvider和RSAProtectedConfigurationProvider 编码形式 新建一个空网站添加Web窗体页面WebConfigRSA.aspx 范例：<html xmlns="http://www.w3.org/1999/x
对web.config进行加密
卟靠谱
05-26 723
更好的解决方法是将数据库连接字符串写入到web.config配置文件中，可问题是将连接字符串写入到web.config文件中之后，任何人都能打开看到所连接的数据库名和密码，又会带来安全隐患，因此为了保证数据库的安全性，我们可以通过使用微软IDE自带的命令aspnet_regiis.exe将配置文件web.config中指定的标签进行加密，如果以后想查看加密后的连接字符串，我们还可以使用aspnet_regiis.exe将加密后的字符串进行解密。在加密过程中，使用了一个基于本机的密钥。
Web.config 加密及解密
tomcat的专栏
09-01 6908
Web应用通常使用web.config存储数据库连接信息及应用的一些基础配置信息。为了安全性，我们采用一些加密手段来加密一些敏感信息，比如数据库连接字符串。 .NET 自带了一个aspnet_regiis的程序，可以帮助我们实现对web.config的加密和解密。 1.加密 1）导航到aspnet_regiis.exe目录下 cd C:\Windows\Microsoft.NET\Fram
C# 对App.config、Web.config的appSettings节点数据进行加密
qq_41863998的博客
03-19 1322
使用ASP.NET提供的命令工具aspnet_regiis来创建加密命令；并且使用aspnet_regiis加密的配置节点在读取时不需要手动解密，.Net会自行解密并返回解密后的数据。这个命令将加密config文件中的connectionStrings节点，C:\MyAppFolder指你程序部署的绝对路径。这个命令将会加密config文件中的appSettings节点，C:\MyAppFolder指你程序部署的绝对路径。这个命令将解密config中的已经加密的connectionStrings节点。
.NET 实战Web.config数据库解密及逆向技巧
ahhacker86的专栏
05-06 1661
干货满满，来自dotnet安全矩阵微信群一位师傅的投稿分享。
.NET 一款用于解密web.config配置的工具
ahhacker86的专栏
09-08 1209
Sharp4DecryptConnect 是一款专门用于解密使用 DataProtection加密的数据库连接字符串的工具。
web.config文件的加密
weixin_34323858的博客
07-05 1106
今天无聊，突然看起了petshop的代码，看到了membership连接字符串为SQLMembershipConnString，但在connectionString中并没有发现对应的名字，而且在connectionString中发现了奇怪的字符： <connectionStringsconfigProtectionProvider="RsaProtectedConfig...
.NET内网实战：通过命令行解密Web.config
ahhacker86的专栏
09-15 877
本文内容部分节选自小报童《.NET 通过命令行解密Web.config配置》，完整的文章内容请加入小报童后订阅查看。现在限时只需59元，永久买断！目前已有200+位朋友抢先预定，我们会长期更新，对.NET内网安全的朋友们请尽快订阅该报刊！
ASP.NET web.config中 数据库连接字符串加密解密
10-22
对于.NET应用程序而言，保持web.config文件的安全性是十分重要的，因此掌握如何利用aspnet_regiis.exe工具加密和解密web.config中的连接字符串对于Web开发者来说是一个不可或缺的技能。 在操作时，应当确保操作者的...
asp.net web.config加密解密方法
01-21
使用命令行工具aspnet_regiis.exe 你还能够使用aspnet_regiis.exe命令行工具来加密和解密Web.config文件配置部分，你可以在”%WINDOWSDIR%\Microsoft.Net\Framework\version”目录下找到这个工具。为了加密Web....
基于模糊控制理论的汽车EPS双层控制器设计及仿真优化研究 · 控制器设计 参考
05-29
内容概要：本文探讨了如何设计一个基于模糊控制理论的双层控制器来优化汽车EPS（电动助力转向系统）的性能。EPS系统通过电子控制技术实现转向助力，提升驾驶的舒适性和操控性。文中详细介绍了PID控制算法的局限性以及引入模糊控制器的原因。模糊控制器作为双层控制器的上层，通过模糊推理机制处理不确定性问题并优化PID控制算法。最后，通过Simulink和MATLAB模型进行了仿真验证，证明了双层控制器的有效性。 适合人群：从事汽车工程、控制系统设计的研究人员和技术人员，特别是对模糊控制理论和EPS系统感兴趣的读者。 使用场景及目标：适用于希望深入了解EPS系统优化方法的研究人员和技术人员，旨在通过模糊控制理论改进现有控制算法，提高EPS系统的鲁棒性和适应性。 阅读建议：读者可以通过本文了解模糊控制理论在EPS系统中的应用，掌握双层控制器的设计思路，并通过提供的Simulink和MATLAB模型进行仿真实验，验证控制算法的效果。
基于PSO算法的IEEE33背靠背互联配电网无功优化策略研究：总损耗与电压偏差双目标优化
05-29
内容概要：本文探讨了基于粒子群优化（PSO）算法，在三个IEEE33背靠背互联配电网中进行无功优化的方法。优化的目标是同时最小化系统的总损耗和电压偏差。文中详细介绍了优化对象（主变档位OTLC和电容器组CB）、优化算法的具体实现步骤以及关键代码片段。此外，还讨论了粒子更新、粒子筛除机制、SOP功率传输处理方法及其改进措施。实验结果显示，优化后的系统不仅显著降低了网损（日均23.7%），而且有效改善了电压分布，使OLTC和CB的动作次数均符合约束条件。 适合人群：电力系统研究人员、从事智能电网优化工作的工程师和技术人员。 使用场景及目标：适用于需要对配电网络进行无功优化的研究项目或工程应用，旨在提高配电网运行效率，减少能量损失并提升电压质量。 阅读建议：重点理解PSO算法在配电网无功优化中的具体应用方式，特别是适应度函数的设计思路、粒子更新规则以及粒子筛除策略等内容。对于实际工程项目而言，可以借鉴文中提到的技术细节和参数配置，以指导类似场景下的优化工作。
基于MATLAB实现的分数阶傅里叶变换代码
最新发布
05-29
分数阶傅里叶变换（Fractional Fourier Transform, FRFT）是对传统傅里叶变换的拓展，它通过非整数阶的变换方式，能够更有效地处理非线性信号以及涉及时频局部化的问题。在信号处理领域，FRFT尤其适用于分析非平稳信号，例如在雷达、声纳和通信系统中，对线性调频（Linear Frequency Modulation, LFM）信号的分析具有显著优势。LFM信号是一种频率随时间线性变化的信号，因其具有宽频带和良好的时频分辨率，被广泛应用于雷达和通信系统。FRFT能够更精准地捕捉LFM信号的时间和频率信息，相比普通傅里叶变换，其性能更为出色。 MATLAB是一种强大的数值计算和科学计算工具，拥有丰富的函数库和用户友好的界面。在MATLAB中实现FRFT，通常需要编写自定义函数或利用信号处理工具箱中的相关函数。例如，一个名为“frft”的文件可能是用于执行分数阶傅里叶变换的MATLAB脚本或函数，并展示其在信号处理中的应用。FRFT的正确性验证通常通过对比变换前后信号的特性来完成，比如评估信号的重构质量、信噪比等。具体而言，可以通过计算原始信号与经过FRFT处理后的信号之间的相似度，或者对比LFM信号的关键参数（如初始频率、扫频率和持续时间）是否在变换后得到准确恢复。 在MATLAB代码实现中，通常包含以下步骤：首先，生成LFM信号模型，设定其初始频率、扫频率、持续时间和采样率等参数；其次，利用自定义的frft函数对LFM信号进行分数阶傅里叶变换；接着，使用MATLAB的可视化工具（如plot或imagesc）展示原始信号的时域和频域表示，以及FRFT后的结果，以便直观对比；最后，通过计算均方误差、峰值信噪比等指标来评估FRFT的性能。深入理解FRFT的数学原理并结合MATLAB编程技巧，可以实现对LFM信号的有效分析和处理。这个代码示例不仅展示了理论知识在
自定义异常的练习,有两个定义的异常类,抛出自己写的异常
05-29
自定义异常练习
2024年测绘程序设计大赛试题：空间数据探索性分析
05-29
2024年测绘程序设计大赛试题：空间数据探索性分析
评论
添加红包

请填写红包祝福语或标题

个

红包个数最小为10个

元

红包金额最低5元

当前余额3.43元 前往充值 >
需支付：10.00元
成就一亿技术人!
领取后你会自动成为博主和红包主的粉丝 规则
hope_wisdom
发出的红包
实付元
使用余额支付
点击重新获取
扫码支付
钱包余额 0

抵扣说明：

1.余额是钱包充值的虚拟货币，按照1:1的比例进行支付金额的抵扣。
2.余额无法直接购买下载，可以购买VIP、付费专栏及课程。

余额充值