学习笔记 | Web系统架构 Golang Lamp Nginx MySQL | Page 9

最近在用go重构iot中的一个服务时，发现库 rocketmq-client-go@v2.0.0-rc1 在初始化消费客户端实现时，实现的极其优雅，代码见https://github.com/apache/rocketmq-client-go/blob/v2.0.0-rc1/examples/consumer/simple/main.go#L32

c, _ := rocketmq.NewPushConsumer(
    consumer.WithGroupName("testGroup"),
    consumer.WithNameServer([]string{"127.0.0.1:9876"}),
)
err := c.Subscribe("test", consumer.MessageSelector{}, func(ctx context.Context,
    msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
    for i := range msgs {
        fmt.Printf("subscribe callback: %v n", msgs[i])
    }

    return consumer.ConsumeSuccess, nil
})

这里创建结构体 rocketmq.NewPushConsumer() 的时候，与我们平时的写法不同并没有写死结构体的字段名和值，而是每个属性都使用了一个函数来实现了，同时也不用考虑属性字段的位置关系，比起以前写kv键值对的方法实在是太灵活了。
我们再看一下其中一个WithGroupName()函数的实现方法

func WithGroupName(group string) Option {
    return func(opts *consumerOptions) {
        if group == "" {
            return
        }
        opts.GroupName = group
    }
}

传递的参数为consumerOptions指针类型，这里用到了一个匿名函数，返回的类型为Option(定义 type Option func(*consumerOptions) )。看到这里大概明白实现原理了吧。
为了确认我们的判断，我们再看一下 rocketmq.NewPushConsumer()函数

func NewPushConsumer(opts ...consumer.Option) (PushConsumer, error) {
	return consumer.NewPushConsumer(opts...)
}

这里直接调用了另一个 consumer包里的 NewPushConsumer() 函数，其内容如下（ 为了方便理解，在代码里直接加了注释）

// opts 为不定参数
func NewPushConsumer(opts ...Option) (*pushConsumer, error) {
    // defaultPushConsumerOptions 见 https://github.com/apache/rocketmq-client-go/blob/7308bc94369320195652243059f63c71bfafc74b/consumer/option.go#L109
	defaultOpts := defaultPushConsumerOptions()

    // 实现动态的给 defaultOpts 属性赋值
	for _, apply := range opts {
        // 重点！重点！重点！传递的是一个指针
        // apply 是一个以 WithXxx 开头的函数的返回值即匿名函数，如
        // func WithGroupName(group string) Option{
        //  return func(opts *consumerOptions) {
        //    if group == "" {
        //      return
        //    }
        //    opts.GroupName = group
        //   }
        // }
		apply(&defaultOpts)
	}
	srvs, err := internal.NewNamesrv(defaultOpts.NameServerAddrs)
	if err != nil {
		return nil, errors.Wrap(err, "new Namesrv failed.")
	}
	if !defaultOpts.Credentials.IsEmpty() {
		srvs.SetCredentials(defaultOpts.Credentials)
	}
	defaultOpts.Namesrv = srvs

	if defaultOpts.Namespace != "" {
		defaultOpts.GroupName = defaultOpts.Namespace + "%" + defaultOpts.GroupName
	}

	dc := &defaultConsumer{
		client:         internal.GetOrNewRocketMQClient(defaultOpts.ClientOptions, nil),
		consumerGroup:  defaultOpts.GroupName,
		cType:          _PushConsume,
		state:          int32(internal.StateCreateJust),
		prCh:           make(chan PullRequest, 4),
		model:          defaultOpts.ConsumerModel,
		consumeOrderly: defaultOpts.ConsumeOrderly,
		fromWhere:      defaultOpts.FromWhere,
		allocate:       defaultOpts.Strategy,
		option:         defaultOpts,
		namesrv:        srvs,
	}

	p := &pushConsumer{
		defaultConsumer: dc,
		subscribedTopic: make(map[string]string, 0),
		queueLock:       newQueueLock(),
		done:            make(chan struct{}, 1),
		consumeFunc:     utils.NewSet(),
	}
	dc.mqChanged = p.messageQueueChanged
	if p.consumeOrderly {
		p.submitToConsume = p.consumeMessageOrderly
	} else {
		p.submitToConsume = p.consumeMessageCurrently
	}

	p.interceptor = primitive.ChainInterceptors(p.option.Interceptors...)

	return p, nil
}

func defaultPushConsumerOptions() consumerOptions {
	opts := consumerOptions{
        // ClientOptions 重点字段
		ClientOptions:              internal.DefaultClientOptions(),

		Strategy:                   AllocateByAveragely,
		MaxTimeConsumeContinuously: time.Duration(60 * time.Second),
		RebalanceLockInterval:      20 * time.Second,
		MaxReconsumeTimes:          -1,
		ConsumerModel:              Clustering,
		AutoCommit:                 true,
	}

    // 这里只对 GroupName 属性进行了初始化，未指定的则使用结构体 ClientOptions 字段类型的默认值
	opts.ClientOptions.GroupName = "DEFAULT_CONSUMER"
	return opts
}

同时他有诸多以 WithXxx 开头的方法体，如 WithGroupName()、WithNameServer()、WithInstance()。

我们再找到 consumerOptions 结构体的定义，最终找到定义如下

type ClientOptions struct {
	GroupName         string
	NameServerAddrs   primitive.NamesrvAddr
	NameServerDomain  string
	Namesrv           *namesrvs
	ClientIP          string
	InstanceName      string
	UnitMode          bool
	UnitName          string
	VIPChannelEnabled bool
	RetryTimes        int
	Interceptors      []primitive.Interceptor
	Credentials       primitive.Credentials
	Namespace         string
}

发现这些才是我们平时使用的属性字段。

这里的实现方法可能还不太好容易理解，强烈推荐阅读 Uber Go 语言编码规范

总结：

动态灵活的实现结构体的属性配置，是通过将每个属性分离出来，重构为一个独立的函数，一般以WithXxx开头，将实现委托给了返回的匿名函数来实现，原理伪代码如下

func WithOptionName(*options Options, optionValue interface{}) {
    options.OptionName = optionValue
}

推荐阅读

根据上面的方法，我们就对 github.com/go-redis/redis 这个库连接数据库配置项的重构。

package main

import (
	"fmt"
	"log"
	"github.com/go-redis/redis"
)

type Option func(*redis.Options)

func NewRedisOptions(opts ...Option) *redis.Options {
	// 默认选项
	defaultOptions := redis.Options{}

	// 遍历指定项
	for _, apply := range opts {
		apply(&defaultOptions)
	}

	return &defaultOptions
}

// address
func WithAddr(addr string) Option {
	return func(opts *redis.Options) {
		opts.Addr = addr
	}
}

// password
func WithPassword(password string) Option {
	return func(opts *redis.Options) {
		opts.Password = password
	}
}

// db
func WithDB(i int) Option {
	return func(opts *redis.Options) {
		opts.DB = i
	}
}

func main() {
	opts := NewRedisOptions(
		WithAddr("127.0.0.1:6379"),
		WithPassword(""),
		WithDB(6),
	)
	RedisClient := redis.NewClient(opts)
	ret, err := RedisClient.Ping().Result()
	if err != nil {
		log.Fatal("连接Redis Server 失败：", err)
	}
	fmt.Printf("%#v", ret)
}