rpcx

package module
v0.0.0-...-12bc8e4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 27, 2016 License: Apache-2.0 Imports: 7 Imported by: 0

README

rpcx

rpcx is a distributed RPC framework like Alibaba Dubbo and Weibo Motan. It is developed based on Go net/rpc and provides extra governance features.

When we talk about RPC frameworks, Dubbo is first framework we should introduced, and there is also Dubbox mantained by dangdang. Dubbo has been widely used in e-commerce companies in China, for example, Alibaba, Jingdong and Dangdang.

Though Dubbo has still used Spring 2.5.6.SEC03 and seems has not been supported by Alibaba no longer, some other companies still use it and maintained their branches.

DUBBO is a distributed service framework , provides high performance and transparent RPC remote service call. It is the core framework of Alibaba SOA service governance programs. There are 3,000,000,000 calls for 2,000+ services per day, and it has been widely used in various member sites of Alibaba Group.

Motan is open source now by Weibo. As Zhang Lei said, he is current main developer of Motan:

Motan started in 2013. There are 100 billion calls for hundreds of service callsevery day.

Those two RPC frameworks are developed by Java. There are other famous RPC frameworks such as thrift、finagle。

Goal of rpcx is implemented a RPC framework like Dubbo in Go ecosphere. It is developed by Go, and for Go.

It is a distributed、plugable RPC framework with governance (service discovery、load balancer、fault tolerance、monitor, etc.).

As you know, there are some RPC frameworks, for example, net/rpc、grpc-go、gorilla-rpc, Then why re-invent a wheel?

Although those Go RPC frameworks work well, but their function is relatively simple and only implement end-to end communications. Some product features of service management functions are lack, such as service discovery, Load balancing, fault tolerance.

So I created rpcx and expect it could become a RPC framework like Dubbo.

The similar project is go-micro.

What's RPC

From wikiPedia:

In distributed computing, a remote procedure call (RPC) is when a computer program causes a procedure (subroutine) to execute in another address space (commonly on another computer on a shared network), which is coded as if it were a normal (local) procedure call, without the programmer explicitly coding the details for the remote interaction. That is, the programmer writes essentially the same code whether the subroutine is local to the executing program, or remote.[1] This is a form of client–server interaction (caller is client, executer is server), typically implemented via a request–response message-passing system. The object-oriented programming analog is remote method invocation (RMI). The RPC model implies a level of location transparency, namely that calling procedures is largely the same whether it is local or remote, but usually they are not identical, so local calls can be distinguished from remote calls. Remote calls are usually orders of magnitude slower and less reliable than local calls, so distinguishing them is useful.

RPCs are a form of inter-process communication (IPC), in that different processes have different address spaces: if on the same host machine, they have distinct virtual address spaces, even though the physical address space is the same; while if they are on different hosts, the physical address space is different. Many different (often incompatible) technologies have been used to implement the concept.

Sequence of events during an RPC

  1. The client calls the client stub. The call is a local procedure call, with parameters pushed on to the stack in the normal way.
  2. The client stub packs the parameters into a message and makes a system call to send the message. Packing the parameters is called marshalling.
  3. The client's local operating system sends the message from the client machine to the server machine.
  4. The local operating system on the server machine passes the incoming packets to the server stub.
  5. The server stub unpacks the parameters from the message. Unpacking the parameters is called unmarshalling.
  6. Finally, the server stub calls the server procedure. The reply traces the same steps in the reverse direction.

There are two ways to implement RPC frameworks. One focusses on cross-language calls and the other focusses on service governance.

Dubbo、DubboX、Motan are RPC framework of service governance . Thrift、gRPC、Hessian、Hprose are RPC framework of cross-language calls.

rpcx is a RPC framework of service governance.

Features

