xzrpc

README

xzrpc

A lightweight RPC framework that implements load balancing, service registration, and other functionalities.

Documentation

Index

• Constants
• Variables
• type Call
• type Client
  • func Dial(network, address string, opts ...*Option) (client *Client, err error)
  • func DialHTTP(network, address string, opts ...*Option) (client *Client, err error)
  • func DialRPC(rpcAddr string, opts ...*Option) (*Client, error)
  • func NewClient(conn net.Conn, option *Option) (*Client, error)
  • func NewHTTPClient(conn net.Conn, option *Option) (*Client, error)
  • func (c *Client) Call(ctx context.Context, serviceMethod string, args, reply interface{}) error
  • func (c *Client) CheckAvailable() bool
  • func (c *Client) Close() error
  • func (c *Client) Go(serviceMethod string, args, reply interface{}, done chan *Call) *Call
• type DefaultDiscovery
  • func NewDefaultDiscovery(servers []string) *DefaultDiscovery
  • func (d *DefaultDiscovery) Get(strategy StrategyMode) (string, error)
  • func (d *DefaultDiscovery) GetAll() ([]string, error)
  • func (d *DefaultDiscovery) Refresh() error
  • func (d *DefaultDiscovery) Update(servers []string) error
• type Discovery
• type Option
• type RegistryDiscovery
  • func NewRegistryDiscovery(registryAddr string, timeout time.Duration) *RegistryDiscovery
  • func (rd *RegistryDiscovery) Get(strategy StrategyMode) (string, error)
  • func (rd *RegistryDiscovery) GetAll() ([]string, error)
  • func (rd *RegistryDiscovery) Refresh() error
  • func (rd *RegistryDiscovery) Update(servers []string) error
• type Server
  • func NewServer() *Server
  • func (s *Server) Accept(listener net.Listener)
  • func (s *Server) HandleCodec(cc codec.Codec, rawProtocol protocol.Protocol)
  • func (s *Server) HandleConn(conn io.ReadWriteCloser)
  • func (s *Server) HandleHTTP()
  • func (s *Server) Register(instance interface{}) error
  • func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)
• type StrategyMode
• type XZClient
  • func NewXZClient(d Discovery, strategy StrategyMode, option *Option) *XZClient
  • func (xzc *XZClient) Broadcast(ctx context.Context, serviceMethod string, args, reply interface{}) error
  • func (xzc *XZClient) Call(ctx context.Context, serviceMethod string, args, reply interface{}) error
  • func (xzc *XZClient) Close() error

Constants

const (
	RandomStrategy     = iota // 随机策略
	RoundRobinStrategy        // 轮询

)

const Magic = 0x9f9f

Variables

var DefaultOption = Option{
	Magic:          Magic,
	CodecType:      codec.GobType,
	ConnectTimeOut: 0,
}

var EmptyData = struct{}{}

Types

type Call

type Call struct {
	Seq           uint64
	ServiceMethod string // 服务名.方法名
	Args          interface{}
	Reply         interface{}
	Error         error
	Done          chan *Call
}

看下net/rpc，简直一模一样

type Client

type Client struct {
	// contains filtered or unexported fields
}

Client xzrpc client

func Dial

func Dial(network, address string, opts ...*Option) (client *Client, err error)

Dial 拨号，链接服务端

func DialHTTP

func DialHTTP(network, address string, opts ...*Option) (client *Client, err error)

DialHTTP 先建立HTTP，然后进行RPC通信

func DialRPC

func DialRPC(rpcAddr string, opts ...*Option) (*Client, error)

DialRPC rpcAddr：格式为 protocol@addr，例如：tcp@127.0.0.1:8081、http@127.0.0.1:8081

func NewClient

func NewClient(conn net.Conn, option *Option) (*Client, error)

NewClient 创建Client实例 1. 协议交换，即发生Option信息给服务端 2. 创建goroutine调用receive接收响应

func NewHTTPClient

func NewHTTPClient(conn net.Conn, option *Option) (*Client, error)

NewHTTPClient 使用HTTP创建一个client实例 CONNECT localhost:8081/_xzrpc_ http/1.0 HTTP/1.0 200 Connected to xzrpc 通过HTTP CONNECT请求建立连接之后，后续的通信都交给xzrpc.Client

func (*Client) Call

func (c *Client) Call(ctx context.Context, serviceMethod string, args, reply interface{}) error

