登录
首页 >  Golang >  Go问答

使用 BLPOP 处理 Redis 队列会导致单元测试中出现竞争条件吗?

来源:stackoverflow

时间:2024-04-06 23:00:32 403浏览 收藏

大家好,今天本人给大家带来文章《使用 BLPOP 处理 Redis 队列会导致单元测试中出现竞争条件吗?》,文中内容主要涉及到,如果你对Golang方面的知识点感兴趣,那就请各位朋友继续看下去吧~希望能真正帮到你们,谢谢!

问题内容

我正在尝试用 go 实现 redis 电子书第 6.4.1 章中描述的先进先出任务队列。出于测试目的,我将 commandexecutor 接口传递给“worker”函数,如下所示:

package service

import (
    "context"

    "github.com/gomodule/redigo/redis"
    "github.com/pkg/errors"
    "github.com/sirupsen/logrus"
)

const commandsqueue = "queuedcommands:"

var pool = redis.pool{
    maxidle:   50,
    maxactive: 1000,
    dial: func() (redis.conn, error) {
        conn, err := redis.dial("tcp", ":6379")
        if err != nil {
            logrus.witherror(err).fatal("initialize redis pool")
        }
        return conn, err
    },
}

// commandexecutor executes a command
type commandexecutor interface {
    execute(string) error
}

func processqueue(ctx context.context, done chan<- struct{}, executor commandexecutor) error {
    rc := pool.get()
    defer rc.close()

    for {
        select {
        case <-ctx.done():
            done <- struct{}{}
            return nil
        default:
            // if the commands queue does not exist, blpop blocks until another client
            // performs an lpush or rpush against it. the timeout argument of zero is
            // used to block indefinitely.
            reply, err := redis.strings(rc.do("blpop", commandsqueue, 0))
            if err != nil {
                logrus.witherror(err).errorf("blpop %s %d", commandsqueue, 0)
                return errors.wrapf(err, "blpop %s %d", commandsqueue, 0)
            }

            if len(reply) < 2 {
                logrus.errorf("expected a reply of length 2, got one of length %d", len(reply))
                return errors.errorf("expected a reply of length 2, got one of length %d", len(reply))
            }

            // blpop returns a two-element multi-bulk with the first element being the
            // name of the key where an element was popped and the second element
            // being the value of the popped element (cf. https://redis.io/commands/blpop#return-value)
            if err := executor.execute(reply[1]); err != nil {
                return errors.wrapf(err, "execute scheduled command: %s", reply[0])
            }
            done <- struct{}{}
        }
    }
}

我使用此代码制作了一个小型示例存储库 https://github.com/kurtpeek/process-queue 并尝试进行单元测试。对于单元测试,我有两个相同的测试(具有不同的名称):

package service

import (
    "context"
    "testing"

    "github.com/stretchr/testify/assert"
    "github.com/stretchr/testify/require"
)

func testprocessqueue(t *testing.t) {
    ctx, cancel := context.withcancel(context.background())
    defer cancel()

    executor := &commandexecutormock{
        executefunc: func(string) error {
            return nil
        },
    }

    done := make(chan struct{})
    go processqueue(ctx, done, executor)

    rc := pool.get()
    defer rc.close()

    _, err := rc.do("rpush", commandsqueue, "foobar")
    require.noerror(t, err)

    <-done

    assert.exactly(t, 1, len(executor.executecalls()))
    assert.exactly(t, "foobar", executor.executecalls()[0].in1)
}

func testprocessqueue2(t *testing.t) {
    ctx, cancel := context.withcancel(context.background())
    defer cancel()

    executor := &commandexecutormock{
        executefunc: func(string) error {
            return nil
        },
    }

    done := make(chan struct{})
    go processqueue(ctx, done, executor)

    rc := pool.get()
    defer rc.close()

    _, err := rc.do("rpush", commandsqueue, "foobar")
    require.noerror(t, err)

    <-done

    assert.exactly(t, 1, len(executor.executecalls()))
    assert.exactly(t, "foobar", executor.executecalls()[0].in1)
}

其中 commandexecutormock 是使用 moq 生成的。如果我单独运行每个测试,它们会通过:

~/g/s/g/k/process-queue> go test ./... -v -run testprocessqueue2
=== run   testprocessqueue2
--- pass: testprocessqueue2 (0.00s)
pass
ok      github.com/kurtpeek/process-queue/service   0.243s

但是,如果我运行所有测试,第二个测试就会超时:

~/g/s/g/k/process-queue> 
go test ./... -v -timeout 10s
=== RUN   TestProcessQueue
--- PASS: TestProcessQueue (0.00s)
=== RUN   TestProcessQueue2
panic: test timed out after 10s

看起来,当第二个测试运行时,第一个测试中启动的 goroutine 仍在运行,并且 blpop 从队列中执行命令,因此第二个测试中的 <-done 行无限期地阻塞。尽管在第一个测试的父上下文上调用了 cancel(),但仍然如此。

如何“隔离”这些测试,以便它们一起运行时都能通过? (我尝试将 -p 1 标志传递给 go test 但无济于事)。


解决方案


尽管在第一个测试的父上下文上调用了 cancel() ,但还是如此。

写入 done 和调用 cancel() 之间有一些时间,这意味着第一个测试可能(并且确实)进入第二个 for/select 迭代,而不是在 <-ctx.done() 上退出。更具体地说,测试代码在取消之前包含 2 个断言:

assert.exactly(t, 1, len(executor.executecalls()))
    assert.exactly(t, "foobar", executor.executecalls()[0].in1)

只有此时 defer cancel() 才会启动,这似乎来不及取消第一个 go 例程的上下文了。

如果您在读取 ​​done 之前移动 cancel() 调用,则测试通过:

func TestProcessQueue(t *testing.T) {
    ctx, cancel := context.WithCancel(context.Background())

    executor := &CommandExecutorMock{
        ExecuteFunc: func(string) error {
            return nil
        },
    }

    done := make(chan struct{})
    go processQueue(ctx, done, executor)

    rc := pool.Get()
    defer rc.Close()

    _, err := rc.Do("RPUSH", commandsQueue, "foobar")
    require.NoError(t, err)

    cancel() // note this change right here
    <-done

    assert.Exactly(t, 1, len(executor.ExecuteCalls()))
    assert.Exactly(t, "foobar", executor.ExecuteCalls()[0].In1)
}

今天带大家了解了的相关知识,希望对你有所帮助;关于Golang的技术知识我们会一点点深入介绍,欢迎大家关注golang学习网公众号,一起学习编程~

声明:本文转载于:stackoverflow 如有侵犯,请联系study_golang@163.com删除
相关阅读
更多>
最新阅读
更多>
课程推荐
更多>