带有通道的 Golang 测试不退出

Question

以下 Golang 测试永远不会退出。我怀疑这与频道僵局有关，但作为菜鸟，我不太确定。

const userName = "xxxxxxxxxxxx"

func TestSynchroninze(t *testing.T) {
    c, err := channel.New(github.ChannelName, authToken)
    if err != nil {
        t.Fatalf("Could not create channel: %s", err)
        return
    }

    state := channel.NewState(nil)
    ctx := context.Background()
    ctx = context.WithValue(ctx, "userId", userName)
    user := api.User{}

    output, errs := c.Synchronize(state, ctx)

    if err = <-errs; err != nil {
        t.Fatalf("Error performing synchronize: %s", err)
        return
    }

    for o := range output {
        switch oo := o.Data.(type) {
        case api.User:
            user = oo
            glog.Infof("we have a USER %s\n", user)
        default:
            t.Errorf("Encountered unexpected data type: %T", oo)
        }
    }
}

这里是正在测试的方法。

type github struct {
    client *api.Client
}

func newImplementation(t auth.UserToken) implementation.Implementation {
    return &github{client: api.NewClient(t)}
}

// -------------------------------------------------------------------------------------

const (
    kLastUserFetch = "lastUserFetch"
)

type synchronizeFunc func(implementation.MutableState, chan *implementation.Output, context.Context) error

// -------------------------------------------------------------------------------------

    func (g *github) Synchronize(state implementation.MutableState, ctx context.Context) (<-chan *implementation.Output, <-chan error) {
        output := make(chan *implementation.Output)
        errors := make(chan error, 1) // buffer allows preflight errors

        // Close output channels once we're done
        defer func() {
            go func() {
                // wg.Wait()

                close(errors)
                close(output)
            }()
        }()

        err := g.fetchUser(state, output, ctx)
        if err != nil {
            errors <- err
        }

        return output, errors
    }

func (g *github) fetchUser(state implementation.MutableState, output chan *implementation.Output, ctx context.Context) error {
    var err error

    var user = api.User{}
    userId, _ := ctx.Value("userId").(string)
    user, err = g.client.GetUser(userId, ctx.Done())

    if err == nil {
        glog.Info("No error in fetchUser")
        output <- &implementation.Output{Data: user}
        state.SetTime(kLastUserFetch, time.Now())
    }

    return err
}

func (c *Client) GetUser(id string, quit <-chan struct{}) (user User, err error) {
    // Execute request
    var data []byte
    data, err = c.get("users/"+id, nil, quit)
    glog.Infof("USER DATA %s", data)

    // Parse response
    if err == nil && len(data) > 0 {
        err = json.Unmarshal(data, &user)

        data, _ = json.Marshal(user)
    }

    return
}

这是我在控制台中看到的（删除了大部分用户详细信息）

I1228 13:25:05.291010   21313 client.go:177] GET https://api.github.com/users/xxxxxxxx
I1228 13:25:06.010085   21313 client.go:36] USER DATA {"login":"xxxxxxxx","id":00000000,"avatar_url":"https://avatars.githubusercontent.com/u/0000000?v=3",...}
I1228 13:25:06.010357   21313 github.go:90] No error in fetchUser

==========编辑==============

这是api 包的相关部分。

package api

type Client struct {
    authToken auth.UserToken
    http      *http.Client
}

func NewClient(authToken auth.UserToken) *Client {
    return &Client{
        authToken: authToken,
        http:      auth.NewClient(authToken),
    }
}




// -------------------------------------------------------------------------------------
type User struct {
    Id             int    `json:"id,omitempty"`
    Username       string `json:"login,omitempty"`
    Email          string `json:"email,omitempty"`
    FullName       string `json:"name,omitempty"`
    ProfilePicture string `json:"avatar_url,omitempty"`
    Bio            string `json:"bio,omitempty"`
    Website        string `json:"blog,omitempty"`
    Company        string `json:"company,omitempty"`
}

还有channel 包

package channel

type Channel struct {
    implementation.Descriptor
    imp implementation.Implementation
}

// New returns a channel implementation with a given name and auth token.
func New(name string, token auth.UserToken) (*Channel, error) {
    if desc, ok := implementation.Lookup(name); ok {
        if imp := implementation.New(name, token); imp != nil {
            return &Channel{Descriptor: desc, imp: imp}, nil
        }
    }

    return nil, ErrInvalidChannel
}

还有implementation 包...

package implementation

import "golang.org/x/net/context"

// -------------------------------------------------------------------------------------

// Implementation is the interface implemented by subpackages.
type Implementation interface {
    // Synchronize performs a synchronization using the given state. A context parameters
    // is provided to provide cancellation as well as implementation-specific behaviors.
    //
    // If a fatal error occurs (see package error definitions), the state can be discarded
    // to prevent the persistence of an invalid state.
    Synchronize(state MutableState, ctx context.Context) (<-chan *Output, <-chan error)

    // FetchDetails gets details for a given timeline item. Any changes to the TimelineItem
    // (including the Meta value) will be persisted.
    FetchDetails(item *TimelineItem, ctx context.Context) (interface{}, error)
}

======编辑#2=======

这是原来的Synchronize 方法。我在测试中删除了一些细节以尝试简化问题。通过删除 go func 调用，我相信我引入了一个新问题，可能会造成混淆。

这是原始的 Synchronize 方法。 Wait Groups 和包含单个函数的函数数组有些东西，因为这种方法最终会同步多个函数。

func (g *github) Synchronize(state implementation.MutableState, ctx context.Context) (<-chan *implementation.Output, <-chan error) {
    wg := sync.WaitGroup{}
    output := make(chan *implementation.Output)
    errors := make(chan error, 1) // buffer allows preflight errors

    // Close output channels once we're done
    defer func() {
        go func() {
            wg.Wait()

            close(errors)
            close(output)
        }()
    }()

    // Perform fetch functions in separate routines
    funcs := []synchronizeFunc{
        g.fetchUser,
    }

    for _, f := range funcs {
        wg.Add(1)

        go func(f synchronizeFunc) {
            defer wg.Done()

            if err := f(state, output, ctx); err != nil {
                errors <- err
            }
        }(f)
    }

    glog.Info("after go sync...")

    return output, errors
}

Answer 1

我认为这两个问题都在

     output <- &implementation.Output{Data: user}

通道没有缓冲区。它将阻塞，直到其他一些 goroutine 从中读取。但是在你的代码中是同一个 goroutine 所以它会阻塞。

第二个：

    // Close output channels once we're done
    defer func() {
        go func() {
            // wg.Wait()

            close(errors)
            close(output)
        }()
    }()

当例程退出时，您启动一​​个 go 例程。 goroutine 被调度，函数返回但它从不调用 goroutine。

我建议将所有这些逻辑统一为一个：

 func (g *github) Synchronize(state implementation.MutableState, ctx context.Context) (<-chan *implementation.Output, <-chan error) {
    output := make(chan *implementation.Output)
    errors := make(chan error, 1) // buffer allows preflight errors

    go func() {
      defer close(output)
      defer close(errors)
      err := g.fetchUser(state, output, ctx)
      if err != nil {
        errors <- err
      }
   }()        
   return output, errors
  }