pubsub with concurrent

https://gounconference.connpass.com/event/129090/ の発表資料です。

Transcript

1. Pub/Sub with concurrent Go(Un)ConferenceʢGo͋Μ͜ʣLTେձ 6kg @takashabe
2. whoami • Takashi Abe (@takashabe) • גࣜձࣾαΠόʔΤʔδΣϯτ ΞυςΫελδΦ • ޿ࠂ഑৴ϓϩμΫτΛGoͰ࡞͍ͬͯ·͢ • Goྺ3೥͘Β͍ • ޷͖ͳඪ४ύοέʔδ͸ `go` ύοέʔδͰ͢ • Macͷ`g`Ωʔ͕յΕ͍ͯͯGo͕ଧͪͮΒͯ͘ࠔ͍ͬͯ·͢
3. Cloud Pub/Sub Λ࢖ͬͯಘͨ Subscriber Workerͷ஌ݟΛ঺հ
4. Cloud Pub/Sub is Կ • GCPͷpub/subαʔϏε • Subscriberͷ࣮૷ʹ͍͔ͭ͘ύλʔϯ͕͋Δ • Pull • worker͕pub/sub APIΛcallͯ͠ϝοηʔδΛ ϑΣον͢Δ • Push • worker͕HTTPαʔόͱͯ͠ಈ࡞͢Δ • pub/sub͕ϝοηʔδΛPOSTϦΫΤετ͢Δ
5. Cloud Pub/Sub is Կ • GCPͷpub/subαʔϏε • Subscriberͷ࣮૷ʹ͍͔ͭ͘ύλʔϯ͕͋Δ • Pull • worker͕pub/sub APIΛcallͯ͠ϝοηʔδΛ ϑΣον͢Δ • Push • worker͕HTTPαʔόͱͯ͠ಈ࡞͢Δ • pub/sub͕ϝοηʔδΛPOSTϦΫΤετ͢Δ
6. Cloud Pub/Sub is Կ https://speakerdeck.com/hlts2/sub-worker-framework- implementation?slide=6 ΑΓҾ༻
7. ࣄྫ 1 1ϝοηʔδͷॲཧʹ͕͔͔࣌ؒͬͯ٧·Δ
8. 1ϝοηʔδͷॲཧʹ͕͔͔࣌ؒͬͯ٧·Δ • 1ϝοηʔδ౰ͨΓͷॲཧ͕࣌ؒ௕͍ • ͔ͭಛఆͷSubscriber worker͕େྔͷϝοηʔδΛॲ ཧ͠Α͏ͱͯ͠͠·͏
9. • Receive Settingͷ࠷దԽ • goroutineͰฒྻॲཧ
10. Receive Settingͷ࠷దԽ https://github.com/golang/dep/blob/master/docs/assets/DigbyShadows.png ΑΓҾ༻ • σϑΥϧτ • MaxOutstandingMessages: 1000 • NumGoroutines: 1 →1 worker͕େྔͷϝοηʔδΛอ࣋ͯ͠͠·͏ type ReceiveSettings struct { ... MaxOutstandingMessages int // ϝϞϦ಺ʹஔ͍͓ͯ͘ϝοηʔδ਺ NumGoroutines int // ϝοηʔδΛॲཧ͢Δworker goroutine }
11. goroutineͰฒྻॲཧ • IOό΢ϯυͳॲཧͳͲɺฒྻԽग़དྷΔͱ͜ΖΛͻͨ͢ ΒฒྻԽ
12. ࣄྫ2 ͋ΔଐੑΛχΞϦΞϧλΠϜͰҰճ͚ͩߋ৽ ͍ͨ͠
13. ͋ΔଐੑΛχΞϦΞϧλΠϜͰҰճ͚ͩߋ৽͍ͨ͠ • ޿ࠂ͕ݟΒΕͨ࣌ɺͦͷ޿ࠂʹඥͮ͘ଐੑΛߋ৽ • ݫີʹຖճߋ৽͢Δඞཁ͸ͳ͍ • 1෼͝ͱʹߋ৽͢ΔΑ͏ʹͨ͠ • σʔλετΞͷ࠷ऴߋ৽೔࣌Ͱ੍ޚ • ॲཧதʹෳ਺ϦΫΤετ͕͋Δͱແବͳߋ৽͕૸Δ
14. golang.org/x/sync/singleflight Ͱॏෳߋ৽Λආ͚Δ
15. golang.org/x/sync/singleflight https://play.golang.org/p/h_lCoqJDp9s package main import ( "fmt" "sync" "time" "golang.org/x/sync/singleflight" ) func main() { var ( sg singleflight.Group wg sync.WaitGroup ) for i := 0; i < 10; i++ { wg.Add(1) go func() { // "key" ͝ͱʹಉ࣮࣌ߦΛߜΔ now, _, shared := sg.Do("key", func() (interface{}, error) { time.Sleep(time.Millisecond * 100) return time.Now(), nil }) fmt.Printf("%v: shared: %t\n", now, shared) wg.Done() }() } wg.Wait() }
16. golang.org/x/sync/singleflight 2009-11-10 23:00:00.1 +0000 UTC m=+0.100000001: shared: true 2009-11-10 23:00:00.1 +0000 UTC m=+0.100000001: shared: true 2009-11-10 23:00:00.1 +0000 UTC m=+0.100000001: shared: true 2009-11-10 23:00:00.1 +0000 UTC m=+0.100000001: shared: true 2009-11-10 23:00:00.1 +0000 UTC m=+0.100000001: shared: true 2009-11-10 23:00:00.1 +0000 UTC m=+0.100000001: shared: true 2009-11-10 23:00:00.1 +0000 UTC m=+0.100000001: shared: true 2009-11-10 23:00:00.1 +0000 UTC m=+0.100000001: shared: true 2009-11-10 23:00:00.1 +0000 UTC m=+0.100000001: shared: true 2009-11-10 23:00:00.1 +0000 UTC m=+0.100000001: shared: true time.Now() ͷ݁Ռ͕ἧ͍ͬͯΔ
17. ࣄྫ3 όʔετੑ͕͋ΔSubscriberͷAPIίʔϧͰ όοΫΤϯυ͕ࢮ͵
18. όʔετੑ͕͋ΔSubscriberͷAPIίʔϧͰόοΫΤϯυ͕ࢮ͵ • ϝοηʔδΛड͚ͯόοΫΤϯυͷgRPCαʔόʹϦΫ ΤετΛૹΔαʔϏε • Publishྔ͕֎෦αʔϏεґଘ • ීஈ͸3,000RPSఔ౓ • ͨ·ʹ20,000RPS͘Β͍
19. golang.org/x/time/rate ͰBackpressureΛ࣮૷͢Δ
20. golang.org/x/time/rate https://play.golang.org/p/HsnbmPfUZ_A package main import ( "context" "fmt" "sync" "time" "golang.org/x/time/rate" ) func main() { limit := rate.Every(time.Second / 1) // τʔΫϯͷཷ·Δ଎౓ limiter := rate.NewLimiter(limit, 1) // limitΛݩʹlimiterΛੜ੒ɻୈ2Ҿ਺͸τʔΫϯͷ࠷େஷଂྔ(burst) var wg sync.WaitGroup wg.Add(1) go func() { for i := 0; i < 10; i++ { limiter.Wait(context.Background()) fmt.Println(time.Now()) } wg.Done() }() wg.Wait() }
21. golang.org/x/time/rate 2009-11-10 23:00:00 +0000 UTC m=+0.000000001 2009-11-10 23:00:01 +0000 UTC m=+1.000000001 2009-11-10 23:00:02 +0000 UTC m=+2.000000001 2009-11-10 23:00:03 +0000 UTC m=+3.000000001 2009-11-10 23:00:04 +0000 UTC m=+4.000000001 2009-11-10 23:00:05 +0000 UTC m=+5.000000001 2009-11-10 23:00:06 +0000 UTC m=+6.000000001 2009-11-10 23:00:07 +0000 UTC m=+7.000000001 2009-11-10 23:00:08 +0000 UTC m=+8.000000001 2009-11-10 23:00:09 +0000 UTC m=+9.000000001 େମ1ඵͣͭॲཧ͞Ε͍ͯΔ
22. ·ͱΊ • Cloud Pub/SubΛGoͰ࢖ͬͯૺ۰ͨ͠ඇಉظॲཧܥͷ ࣄྫʹ͍ͭͯ঺հͨ͠ • Pub/Subʹݶͬͨ࿩Ͱ͸ͳ͍ • `golang.org/x/…` ύοέʔδ͸syncΛ͸͡Ίɺศརͳ ΋ͷ͕ଟ͍ͷͰͨ·ʹ८ճ͢Δͱָ͍͠ • golang.org/x/net/netutil ͱ͔΋޷͖