我写了一个简单的并发调度器，但它似乎在高并发时有性能问题。

这是代码(调度器+并发速率限制器测试):

package main

import (
    "flag"
    "fmt"
    "log"
    "os"
    "runtime"
    "runtime/pprof"
    "sync"
    "time"

    "github.com/gomodule/redigo/redis"
)

// a scheduler is composed by load function and process function
type Scheduler struct {
    // query channel
    reqChan chan interface{}

    // max routine
    maxRoutine int

    // max routine
    chanSize int

    wg sync.WaitGroup

    // query process function
    process func(interface{})
}

func NewScheduler(maxRoutine int, chanSize int, process func(interface{})) *Scheduler {
    s := &Scheduler{}
    if maxRoutine == 0 {
        s.maxRoutine = 10
    } else {
        s.maxRoutine = maxRoutine
    }

    if chanSize == 0 {
        s.chanSize = 100
    } else {
        s.chanSize = chanSize
    }

    s.reqChan = make(chan interface{}, s.chanSize)
    s.process = process
    return s
}

func (s *Scheduler) Start() {
    // start process
    for i := 0; i < s.maxRoutine; i++ {
        go s.processRequest()
    }
}

func (s *Scheduler) processRequest() {
    for {
        select {
        case req := <-s.reqChan:
            s.process(req)
            s.wg.Done()
        }
    }
}

func (s *Scheduler) Enqueue(req interface{}) {
    select {
    case s.reqChan <- req:
        s.wg.Add(1)
    }
}

func (s *Scheduler) Wait() {
    s.wg.Wait()
}

const script = `
local required_permits = tonumber(ARGV[2]);

local next_free_micros = redis.call('hget',KEYS[1],'next_free_micros');
if(next_free_micros == false) then
    next_free_micros = 0;
else
    next_free_micros = tonumber(next_free_micros);
end;

local time = redis.call('time');
local now_micros = tonumber(time[1])*1000000 + tonumber(time[2]);

--[[
try aquire
--]]
if(ARGV[3] ~= nil) then
    local micros_to_wait = next_free_micros - now_micros;
    if(micros_to_wait > tonumber(ARGV[3])) then
        return micros_to_wait;
    end
end

local stored_permits = redis.call('hget',KEYS[1],'stored_permits');
if(stored_permits == false) then
    stored_permits = 0;
else
    stored_permits = tonumber(stored_permits);
end

local stable_interval_micros = 1000000/tonumber(ARGV[1]);
local max_stored_permits = tonumber(ARGV[1]);

if(now_micros > next_free_micros) then
    local new_stored_permits = stored_permits + (now_micros - next_free_micros) / stable_interval_micros;
    if(max_stored_permits < new_stored_permits) then
        stored_permits = max_stored_permits;
    else
        stored_permits = new_stored_permits;
    end
    next_free_micros = now_micros;
end

local moment_available = next_free_micros;
local stored_permits_to_spend = 0;
if(stored_permits < required_permits) then
    stored_permits_to_spend = stored_permits;
else
    stored_permits_to_spend = required_permits;
end
local fresh_permits = required_permits - stored_permits_to_spend;
local wait_micros = fresh_permits * stable_interval_micros;

redis.replicate_commands();
redis.call('hset',KEYS[1],'stored_permits',stored_permits - stored_permits_to_spend);
redis.call('hset',KEYS[1],'next_free_micros',next_free_micros + wait_micros);
redis.call('expire',KEYS[1],10);

return moment_available - now_micros;
`

var (
    rlScript *redis.Script
)

func init() {
    rlScript = redis.NewScript(1, script)
}

func take(key string, qps, requires int, pool *redis.Pool) (int64, error) {
    c := pool.Get()
    defer c.Close()

    var err error
    if err := c.Err(); err != nil {
        return 0, err
    }

    reply, err := rlScript.Do(c, key, qps, requires)
    if err != nil {
        return 0, err
    }
    return reply.(int64), nil
}