Call 同步调用 ctx超时处理

func (*Client) CheckAvailable

func (c *Client) CheckAvailable() bool

func (*Client) Close

func (c *Client) Close() error

Close 关闭connection

func (*Client) Go

func (c *Client) Go(serviceMethod string, args, reply interface{}, done chan *Call) *Call

Go 异步调用 返回一个call对象，这个call对象表示一次调用

type DefaultDiscovery

type DefaultDiscovery struct {
	// contains filtered or unexported fields
}

DefaultDiscovery 一个Discovery示例

func NewDefaultDiscovery

func NewDefaultDiscovery(servers []string) *DefaultDiscovery

func (*DefaultDiscovery) Get

func (d *DefaultDiscovery) Get(strategy StrategyMode) (string, error)

Get 根据策略，选择对应的服务 保证线程安全，感觉不需要整个方法加锁

func (*DefaultDiscovery) GetAll

func (d *DefaultDiscovery) GetAll() ([]string, error)

GetAll 返回所有的服务

func (*DefaultDiscovery) Refresh

func (d *DefaultDiscovery) Refresh() error

func (*DefaultDiscovery) Update

func (d *DefaultDiscovery) Update(servers []string) error

type Discovery

type Discovery interface {
	Refresh() error                            // 从注册中心更新服务列表
	Update(servers []string) error             // 手动更新服务列表
	Get(strategy StrategyMode) (string, error) // 根据负载均衡策略，选择一个服务实例
	GetAll() ([]string, error)                 // 返回所有的服务实例
}

type Option

type Option struct {
	Magic     int
	CodecType codec.Type
	// 链接超时，@xz暂时先放在这里，后面优化，拿走
	ConnectTimeOut time.Duration
	// 处理超时
	HandleTimeOut time.Duration
}

type RegistryDiscovery

type RegistryDiscovery struct {
	*DefaultDiscovery
	// contains filtered or unexported fields
}

func NewRegistryDiscovery

func NewRegistryDiscovery(registryAddr string, timeout time.Duration) *RegistryDiscovery

func (*RegistryDiscovery) Get

func (rd *RegistryDiscovery) Get(strategy StrategyMode) (string, error)

func (*RegistryDiscovery) GetAll

func (rd *RegistryDiscovery) GetAll() ([]string, error)

func (*RegistryDiscovery) Refresh

func (rd *RegistryDiscovery) Refresh() error

func (*RegistryDiscovery) Update

func (rd *RegistryDiscovery) Update(servers []string) error

TODO 是不是应该也支持一个？好像没必要！

type Server

type Server struct {
	// contains filtered or unexported fields
}

func NewServer

func NewServer() *Server

func (*Server) Accept

func (s *Server) Accept(listener net.Listener)

TODO-这里可以优化的，看下xzlink

func (*Server) HandleCodec

func (s *Server) HandleCodec(cc codec.Codec, rawProtocol protocol.Protocol)

根据编解码器处理数据 并发处理请求，有序返回数据

func (*Server) HandleConn

func (s *Server) HandleConn(conn io.ReadWriteCloser)

首先使用 json.NewDecoder 反序列化得到 Option 实例，检查 MagicNumber 和 CodeType 的值是否正确。 然后根据 CodeType 得到对应的消息编解码器，接下来的处理交给 serverCodec。

func (*Server) HandleHTTP

func (s *Server) HandleHTTP()

func (*Server) Register

func (s *Server) Register(instance interface{}) error

func (*Server) ServeHTTP

func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)

实现Handler接口 http.Handle("xx", handler) 这里的HTTP请求只允许处理CONNECT的请求

type StrategyMode

type StrategyMode int

type XZClient

type XZClient struct {
	// contains filtered or unexported fields
}

XZClient 支持负载均衡的client 1. 需要服务发现实例；2. 负载均衡策略；3. 协议选择

func NewXZClient

func NewXZClient(d Discovery, strategy StrategyMode, option *Option) *XZClient

func (*XZClient) Broadcast

func (xzc *XZClient) Broadcast(ctx context.Context, serviceMethod string, args, reply interface{}) error

TODO 将请求广播到所有的服务实例

func (*XZClient) Call

func (xzc *XZClient) Call(ctx context.Context, serviceMethod string, args, reply interface{}) error

Call @xz 这块处理相当优秀

func (*XZClient) Close

func (xzc *XZClient) Close() error