github/7days-golang
4
0
Fork 0
mirror of https://github.com/geektutu/7days-golang.git synced 2024-04-21 12:32:11 +00:00
Code Issues Releases Wiki Activity

gee-rpc day6 add discovery

Browse Source
This commit is contained in:
gzdaijie
2020-10-05 02:16:39 +08:00
parent c2cdbf7da1
commit 84fe60f280
10 changed files with 152 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gee-rpc/day1-codec/codec/codec.go
+1 -1
View File
@@ -23,7 +23,7 @@ type Type string
const (
GobType Type = "application/gob"
JsonType Type = "application/json"
JsonType Type = "application/json" // not implemented
)
var NewCodecFuncMap map[Type]NewCodecFunc
gee-rpc/day2-client/codec/codec.go
+1 -1
View File
@@ -23,7 +23,7 @@ type Type string
const (
GobType Type = "application/gob"
JsonType Type = "application/json"
JsonType Type = "application/json" // not implemented
)
var NewCodecFuncMap map[Type]NewCodecFunc
gee-rpc/day3-service/codec/codec.go
+1 -1
View File
@@ -23,7 +23,7 @@ type Type string
const (
GobType Type = "application/gob"
JsonType Type = "application/json"
JsonType Type = "application/json" // not implemented
)
var NewCodecFuncMap map[Type]NewCodecFunc
gee-rpc/day4-timeout/codec/codec.go
+1 -1
View File
@@ -23,7 +23,7 @@ type Type string
const (
GobType Type = "application/gob"
JsonType Type = "application/json"
JsonType Type = "application/json" // not implemented
)
var NewCodecFuncMap map[Type]NewCodecFunc
gee-rpc/day5-http-debug/codec/codec.go
+1 -1
View File
@@ -23,7 +23,7 @@ type Type string
const (
GobType Type = "application/gob"
JsonType Type = "application/json"
JsonType Type = "application/json" // not implemented
)
var NewCodecFuncMap map[Type]NewCodecFunc
gee-rpc/day6-discovery/client.go
+2 -2
View File
@@ -316,9 +316,9 @@ func XDial(rpcAddr string, opts ...*Option) (*Client, error) {
protocol, addr := parts[0], parts[1]
switch protocol {
case "http":
return DialHTTP("tcp", addr)
return DialHTTP("tcp", addr, opts...)
default:
// tcp, unix or other transport protocol
return Dial(protocol, addr)
return Dial(protocol, addr, opts...)
}
}
gee-rpc/day6-discovery/codec/codec.go
+1 -1
View File
@@ -23,7 +23,7 @@ type Type string
const (
GobType Type = "application/gob"
JsonType Type = "application/json"
JsonType Type = "application/json" // not implemented
)
var NewCodecFuncMap map[Type]NewCodecFunc
gee-rpc/day6-discovery/main/main.go
+62 -20
View File
@@ -3,10 +3,11 @@ package main
import (
"context"
"geerpc"
"geerpc/xclient"
"log"
"net"
"net/http"
"sync"
"time"
)
type Foo int
@@ -18,38 +19,79 @@ func (f Foo) Sum(args Args, reply *int) error {
return nil
}
func startServer(addrCh chan string) {
var foo Foo
l, _ := net.Listen("tcp", ":9999")
_ = geerpc.Register(&foo)
geerpc.HandleHTTP()
addrCh <- l.Addr().String()
_ = http.Serve(l, nil)
func (f Foo) Sleep(args Args, reply *int) error {
time.Sleep(time.Second * time.Duration(args.Num1))
*reply = args.Num1 + args.Num2
return nil
}
func call(addrCh chan string) {
client, _ := geerpc.DialHTTP("tcp", <-addrCh)
defer func() { _ = client.Close() }()
func startServer(addrCh chan string) {
var foo Foo
l, _ := net.Listen("tcp", ":0")
server := geerpc.NewServer()
_ = server.Register(&foo)
addrCh <- l.Addr().String()
server.Accept(l)
}
func foo(xc *xclient.XClient, ctx context.Context, typ, serviceMethod string, args *Args) {
var reply int
var err error
switch typ {
case "call":
err = xc.Call(ctx, serviceMethod, args, &reply)
case "broadcast":
err = xc.Broadcast(ctx, serviceMethod, args, &reply)
}
if err != nil {
log.Printf("%s %s error: %v", typ, serviceMethod, err)
} else {
log.Printf("%s Foo.Sum success: %d + %d = %d", typ, args.Num1, args.Num2, reply)
}
}
func call(addr1, addr2 string) {
d := xclient.NewMultiServerDiscovery([]string{"tcp@" + addr1, "tcp@" + addr2})
xc := xclient.NewXClient(d, xclient.RandomSelect, nil)
defer func() { _ = xc.Close() }()
// send request & receive response
var wg sync.WaitGroup
for i := 0; i < 5; i++ {
wg.Add(1)
go func(i int) {
defer wg.Done()
args := &Args{Num1: i, Num2: i * i}
var reply int
if err := client.Call(context.Background(), "Foo.Sum", args, &reply); err != nil {
log.Fatal("call Foo.Sum error:", err)
}
log.Printf("%d + %d = %d", args.Num1, args.Num2, reply)
foo(xc, context.Background(), "call", "Foo.Sum", &Args{Num1: i, Num2: i * i})
}(i)
}
wg.Wait()
}
func broadcast(addr1, addr2 string) {
d := xclient.NewMultiServerDiscovery([]string{"tcp@" + addr1, "tcp@" + addr2})
xc := xclient.NewXClient(d, xclient.RandomSelect, nil)
var wg sync.WaitGroup
for i := 0; i < 5; i++ {
wg.Add(1)
go func(i int) {
defer wg.Done()
foo(xc, context.Background(), "broadcast", "Foo.Sum", &Args{Num1: i, Num2: i * i})
// expect 2 - 5 timeout
ctx, _ := context.WithTimeout(context.Background(), time.Second*2)
foo(xc, ctx, "broadcast", "Foo.Sleep", &Args{Num1: i, Num2: i * i})
}(i)
}
wg.Wait()
}
func main() {
ch := make(chan string)
go call(ch)
startServer(ch)
ch1 := make(chan string)
ch2 := make(chan string)
// start two servers
go startServer(ch1)
go startServer(ch2)
addr1 := <-ch1
addr2 := <-ch2
call(addr1, addr2)
broadcast(addr1, addr2)
}
gee-rpc/day6-discovery/xclient/discovery.go
+11 -1
View File
@@ -10,11 +10,12 @@ type SelectMode int
const (
RandomSelect SelectMode = iota // select randomly
RobbinSelect // select using Robbin algorithm
RobbinSelect // select using Robbin algorithm, not implemented
)
type Discovery interface {
Get(mode SelectMode) string
All() []string
}
var _ Discovery = (*MultiServersDiscovery)(nil)
@@ -48,6 +49,15 @@ func (d *MultiServersDiscovery) Get(mode SelectMode) string {
}
}
func (d *MultiServersDiscovery) All() []string {
d.mu.RLock()
defer d.mu.RUnlock()
// return a copy of d.servers
servers := make([]string, len(d.servers), len(d.servers))
copy(servers, d.servers)
return servers
}
// NewMultiServerDiscovery creates a MultiServersDiscovery instance
func NewMultiServerDiscovery(servers []string) *MultiServersDiscovery {
return &MultiServersDiscovery{
gee-rpc/day6-discovery/xclient/xclient.go
+71 -16
View File
@@ -4,6 +4,7 @@ import (
"context"
. "geerpc"
"io"
"reflect"
"sync"
)
@@ -11,38 +12,92 @@ type XClient struct {
d Discovery
mode SelectMode
opt *Option
clients sync.Map
mu sync.Mutex // protect following
clients map[string]*Client
}
var _ io.Closer = (*XClient)(nil)
func NewXClient(d Discovery, mode SelectMode, opt *Option) *XClient {
return &XClient{d: d, mode: mode, opt: opt}
return &XClient{d: d, mode: mode, opt: opt, clients: make(map[string]*Client)}
}
func (xc *XClient) Close() error {
xc.clients.Range(func(k, v interface{}) bool {
xc.mu.Lock()
defer xc.mu.Unlock()
for key, client := range xc.clients {
// I have no idea how to deal with error, just ignore it.
_ = v.(*Client).Close()
return true
})
xc.clients = sync.Map{}
_ = client.Close()
delete(xc.clients, key)
}
return nil
}
func (xc *XClient) dial(rpcAddr string) (*Client, error) {
xc.mu.Lock()
defer xc.mu.Unlock()
client, ok := xc.clients[rpcAddr]
if ok && !client.IsAvailable() {
_ = client.Close()
delete(xc.clients, rpcAddr)
client = nil
}
if client == nil {
var err error
client, err = XDial(rpcAddr, xc.opt)
if err != nil {
return nil, err
}
xc.clients[rpcAddr] = client
}
return client, nil
}
func (xc *XClient) call(rpcAddr string, ctx context.Context, serviceMethod string, args, reply interface{}) error {
client, err := xc.dial(rpcAddr)
if err != nil {
return err
}
return client.Call(ctx, serviceMethod, args, reply)
}
// Call invokes the named function, waits for it to complete,
// and returns its error status.
// xc will choose a proper server.
func (xc *XClient) Call(ctx context.Context, serviceMethod string, args, reply interface{}) error {
rpcAddr := xc.d.Get(xc.mode)
client, ok := xc.clients.Load(rpcAddr)
if !ok {
var err error
client, err = XDial(rpcAddr, xc.opt)
if err != nil {
return err
}
xc.clients.Store(rpcAddr, client)
return xc.call(rpcAddr, ctx, serviceMethod, args, reply)
}
// Broadcast invokes the named function for every server registered in discovery
func (xc *XClient) Broadcast(ctx context.Context, serviceMethod string, args, reply interface{}) error {
servers := xc.d.All()
var wg sync.WaitGroup
var mu sync.Mutex
var e error
replyDone := reply == nil // if reply is nil, don't need to set value
ctx, cancel := context.WithCancel(ctx)
for _, rpcAddr := range servers {
wg.Add(1)
go func() {
defer wg.Done()
var clonedReply interface{}
if reply != nil {
clonedReply = reflect.New(reflect.ValueOf(reply).Elem().Type()).Interface()
}
err := xc.call(rpcAddr, ctx, serviceMethod, args, clonedReply)
mu.Lock()
if err != nil && e == nil {
e = err
cancel() // if any call failed, cancel unfinished calls
}
if err == nil && !replyDone {
reflect.ValueOf(reply).Elem().Set(reflect.ValueOf(clonedReply).Elem())
replyDone = true
}
mu.Unlock()
}()
}
return client.(*Client).Call(ctx, serviceMethod, args, reply)
wg.Wait()
return e
}