func NewRedisPool(address, password string) *redis.Pool {
    pool := &redis.Pool{
        MaxIdle:     50,
        IdleTimeout: 240 * time.Second,
        TestOnBorrow: func(c redis.Conn, t time.Time) error {
            _, err := c.Do("PING")
            return err
        },
        Dial: func() (redis.Conn, error) {
            return dial("tcp", address, password)
        },
    }
    return pool
}

func dial(network, address, password string) (redis.Conn, error) {
    c, err := redis.Dial(network, address)
    if err != nil {
        return nil, err
    }
    if password != "" {
        if _, err := c.Do("AUTH", password); err != nil {
            c.Close()
            return nil, err
        }
    }
    return c, err
}

func main() {
    var cpuprofile = flag.String("cpuprofile", "", "write cpu profile `file`")
    var memprofile = flag.String("memprofile", "", "write memory profile to `file`")
    flag.Parse()
    if *cpuprofile != "" {
        f, err := os.Create(*cpuprofile)
        if err != nil {
            log.Fatal("could not create CPU profile: ", err)
        }
        if err := pprof.StartCPUProfile(f); err != nil {
            log.Fatal("could not start CPU profile: ", err)
        }
        defer pprof.StopCPUProfile()
    }

    test()

    if *memprofile != "" {
        f, err := os.Create(*memprofile)
        if err != nil {
            log.Fatal("could not create memory profile: ", err)
        }
        runtime.GC() // get up-to-date statistics
        if err := pprof.WriteHeapProfile(f); err != nil {
            log.Fatal("could not write memory profile: ", err)
        }
        f.Close()
    }
}

func test() {
    pool := NewRedisPool("127.0.0.1:6379", "")

    s1 := NewScheduler(10000, 1000000, func(r interface{}) {
        take("xxx", 1000000, 1, pool)
    })
    s1.Start()

    start := time.Now()
    for i := 0; i < 100000; i++ {
        s1.Enqueue(i)
    }
    fmt.Println(time.Since(start))
    s1.Wait()
    fmt.Println(time.Since(start))
}

问题是在 10000 个例程时，有时即使没有向 redis 发送命令(检查“redis-cli monitor”)程序也会卡住，我的系统最大打开文件数设置为 20000。

我做了分析，很多“syscall.Syscall”，有人可以提供任何建议吗？我的调度程序有问题吗？

最佳答案

从表面上看，我唯一有疑问的是递增 WaitGroup 的排序和工作排队:

func (s *Scheduler) Enqueue(req interface{}) {
    select {
    case s.reqChan <- req:
        s.wg.Add(1)
    }
}

我认为上述内容在如此大的工作量下不会在实践中造成太大问题，但我认为这可能是一种合乎逻辑的竞争条件。在较低的并发级别和较小的工作量下，它可能会将消息排入队列，然后切换到开始处理该消息的 goroutine，然后是 WaitGroup 中的工作。

接下来你确定 process 方法是线程安全的吗？？我假设基于 redis go 文档，使用 go run -race 运行是否有任何输出？

在某些时候，性能下降是完全合理和预期的。我建议开始性能测试，看看延迟和吞吐量从哪里开始下降:

可能是 10、100、500、1000、2500、5000、10000 或任何有意义的池。 IMO 看起来有 3 个重要变量需要调整:

• 工作人员池大小
• 工作队列缓冲区大小
• Redis MaxActive

跳出来的最大的东西就是看起来像redis.Pool is configured to allow an unbounded number of connections :

 pool := &redis.Pool{
        MaxIdle:     50,
        IdleTimeout: 240 * time.Second,
        TestOnBorrow: func(c redis.Conn, t time.Time) error {
            _, err := c.Do("PING")
            return err
        },
        Dial: func() (redis.Conn, error) {
            return dial("tcp", address, password)
        },
    }

// Maximum number of connections allocated by the pool at a given time. // When zero, there is no limit on the number of connections in the pool. MaxActive int

我个人会尝试了解相对于您的工作线程池大小，性能在何时何地开始下降。这可能会让您更容易理解您的程序受什么约束。

关于golang + redis 并发调度器性能问题，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/55768699/