more features

  • bases on net/rpc. a Go net/prc project can be converted rpcx project whit few changes.
  • Plugable. Features are implemented by Plugins such as service discovery.
  • Commnuicates with TCP long connections.
  • support many codec. for example, Gob、Json、MessagePack、gencode、ProtoBuf等
  • Service dicovery. support ZooKeeper、Etcd.
  • Fault tolerance：Failover、Failfast、Failtry
  • Load banlancer：support randomSelecter, RoundRobin, consistent hash etc.
  • scalable
  • Other: metrics、log

Architecture

rpcx contains three roles : RPC Server，RPC Client and Registry.

  • Server registers services on Registry
  • Client queries service list and select a server from server list returned from Registry.
  • When a Server is down, Registry can remove this server and then client can remove it too.

So far rpcx support zookeeper, etcd as Registry，Consul support is developing。

Benchmark

[root@localhost rpcx]# go test -bench . -test.benchmem
PASS
BenchmarkNetRPC_gob-16            100000             18742 ns/op             321 B/op          9 allocs/op
BenchmarkNetRPC_jsonrpc-16        100000             21360 ns/op            1170 B/op         31 allocs/op
BenchmarkNetRPC_msgp-16           100000             18617 ns/op             776 B/op         35 allocs/op
BenchmarkRPCX_gob-16              100000             18718 ns/op             320 B/op          9 allocs/op
BenchmarkRPCX_json-16             100000             21238 ns/op            1170 B/op         31 allocs/op
BenchmarkRPCX_msgp-16             100000             18635 ns/op             776 B/op         35 allocs/op
BenchmarkRPCX_gencodec-16         100000             18454 ns/op            4485 B/op         17 allocs/op
BenchmarkRPCX_protobuf-16         100000             17234 ns/op             733 B/op         13 allocs/op

Documentation

Index

Constants

View Source 
const (
	//DefaultRPCPath is the defaut HTTP RPC PATH
	DefaultRPCPath = "/_goRPC_"
)

Variables

View Source 
var (
	// ErrPluginAlreadyExists returns an error with message: 'Cannot activate the same plugin again, plugin '+plugin name[+plugin description]' is already exists'
	ErrPluginAlreadyExists = NewRPCError("Cannot use the same plugin again, '%s[%s]' is already exists")
	// ErrPluginActivate returns an error with message: 'While trying to activate plugin '+plugin name'. Trace: +specific error'
	ErrPluginActivate = NewRPCError("While trying to activate plugin '%s'. Trace: %s")
	// ErrPluginRemoveNoPlugins returns an error with message: 'No plugins are registed yet, you cannot remove a plugin from an empty list!'
	ErrPluginRemoveNoPlugins = NewRPCError("No plugins are registed yet, you cannot remove a plugin from an empty list!")
	// ErrPluginRemoveEmptyName returns an error with message: 'Plugin with an empty name cannot be removed'
	ErrPluginRemoveEmptyName = NewRPCError("Plugin with an empty name cannot be removed")
	// ErrPluginRemoveNotFound returns an error with message: 'Cannot remove a plugin which doesn't exists'
	ErrPluginRemoveNotFound = NewRPCError("Cannot remove a plugin which doesn't exists")
)

Functions

func Close 

func Close() error

Close closes RPC server.

func GetListenedAddress 

func GetListenedAddress() string

GetListenedAddress return the listening address.

func NewDirectRPCClient 

func NewDirectRPCClient(clientCodecFunc ClientCodecFunc, network, address string, timeout time.Duration) (*rpc.Client, error)

NewDirectRPCClient creates a rpc client

func RegisterName 

func RegisterName(name string, service interface{})

RegisterName publishes in the server the set of methods .

func Serve 

func Serve(n, address string)

Serve starts and listens RCP requests. It is blocked until receiving connectings from clients.

func ServeListener 

func ServeListener(ln net.Listener)

ServeListener serve with a listener

func SetServerCodecFunc 

func SetServerCodecFunc(fn ServerCodecFunc)

SetServerCodecFunc sets a ServerCodecFunc

