基于Golang 高并发问题的解决方案

时间:2022-06-01 20:50:00

Golang 高并发问题的解决

 

Golang在高并发问题上,由于协程的使用,相对于其他编程语言,已经有了很大的优势,即相同的配置上,Golang可以以更低的代价处理更多的线程,同样的线程数,占用更低的资源!及时这样,只是解决了一部分问题而已,因为在每个协程里,处理逻辑还是会有问题。

高并发时,还是要考虑服务器所能承受的最大压力,数据库读取时的io问题,连接数问题,带宽问题等等

研究了一下并发解决方案,在此记录一下

参考文章:Handling 1 Million Requests per Minute with Go

地址:http://marcio.io/2015/07/handling-1-million-requests-per-minute-with-golang/

代码如下:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
//==================================
//  * Name:Jerry
//  * Tel:18017448610
//  * DateTime:2019/2/24 14:02
//==================================
package main
import (
    "github.com/lunny/log"
    "runtime"
    "sync"
    "time"
)
//工厂模型
type Factory struct {
    Wg        *sync.WaitGroup //任务监控系统
    MaxWorker int             //最大机器数
    MaxJobs   int             //最大工作数量
    JobQueue  chan int        //工作队列管道
    Quit      chan bool       //是否关闭机器
}
//创建工厂模型
func NewFactory(maxWorker int, wg *sync.WaitGroup) Factory {
    return Factory{
        Wg:        wg,                        //引用任务监控系统
        MaxWorker: maxWorker,                 //机器数量(数量多少,根据服务器性能而定)
        JobQueue:  make(chan int, maxWorker), //工作管道,数量大于等于机器数
        Quit:      make(chan bool),
    }
}
//设置最大订单数量
func (f *Factory) SetMaxJobs(taskNum int) {
    f.MaxJobs = taskNum
}
//开始上班
func (f *Factory) Start() {
    //机器开机,MaxWorker
    for i := 0; i < f.MaxWorker; i++ {
        //每一台机器开启后,去工作吧
        go func() {
            //等待下发命令
            for {
                select {
                case i := <-f.JobQueue:
                    //接到工作,开工!
                    f.doWork(i)
                case <-f.Quit:
                    log.Println("机器关机")
                    return
                }
            }
        }()
    }
}
//分配每个任务到管道中
func (f *Factory) AddTask(taskNum int) {
    //系统监控任务 +1
    f.Wg.Add(1)
    //分配任务到管道中
    f.JobQueue <- taskNum
}
//模拟耗时工作
func (f *Factory) doWork(taskNum int) {
    //生产产品的工作
    time.Sleep(200 * time.Millisecond)
    //完成工作报告
    f.Wg.Done()
    //log.Println("完工:", taskNum)
}
//创建工厂
func Begin() {
    //配置工作核数
    gomaxprocs := runtime.GOMAXPROCS(runtime.NumCPU())
    log.Println("核数:", gomaxprocs)
    //配置监控系统
    wg := new(sync.WaitGroup)
    //开工厂
    factory := NewFactory(1000, wg)
    //订单量
    factory.SetMaxJobs(10000)
    //开始上班
    factory.Start()
    log.Println("开始生产")
    //讲所有的订单,添加到任务队列
    for i := 0; i < factory.MaxJobs; i++ {
        factory.AddTask(i)
    }
    factory.Wg.Wait()
    log.Println("所有订单任务生产完成")
}

测试代码及结果

上面代码中,MaxWorker的数量很重要,取决于服务器所能承受的压力,当然也不能无限增大,合理数值效率最高(具体多少合适,自己测试)

代码:

?
1
2
3
func Benchmark_Begin(b *testing.B) {
 Begin()
}

结果:

1000台机器(协程),10000的工作量,我的个人PC测试结果如下:

2019/02/26 16:42:31 核数: 4

2019/02/26 16:42:31 开始生产

2019/02/26 16:42:33 所有订单任务生产完成

goos: windows

goarch: amd64

pkg: day11

Benchmark_hight2-4 1 2035574000 ns/op

PASS

Process finished with exit code 0

总结:

 

此方法仅仅是在代码层面解决一定的问题,高并发 产生的原因还包括其他原因,如带宽,数据库读取速度等等,还需加大带宽,多级数据库,优化数据的检索等等方法

补充:golang 高并发任务处理方案

这个主要用golang 的chan 和routine属性做的,比很多语言方便多了,可以参考参考

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
//任务的请求
type MtaskRequest struct {
    Ceshi int
    // [redacted]
}
 
//job队列+work池
var (
    MaxWorker = os.Getenv("MAX_WORKERS")
    MaxQueue  = os.Getenv("MAX_QUEUE")
)
 
// Job represents the job to be run
type Job struct {
    MtaskRequest MtaskRequest
}
 
// A buffered channel that we can send work requests on.
 
// var JobQueue chan Job ---这样申明会卡主,没有初始化
var JobQueue = make(chan Job)
 
// Worker represents the worker that executes the job
type Worker struct {
    WorkerPool chan chan Job
    JobChannel chan Job
    quit       chan bool
}
 
func NewWorker(workerPool chan chan Job) Worker {
    return Worker{
        WorkerPool: workerPool,
        JobChannel: make(chan Job),
        quit:       make(chan bool)}
}
 
// Stop signals the worker to stop listening for work requests.
func (w Worker) Stop() {
    go func() {
        w.quit <- true
    }()
}
 
type Dispatcher struct {
    // A pool of workers channels that are registered with the dispatcher
    WorkerPool chan chan Job
    maxWorkers int
}
 
func NewDispatcher(maxWorkers int) *Dispatcher {
    pool := make(chan chan Job, maxWorkers)
    return &Dispatcher{WorkerPool: pool, maxWorkers: maxWorkers}
}
 
// Start method starts the run loop for the worker, listening for a quit channel in
// case we need to stop it
func (w Worker) Start() {
    go func() {
        for {
            // register the current worker into the worker queue.
            w.WorkerPool <- w.JobChannel
            select {
            case <-w.JobChannel:
                time.Sleep(5 * time.Second)
                // we have received a work request.
                fmt.Println("调起worker")
            case <-w.quit:
                // we have received a signal to stop
                return
                //不能写default
            }
        }
    }()
}
 
func (d *Dispatcher) Run() {
    //启动一定数量的worker
    fmt.Println("启动一定数量的worker")
    for i := 0; i < d.maxWorkers; i++ {
        worker := NewWorker(d.WorkerPool)
        worker.Start()
    }
 
    go d.dispatch()
}
 
//分派任务
func (d *Dispatcher) dispatch() {
    for {
        select {
        case job := <-JobQueue: //接收一个job请求
            fmt.Println("JobQueue 收到请求")
 
            go func(job Job) {
                // try to obtain a worker job channel that is available.
                // this will block until a worker is idle
                jobChannel := <-d.WorkerPool
                // dispatch the job to the worker job channel
                jobChannel <- job
            }(job)
        }
    }
}
 
//接收到红包数据
func (this *TaskRedbao) UserGetRedbao(red_id, uid, shop_id, rand_arr, Amoney string) error {
    fmt.Println("收到 接收到红包数据 http请求")
    mtaskRequest := MtaskRequest{67}
    work := Job{MtaskRequest: mtaskRequest}
 
    JobQueue <- work
    return nil
}

以上为个人经验,希望能给大家一个参考,也希望大家多多支持服务器之家。如有错误或未考虑完全的地方,望不吝赐教。

原文链接:https://blog.csdn.net/ming2316780/article/details/87938016