云原生 | go-micro@v4.9.0源码阅读

service Helloworld {
rpc Call(CallRequest) returns (CallResponse) {}
rpc ClientStream(stream ClientStreamRequest) returns (ClientStreamResponse) {}
rpc ServerStream(ServerStreamRequest) returns (stream ServerStreamResponse) {}
rpc BidiStream(stream BidiStreamRequest) returns (stream BidiStreamResponse) {}
}

type Service interface {
// The service name
Name() string
// Init initialises options
Init(...Option)
// Options returns the current options
Options() Options
// Client is used to call services
Client() client.Client
// Server is for handling requests and events
Server() server.Server
// Run the service
Run() error
// The service implementation
String() string
}

type service struct {
opts Options
once sync.Once
}

type Server interface {
// Initialise options
Init(...Option) error
// Retrieve the options
Options() Options
// Register a handler
Handle(Handler) error
// Create a new handler
NewHandler(interface{}, ...HandlerOption) Handler
// Create a new subscriber
NewSubscriber(string, interface{}, ...SubscriberOption) Subscriber
// Register a subscriber
Subscribe(Subscriber) error
// Start the server
Start() error
// Stop the server
Stop() error
// Server implementation
String() string
}

type rpcServer struct {
router *router
exit chan chan error
sync.RWMutex
opts Options
handlers map[string]Handler
subscribers map[Subscriber][]broker.Subscriber
// marks the serve as started
started bool
// used for first registration
registered bool
// subscribe to service name
subscriber broker.Subscriber
// graceful exit
wg *sync.WaitGroup
rsvc *registry.Service
}

type Client interface {
Init(...Option) error
Options() Options
NewMessage(topic string, msg interface{}, opts ...MessageOption) Message // 构建消息
NewRequest(service, endpoint string, req interface{}, reqOpts ...RequestOption)
Request // 构建服务请求
Call(ctx context.Context, req Request, rsp interface{}, opts ...CallOption) error //
调⽤服务端
Stream(ctx context.Context, req Request, opts ...CallOption) (Stream, error) // 构建
双向通信
Publish(ctx context.Context, msg Message, opts ...PublishOption) error // 发布消息
String() string
}

type rpcClient struct {
seq uint64
once atomic.Value
opts Options
pool pool.Pool
}

type Broker interface {
Init(...Option) error
Options() Options
Address() string
Connect() error
Disconnect() error
Publish(topic string, m *Message, opts ...PublishOption) error
Subscribe(topic string, h Handler, opts ...SubscribeOption) (Subscriber, error)
String() string
}

type httpBroker struct {
id string
address string
opts Options
mux *http.ServeMux
c *http.Client
r registry.Registry
sync.RWMutex
subscribers map[string][]*httpSubscriber
running bool
exit chan chan error
// offline message inbox
mtx sync.RWMutex
inbox map[string][][]byte
}

type Registry interface {
Init(...Option) error
Options() Options
Register(*Service, ...RegisterOption) error
Deregister(*Service, ...DeregisterOption) error
GetService(string, ...GetOption) ([]*Service, error)
ListServices(...ListOption) ([]*Service, error)
Watch(...WatchOption) (Watcher, error)
String() string
}

type etcdRegistry struct {
client *clientv3.Client
options registry.Options
sync.RWMutex
register map[string]uint64
leases map[string]clientv3.LeaseID
}

type httpTransport struct {
opts Options
}

type Auth interface {
// Init the auth
Init(opts ...Option)
// Options set for auth
Options() Options
// Generate a new account
Generate(id string, opts ...GenerateOption) (*Account, error)
// Inspect a token
Inspect(token string) (*Account, error)
// Token generated using refresh token or credentials
Token(opts ...TokenOption) (*Token, error)
// String returns the name of the implementation
String() string
}

type jwt struct {
sync.Mutex
options auth.Options
jwt jwtToken.Provider
}

type Cache interface {
// Get gets a cached value by key.
Get(ctx context.Context, key string) (interface{}, time.Time, error)
// Put stores a key-value pair into cache.
Put(ctx context.Context, key string, val interface{}, d time.Duration) error
// Delete removes a key from cache.
Delete(ctx context.Context, key string) error
// String returns the name of the implementation.
String() string
}