func Start 

func Start(n, address string)

Start starts and listens RCP requests without blocking.

Types

type Client 

type Client struct {
	ClientSelector  ClientSelector
	ClientCodecFunc ClientCodecFunc
	PluginContainer IClientPluginContainer
	FailMode        FailMode
	Retries         int
	// contains filtered or unexported fields
}

Client represents a RPC client.

func NewClient 

func NewClient(s ClientSelector) *Client

NewClient create a client.

func (*Client) Call 

func (c *Client) Call(serviceMethod string, args interface{}, reply interface{}) (err error)

Call invokes the named function, waits for it to complete, and returns its error status.

func (*Client) Close 

func (c *Client) Close() error

Close closes the connection

func (*Client) Go 

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

Go invokes the function asynchronously. It returns the Call structure representing the invocation. The done channel will signal when the call is complete by returning the same Call object. If done is nil, Go will allocate a new channel. If non-nil, done must be buffered or Go will deliberately crash.

type ClientCodecFunc 

type ClientCodecFunc func(conn io.ReadWriteCloser) rpc.ClientCodec

ClientCodecFunc is used to create a rpc.ClientCodecFunc from net.Conn.

type ClientPluginContainer 

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

ClientPluginContainer implements IPluginContainer interface.

func (*ClientPluginContainer) Add 

func (p *ClientPluginContainer) Add(plugin IPlugin) error

Add adds a plugin.

func (*ClientPluginContainer) DoPostReadResponseBody 

func (p *ClientPluginContainer) DoPostReadResponseBody(body interface{})

DoPostReadResponseBody invokes DoPostReadResponseBody plugin.

func (*ClientPluginContainer) DoPostReadResponseHeader 

func (p *ClientPluginContainer) DoPostReadResponseHeader(r *rpc.Response)

DoPostReadResponseHeader invokes DoPostReadResponseHeader plugin.

func (*ClientPluginContainer) DoPostWriteRequest 

func (p *ClientPluginContainer) DoPostWriteRequest(r *rpc.Request, body interface{})

DoPostWriteRequest invokes DoPostWriteRequest plugin.

func (*ClientPluginContainer) DoPreReadResponseBody 

func (p *ClientPluginContainer) DoPreReadResponseBody(body interface{})

DoPreReadResponseBody invokes DoPreReadResponseBody plugin.

func (*ClientPluginContainer) DoPreReadResponseHeader 

func (p *ClientPluginContainer) DoPreReadResponseHeader(r *rpc.Response)

DoPreReadResponseHeader invokes DoPreReadResponseHeader plugin.

func (*ClientPluginContainer) DoPreWriteRequest 

func (p *ClientPluginContainer) DoPreWriteRequest(r *rpc.Request, body interface{})

DoPreWriteRequest invokes DoPreWriteRequest plugin.

func (*ClientPluginContainer) GetAll 

func (p *ClientPluginContainer) GetAll() []IPlugin

GetAll returns all activated plugins

func (*ClientPluginContainer) GetByName 

func (p *ClientPluginContainer) GetByName(pluginName string) IPlugin

GetByName returns a plugin instance by it's name

func (*ClientPluginContainer) GetDescription 

func (p *ClientPluginContainer) GetDescription(plugin IPlugin) string

GetDescription returns the name of a plugin, if no GetDescription() implemented it returns an empty string ""

func (*ClientPluginContainer) GetName 

func (p *ClientPluginContainer) GetName(plugin IPlugin) string

GetName returns the name of a plugin, if no GetName() implemented it returns an empty string ""

func (*ClientPluginContainer) Remove 

func (p *ClientPluginContainer) Remove(pluginName string) error

Remove removes a plugin by it's name.

type ClientSelector 

type ClientSelector interface {
	Select(clientCodecFunc ClientCodecFunc, options ...interface{}) (*rpc.Client, error)
}

ClientSelector defines an interface to create a rpc.Client from cluster or standalone.

type DirectClientSelector 

type DirectClientSelector struct {
	Network, Address string
	Timeout          time.Duration
}

DirectClientSelector is used to a direct rpc server. It don't select a node from service cluster but a specific rpc server.

func (*DirectClientSelector) Select 

func (s *DirectClientSelector) Select(clientCodecFunc ClientCodecFunc, options ...interface{}) (*rpc.Client, error)

Select returns a rpc client

type FailMode 

type FailMode int

FailMode is a feature to decide client actions when clients fail to invoke services 

const (
	//Failover selects another server automaticaly
	Failover FailMode = iota
	//Failfast returns error immediately
	Failfast
	//Failtry use current client again
	Failtry
)

type IClientPluginContainer 

type IClientPluginContainer interface {
	Add(plugin IPlugin) error
	Remove(pluginName string) error
	GetName(plugin IPlugin) string
	GetDescription(plugin IPlugin) string
	GetByName(pluginName string) IPlugin
	GetAll() []IPlugin

	DoPreReadResponseHeader(*rpc.Response)
	DoPostReadResponseHeader(*rpc.Response)
	DoPreReadResponseBody(interface{})
	DoPostReadResponseBody(interface{})

	DoPreWriteRequest(*rpc.Request, interface{})
	DoPostWriteRequest(*rpc.Request, interface{})
}

IClientPluginContainer represents a plugin container that defines all methods to manage plugins. And it also defines all extension points.

type IPlugin 

type IPlugin interface {
	Name() string
	Description() string
}

IPlugin represents a plugin.

type IPostConnAcceptPlugin 

type IPostConnAcceptPlugin interface {
	HandleConnAccept(net.Conn) bool
}

IPostConnAcceptPlugin represents connection accept plugin. if returns false, it means subsequent IPostConnAcceptPlugins should not contiune to handle this conn and this conn has been closed.

type IPostReadRequestBodyPlugin 

type IPostReadRequestBodyPlugin interface {
	PostReadRequestBody(body interface{})
}

IPostReadRequestBodyPlugin represents .

type IPostReadRequestHeaderPlugin 

type IPostReadRequestHeaderPlugin interface {
	PostReadRequestHeader(r *rpc.Request)
}

IPostReadRequestHeaderPlugin represents .

type IPostReadResponseBodyPlugin 

type IPostReadResponseBodyPlugin interface {
	PostReadResponseBody(interface{})
}

IPostReadResponseBodyPlugin represents .

type IPostReadResponseHeaderPlugin 

type IPostReadResponseHeaderPlugin interface {
	PostReadResponseHeader(*rpc.Response)
}

IPostReadResponseHeaderPlugin represents .

type IPostWriteRequestPlugin 

type IPostWriteRequestPlugin interface {
	PostWriteRequest(*rpc.Request, interface{})
}

IPostWriteRequestPlugin represents .

type IPostWriteResponsePlugin 

type IPostWriteResponsePlugin interface {
	PostWriteResponse(*rpc.Response, interface{})
}

IPostWriteResponsePlugin represents .

type IPreReadRequestBodyPlugin 

type IPreReadRequestBodyPlugin interface {
	PreReadRequestBody(body interface{})
}

IPreReadRequestBodyPlugin represents .

type IPreReadRequestHeaderPlugin 

type IPreReadRequestHeaderPlugin interface {
	PreReadRequestHeader(r *rpc.Request)
}

IPreReadRequestHeaderPlugin represents .

type IPreReadResponseBodyPlugin 

type IPreReadResponseBodyPlugin interface {
	PreReadResponseBody(interface{})
}

IPreReadResponseBodyPlugin represents .

type IPreReadResponseHeaderPlugin 

type IPreReadResponseHeaderPlugin interface {
	PreReadResponseHeader(*rpc.Response)
}

IPreReadResponseHeaderPlugin represents .