type redisCache struct {
opts cache.Options
client redis.UniversalClient
}

type Cmd interface {
// The cli app within this cmd
App() *cli.App
// Adds options, parses flags and initialise
// exits on error
Init(opts ...Option) error
// Options set within this command
Options() Options
}

type cmd struct {
opts Options
app *cli.App
}

type Config interface {
// provide the reader.Values interface
reader.Values
// Init the config
Init(opts ...Option) error
// Options in the config
Options() Options
// Stop the config loader/watcher
Close() error
// Load config sources
Load(source ...source.Source) error
// Force a source changeset sync
Sync() error
// Watch a value for changes
Watch(path ...string) (Watcher, error)
}

type config struct {
exit chan bool
opts Options
sync.RWMutex
// the current snapshot
snap *loader.Snapshot
// the current values
vals reader.Values
}

type Runtime interface {
// Init initializes runtime
Init(...Option) error
// Create registers a service
Create(*Service, ...CreateOption) error
// Read returns the service
Read(...ReadOption) ([]*Service, error)
// Update the service in place以kubernetes实现为例。go-micro的默认实现为runtime.runtime，只适⽤于所有服务在同⼀个操作系统中的场
景。组件的kubernetes实现为kubernetes.kubernetes，适⽤于kubernetes环境。
为了区分，以下通过kubernetes表示runtime的组件， k8s表示kubernetes环境
go-micro中的微服务节点对应kubernetes中的pod，可通过Deployment和Service进⾏控制。故对runtime对微服
务的管理，就是对k8s资源的管理。
kubernetes包含以下信息
kubernetes客户端
kubernetes客户端⽤于请求kubernetes的kube-apiserver。管理k8s资源。
监听微服务事件：runtime组件启动后，会监听从微服务传递过来的事件，事件包含：服务创建，服务更新，服务
删除。
传递到kubernetes：调⽤kube-apiserver，操作Deployment和Service资源。
配置信息
Update(*Service, ...UpdateOption) error
// Remove a service
Delete(*Service, ...DeleteOption) error
// Logs returns the logs for a service
Logs(*Service, ...LogsOption) (LogStream, error)
// Start starts the runtime
Start() error
// Stop shuts down the runtime
Stop() error
// String describes runtime
String() string
}

type kubernetes struct {
sync.RWMutex
// options configure runtime
options *runtime.Options
// indicates if we're running
running bool
// used to stop the runtime
closed chan bool
// client is kubernetes client
client client.Client
// namespaces which exist
namespaces []client.Namespace
}

type Profile interface {
// Start the profiler
Start() error
// Stop the profiler
Stop() error
// Name of the profiler
String() string
}

type profiler struct {
opts profile.Options
sync.Mutex
running bool
// where the cpu profile is written
cpuFile *os.File
// where the mem profile is written
memFile *os.File
}

type Store interface {
// Init initialises the store. It must perform any required setup on the backing
storage implementation and check that it is ready for use, returning any errors.
Init(...Option) error
// Options allows you to view the current options.
Options() Options
// Read takes a single key name and optional ReadOptions. It returns matching
[]*Record or an error.
Read(key string, opts ...ReadOption) ([]*Record, error)
// Write() writes a record to the store, and returns an error if the record was not
written.
Write(r *Record, opts ...WriteOption) error
// Delete removes the record with the corresponding key from the store.
Delete(key string, opts ...DeleteOption) error
// List returns any keys that match, or an empty list with no error if none matched.
List(opts ...ListOption) ([]string, error)
// Close the store
Close() error
// String returns the name of the implementation.
String() string
}

type rkv struct {
ctx context.Context
options store.Options
Client redis.UniversalClient
}

type Logger interface {
// Init initialises options
Init(options ...Option) error
// The Logger options
Options() Options
// Fields set fields to always be logged
Fields(fields map[string]interface{}) Logger
// Log writes a log entry
Log(level Level, v ...interface{})
// Logf writes a formatted log entry
Logf(level Level, format string, v ...interface{})
// String returns the name of logger
String() string
}

type defaultLogger struct {
sync.RWMutex
opts Options
}