type IPreWriteRequestPlugin 

type IPreWriteRequestPlugin interface {
	PreWriteRequest(*rpc.Request, interface{})
}

IPreWriteRequestPlugin represents .

type IPreWriteResponsePlugin 

type IPreWriteResponsePlugin interface {
	PreWriteResponse(*rpc.Response, interface{})
}

IPreWriteResponsePlugin represents .

type IRegisterPlugin 

type IRegisterPlugin interface {
	Register(name string, rcvr interface{}) error
}

IRegisterPlugin represents register plugin.

type IServerCodecPlugin 

type IServerCodecPlugin interface {
	IPreReadRequestHeaderPlugin
	IPostReadRequestHeaderPlugin
	IPreReadRequestBodyPlugin
	IPostReadRequestBodyPlugin
	IPreWriteResponsePlugin
	IPostWriteResponsePlugin
}

IServerCodecPlugin represents .

type IServerPluginContainer 

type IServerPluginContainer interface {
	Add(plugin IPlugin) error
	Remove(pluginName string) error
	GetName(plugin IPlugin) string
	GetDescription(plugin IPlugin) string
	GetByName(pluginName string) IPlugin
	GetAll() []IPlugin

	DoRegister(name string, rcvr interface{}) error

	DoPostConnAccept(net.Conn) bool

	DoPreReadRequestHeader(r *rpc.Request)
	DoPostReadRequestHeader(r *rpc.Request)

	DoPreReadRequestBody(body interface{})
	DoPostReadRequestBody(body interface{})

	DoPreWriteResponse(*rpc.Response, interface{})
	DoPostWriteResponse(*rpc.Response, interface{})
}

IServerPluginContainer represents a plugin container that defines all methods to manage plugins. And it also defines all extension points.

func GetPluginContainer 

func GetPluginContainer() IServerPluginContainer

GetPluginContainer get PluginContainer of default server.

type MultiError 

type MultiError struct {
	Errors []error
}

MultiError holds multiple errors

func NewMultiError 

func NewMultiError(errors []error) *MultiError

NewMultiError creates and returns an Error with error splice

func (*MultiError) Error 

func (e *MultiError) Error() string

Error returns the message of the actual error

type RPCError 

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

RPCError holds the error

func NewRPCError 

func NewRPCError(errMsg string) *RPCError

NewRPCError creates and returns an Error with a message

func (*RPCError) Error 

func (e *RPCError) Error() string

Error returns the message of the actual error

func (*RPCError) Format 

func (e *RPCError) Format(args ...interface{}) error

Format returns a formatted new error based on the arguments

func (*RPCError) Panic 

func (e *RPCError) Panic()

Panic output the message and after panics

func (*RPCError) Panicf 

func (e *RPCError) Panicf(args ...interface{})

Panicf output the formatted message and after panics

func (*RPCError) Return 

func (e *RPCError) Return() error

Return returns the actual error as it is

func (*RPCError) With 

func (e *RPCError) With(err error) error

With does the same thing as Format but it receives an error type which if it's nil it returns a nil error

type SelectMode 

type SelectMode int
const (
	RandomSelect SelectMode = iota
	RoundRobin
	LeastActive
	ConsistentHash
)

func (SelectMode) String 

func (s SelectMode) String() string

type Server 

type Server struct {
	ServerCodecFunc ServerCodecFunc
	//PluginContainer must be configured before starting and Register plugins must be configured before invoking RegisterName method
	PluginContainer IServerPluginContainer
	// contains filtered or unexported fields
}

Server represents a RPC Server.

func NewServer 

func NewServer() *Server

NewServer returns a new Server.

func (*Server) Address 

func (s *Server) Address() string

Address return the listening address.

func (*Server) Close 

func (s *Server) Close() error

Close closes RPC server.

func (*Server) RegisterName 

func (s *Server) RegisterName(name string, service interface{})

RegisterName publishes in the server the set of methods of the receiver value that satisfy the following conditions:

  • exported method of exported type
  • two arguments, both of exported type
  • the second argument is a pointer
  • one return value, of type error

It returns an error if the receiver is not an exported type or has no suitable methods. It also logs the error using package log. The client accesses each method using a string of the form "Type.Method", where Type is the receiver's concrete type.

func (*Server) Serve 

func (s *Server) Serve(network, address string)

Serve starts and listens RCP requests. It is blocked until receiving connectings from clients.

func (*Server) ServeListener 

func (s *Server) ServeListener(ln net.Listener)

ServeListener starts

func (*Server) Start 

func (s *Server) Start(network, address string)

Start starts and listens RCP requests without blocking.

type ServerCodecFunc 

type ServerCodecFunc func(conn io.ReadWriteCloser) rpc.ServerCodec

ServerCodecFunc is used to create a rpc.ServerCodec from net.Conn.

type ServerPluginContainer 

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

ServerPluginContainer implements IPluginContainer interface.

func (*ServerPluginContainer) Add 

func (p *ServerPluginContainer) Add(plugin IPlugin) error

Add adds a plugin.

func (*ServerPluginContainer) DoPostConnAccept 

func (p *ServerPluginContainer) DoPostConnAccept(conn net.Conn) bool

DoPostConnAccept handle accepted conn

func (*ServerPluginContainer) DoPostReadRequestBody 

func (p *ServerPluginContainer) DoPostReadRequestBody(body interface{})

DoPostReadRequestBody invokes DoPostReadRequestBody plugin.

func (*ServerPluginContainer) DoPostReadRequestHeader 

func (p *ServerPluginContainer) DoPostReadRequestHeader(r *rpc.Request)

DoPostReadRequestHeader invokes DoPostReadRequestHeader plugin.

func (*ServerPluginContainer) DoPostWriteResponse 

func (p *ServerPluginContainer) DoPostWriteResponse(resp *rpc.Response, body interface{})

DoPostWriteResponse invokes DoPostWriteResponse plugin.

func (*ServerPluginContainer) DoPreReadRequestBody 

func (p *ServerPluginContainer) DoPreReadRequestBody(body interface{})

DoPreReadRequestBody invokes DoPreReadRequestBody plugin.

func (*ServerPluginContainer) DoPreReadRequestHeader 

func (p *ServerPluginContainer) DoPreReadRequestHeader(r *rpc.Request)

DoPreReadRequestHeader invokes DoPreReadRequestHeader plugin.

func (*ServerPluginContainer) DoPreWriteResponse 

func (p *ServerPluginContainer) DoPreWriteResponse(resp *rpc.Response, body interface{})

DoPreWriteResponse invokes DoPreWriteResponse plugin.

func (*ServerPluginContainer) DoRegister 

func (p *ServerPluginContainer) DoRegister(name string, rcvr interface{}) error

DoRegister invokes DoRegister plugin.

func (*ServerPluginContainer) GetAll 

func (p *ServerPluginContainer) GetAll() []IPlugin

GetAll returns all activated plugins

func (*ServerPluginContainer) GetByName 

func (p *ServerPluginContainer) GetByName(pluginName string) IPlugin

GetByName returns a plugin instance by it's name

func (*ServerPluginContainer) GetDescription 

func (p *ServerPluginContainer) GetDescription(plugin IPlugin) string

GetDescription returns the name of a plugin, if no GetDescription() implemented it returns an empty string ""

func (*ServerPluginContainer) GetName 

func (p *ServerPluginContainer) GetName(plugin IPlugin) string

GetName returns the name of a plugin, if no GetName() implemented it returns an empty string ""

func (*ServerPluginContainer) Remove 

func (p *ServerPluginContainer) Remove(pluginName string) error

Remove removes a plugin by it's name.

Source Files

View all Source files

Directories

Path Synopsis
_examples
end2end
multi_server

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.