Exploration and practice of growingio responsive programming

{"type":"doc","content":[{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" author ： Lin Shengsheng ,GrowingIO Operational product line R & D Manager , Mainly responsible for GrowingIO R & D management of intelligent operation product line ."}]},{"type":"heading","attrs":{"align":null,"level":1},"content":[{"type":"text","text":" background "}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"GrowingIO It's a company that provides a platform for growth . stay 2018 At the beginning of the year, we launched an intelligent operation platform based on the underlying data capabilities , Combined with accurate user clustering , Data collection and a variety of operations , Help enterprise customers drive user operation with data , Test hypotheses at any time , Boost product growth . The product has the following characteristics ："}]},{"type":"bulletedlist","content":[{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" Support multiple channels to reach users ： Station ： Popup 、 Resource bit , Outside the station ：Push、 SMS 、Webhook."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" Multi platform support , Pop up support ：App、Web、H5 And small program ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" It is easy to establish a closed loop for data operation ."}]}]}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" The following figure shows the business flow chart of the external touch service of the operation platform . Users can initiate an off-site operation activity at any time , It's usually a contact outside the station （ push 、 SMS 、Webhook）. Background system needs to query the underlying data platform interface , Get the crowd information for the event , At the same time, the activity data is assembled and the task is delivered to the outside world ."}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/27/27cae57ec4ca5a91dc92205951b69ac8.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" In this business scenario , The following problems need to be solved ："}]},{"type":"numberedlist","attrs":{"start":null,"normalizeStart":1},"content":[{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":1,"align":null,"origin":null},"content":[{"type":"text","text":" The external input of the system is sudden , You can't predict the magnitude in advance , The system needs to be responsive to changing loads ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":2,"align":null,"origin":null},"content":[{"type":"text","text":" Rely on the underlying data services , If the external system doesn't work , In order to ensure resilience, there needs to be a fusing and recovery mechanism ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":3,"align":null,"origin":null},"content":[{"type":"text","text":" The business process is long , In order to ensure timely response, the task needs to be processed asynchronously ."}]}]}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" Sum up , To maximize the use of server resources 、 Improve service stability and optimize the end user experience ,GrowingIO The server team has done some practice in asynchronous and reactive programming . This article will introduce the exploration and thinking in the process of optimization , I hope it can help readers ."}]},{"type":"heading","attrs":{"align":null,"level":1},"content":[{"type":"text","text":" Asynchronous and responsive "}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" Traditional server-side programs generally adopt synchronous blocking model , By allocating more threads to support more requests , It's in line with ordinary people's mode of thinking , But in the case of sudden traffic , The synchronization model can cause the thread pool to run out , The service mode based on one request and one thread can't achieve dynamic scaling ."}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/3c/3c587c075ca798b4a5133d9636248070.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" Asynchronous programming is based on a shared thread pool , All operations are callbacks . If you encounter time-consuming operations , Threads don't block waiting for an operation to complete , Instead, it will be released back to the thread pool to accept new requests . Wait until the time-consuming operation is complete （ It's usually IO operation ）, Through the message mechanism, re apply to the thread pool for the request code before thread recovery ."}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/b6/b65f748e1ae39ae02f6a018b59324571.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"  We can write a simple program, a simple experiment , Implement the same logic ：1. Inquire about db 2. Query external systems 3. Assembly information returned . The only difference is the implementation of synchronous calls , The other is to implement in a completely asynchronous way ."}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" Local use of the same jmeter Parameters simulate concurrent testing , The results are as follows , The meaning of each column from left to right is ： Name of the request 、 Number of requests 、 Number of failed requests 、 Error rate ( The number of requests with errors in this test / Total number of requests )、 Mean response time 、 Minimum response time 、 Maximum response time 、90% User response time 、95% User response time 、99% User response time 、 throughput ."}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" The overall test results are as follows ："}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/04/04a4fa4abb53228258e3b9a7cd60614a.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/57/578322ef9871b221c27aa582ab1ac243.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":"center","origin":null},"content":[{"type":"text","marks":[{"type":"size","attrs":{"size":10}}],"text":" Synchronization code test results "}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/49/49e9f5b8076db88f9e03a4a73f61f1ab.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":"center","origin":null},"content":[{"type":"text","marks":[{"type":"size","attrs":{"size":10}}],"text":" Asynchronous code test results "}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" The synchronization code is complete in total 260 Requests , The average response time is about 5 second , Because the blocking program runs out of thread pool, the program has a denial of service condition , Produced 13% Error rate ."}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" The overall throughput of asynchronous code has been significantly improved , Finished in the same time 3000 Requests . Error rate for 0 , And there was no denial of service overall ."}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" We can see that asynchronous systems based on message driven mechanism can greatly improve resource utilization , Improve system throughput . The responsive system adds three requirements to the message driven system ： Timely response 、 Resilience and scalability ."}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" In short, a system with the following four characteristics can be called a responsive system ："}]},{"type":"numberedlist","attrs":{"start":null,"normalizeStart":1},"content":[{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":1,"align":null,"origin":null},"content":[{"type":"text","text":" Immediate responsiveness , This is the core goal of responsive systems . A responsive system is a system that can quickly give feedback to the customer's operation no matter what the situation is , Include events 、 User request 、 Failure scenarios , The ultimate goal is to ensure a good customer experience ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":2,"align":null,"origin":null},"content":[{"type":"text","text":" Resilience , Refers to the ability of a system to recover from a failure disaster . There are mainly two parts , One is that the system needs to consider failures , Second, the system should be able to recover from failure ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":3,"align":null,"origin":null},"content":[{"type":"text","text":" Extensibility （ elastic ）, It means that the system remains responsive to changing workloads . Scalability can be divided into single vertical expansion and horizontal linear expansion . This mainly refers to that the system can be partitioned 、 Horizontal expansion by copying and so on , So as to avoid the obvious performance bottleneck of the system ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":4,"align":null,"origin":null},"content":[{"type":"text","text":" Message driven , This is the basis of responsive systems . From the above analysis of the advantages and principles of asynchronous systems, we can see that , Message driven programs maximize the use of machine resources , At the same time, loosely coupled design creates an environment in which business logic can be kept clean , Explicit isolation failure is conducive to the automatic recovery of the system ."}]}]}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/91/910e210c8013f2b66a498fd2b09d4af5.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" The corresponding responsive programming is a programming idea , stay java 8 The specification of responsive flow is introduced for the first time in , namely Reactive Streams Interface .Reactive Streams Very similar to JPA or JDBC, All are API standard , In actual use, the corresponding concrete implementation is needed .JDK Provided Reactive Streams Interface ："}]},{"type":"bulletedlist","content":[{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"link","attrs":{"href":"https://github.com/reactive-streams/reactive-streams-jvm/blob/master/api/src/main/java/org/reactivestreams/Publisher.java","title":"https://github.com/reactive-streams/reactive-streams-jvm/blob/master/api/src/main/java/org/reactivestreams/Publisher.java"},"content":[{"type":"text","text":"org.reactivestreams.Publisher"}]},{"type":"text","text":": Represents a potential unbounded data source , according to Subscriber We need to publish new data ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"link","attrs":{"href":"https://github.com/reactive-streams/reactive-streams-jvm/blob/master/api/src/main/java/org/reactivestreams/Subscriber.java","title":"https://github.com/reactive-streams/reactive-streams-jvm/blob/master/api/src/main/java/org/reactivestreams/Subscriber.java"},"content":[{"type":"text","text":"org.reactivestreams.Subscriber"}]},{"type":"text","text":": Data source consumers , adopt Subscription Request data from the data source ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"link","attrs":{"href":"https://github.com/reactive-streams/reactive-streams-jvm/blob/master/api/src/main/java/org/reactivestreams/Subscription.java","title":"https://github.com/reactive-streams/reactive-streams-jvm/blob/master/api/src/main/java/org/reactivestreams/Subscription.java"},"content":[{"type":"text","text":"org.reactivestreams.Subscription"}]},{"type":"text","text":": Represents a data consumption request ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"link","attrs":{"href":"https://github.com/reactive-streams/reactive-streams-jvm/blob/master/api/src/main/java/org/reactivestreams/Processor.java","title":"https://github.com/reactive-streams/reactive-streams-jvm/blob/master/api/src/main/java/org/reactivestreams/Processor.java"},"content":[{"type":"text","text":"org.reactivestreams.Processor"}]},{"type":"text","text":": processor , Represents a component that can both publish and consume data ."}]}]}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/77/77e3a93604df2dc0a0b58fe96fa4b955.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"Reactive Streams API The scope is to find the smallest set of interfaces , These interfaces will describe the necessary operations and entities , So as to realize asynchronous data flow with non blocking back pressure . Community for Reactive Streams There are many implementations of , Here's a simple summary and comparison ."}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/df/df444a5aafde6247ea0c6c40596912b0.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" To sum up , If it's a mobile device rxjava It 's a better choice . If it is used on the server side spring Framework development , The adoption is based on reactor Realized webflux More appropriate . If it's a high performance requirement , A relatively simple business scenario , choice vertx Can maximize the performance of the machine . and gio The real scene is a complex business system on the server side , Use at the same time scala As a development language and use play As web Development framework . So at the beginning of the system construction, it's natural to choose akka As the implementation basis of our responsive system ."}]},{"type":"heading","attrs":{"align":null,"level":1},"content":[{"type":"text","text":" Use Actor Building reactive systems "}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" In the first place, there was no direct use of akka-stream, It's a simpler choice , More powerful in modeling akka-actor As the basis of system implementation .Akka-actor Is based on actor Asynchronous toolkit for model building , Use akka-actor It's easy to program asynchronously based on message driven .Actor The basis of this is messaging , One actor It can be regarded as a basic unit of calculation , It can receive messages and perform operations , It can also send messages to other actor.Actors Separate from each other , They don't share memory , therefore Actor There is no need to pay attention to a series of common multithreading problems such as locking and memory atomicity ."}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/2c/2c89fb57c5767845107b433d3df02b24.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"Akka-actor The core implementation consists of three parts ："}]},{"type":"numberedlist","attrs":{"start":null,"normalizeStart":1},"content":[{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":1,"align":null,"origin":null},"content":[{"type":"text","text":"Mailbox： It can be a bounded or unbounded message queue , Used to store all received messages ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":2,"align":null,"origin":null},"content":[{"type":"text","text":"Behavior： Specific message processing logic ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":3,"align":null,"origin":null},"content":[{"type":"text","text":"State：actor The state of inclusion , Every actor The states of are independent to avoid lock contention ."}]}]}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"Actor It's not thread bound , Of the same process actor Share a thread pool ,mailbox It's a runnable object , The core logic is to take the message out of the queue and call behavior To deal with ."}]},{"type":"codeblock","attrs":{"lang":"text"},"content":[{"type":"text","text":" override final def run(): Unit = {\n try {\n if (!isClosed) { //Volatile read, needed here\n processAllSystemMessages() // Process system level messages first \n processMailbox() // And then deal with ordinary messages \n }\n } finally {\n setAsIdle() //Volatile write, needed here\n dispatcher.registerForExecution(this, false, false)\n }\n }"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" In the same process , Can be adjusted by akka-actor Thread pool size for vertical load scaling . meanwhile ,akka-actor Support binding different types in a system 、 Number of thread pools . For example, in some time-consuming IO In the scenario, a thread pool can be configured separately for the purpose of isolation . For scenarios that need to scale out ,akka Provides the basis for gossip The point-to-point decentralized clustering solution of the protocol akka-cluster."}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/bc/bc480b382871e121d45c5d10c206bc0a.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"Akka-cluster adopt gossip The protocol performs discovery and state synchronization among members , At the same time, it provides a higher level cluster tool ："}]},{"type":"numberedlist","attrs":{"start":null,"normalizeStart":1},"content":[{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":1,"align":null,"origin":null},"content":[{"type":"text","text":"Cluster Singleton： Global unique instance , It can guarantee the global uniqueness of the instance , At the same time, when the instance has problems clsuter It can be rebuilt on another node ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":2,"align":null,"origin":null},"content":[{"type":"text","text":"Cluster Sharding： adopt sharding, In the cluster actor Can cross multiple nodes through actorRef Identification for interaction , Don't care about their physical location in the cluster ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":3,"align":null,"origin":null},"content":[{"type":"text","text":"Distributed Data"},{"type":"text","marks":[{"type":"strong"}],"text":"："},{"type":"text","text":" When needed in a cluster When sharing data between nodes of ,Distributed Data Provides k/v Storage API."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":4,"align":null,"origin":null},"content":[{"type":"text","text":"Distributed Publish Subscribe"},{"type":"text","marks":[{"type":"strong"}],"text":"："},{"type":"text","text":" In the cluster actor Can publish and subscribe to peer-to-peer broadcast messages ."}]}]}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" In theory use akka-actor and akka-cluster It can make the system have strong expansibility （ elastic ）. But in practice, we don't use akka-cluster To expand the system , The reason is also simple ,akka-cluster Too few production cases , Too complex in function , Not conducive to large-scale promotion . Finally, we use the traditional message oriented middleware as the solution for the horizontal expansion of the system . Use in a single machine akka-actor, Scenarios involving cross node communication use message middleware for communication ."}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/e6/e6c594f564fdeff687b65ebdf2fa2bdc.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" In terms of system resilience ,akka-actor Provides a hierarchy based oversight mechanism . You can put the whole actor The system is seen as a tree , Every actor An instance is a node in the tree . The monitoring mechanism refers to every actor It's all his sons actor Supervisor of , You need a needle pair actor Develop an error handling strategy ."}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/fb/fb806beb0b70f95f196c2664445c407a.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" "}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" Corresponding to the specific business system , We split the whole process into several actor Realization , In order to achieve supervision and error recovery , You need to create a top level route actor To refer to all specific businesses actor . If a business actor Encountered a problem and threw an exception , Exceptions can be regulated route actor To deal with it . Regulators can choose to restore the problematic actor Or restart , It may also stop it , It depends on the severity of the problem and the recovery strategy .Akka-actor There are the following 4 Error handling strategies ："}]},{"type":"numberedlist","attrs":{"start":null,"normalizeStart":1},"content":[{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":1,"align":null,"origin":null},"content":[{"type":"text","text":" Restore child nodes , Keep the internal state of the current accumulation of child nodes ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":2,"align":null,"origin":null},"content":[{"type":"text","text":" Restart the child node , Clear the internal state of child nodes ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":3,"align":null,"origin":null},"content":[{"type":"text","text":" Permanently stop child nodes ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":4,"align":null,"origin":null},"content":[{"type":"text","text":" Throw error, pass error up , Handled by more advanced nodes ."}]}]}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" In the end, we are based on actor The whole business process is realized ： When a user initiates an out of site activity request , Main application （ be based on play） The active metadata will be written to the database and the result will be returned to the front end immediately , To achieve the goal of timely response . At the same time, encapsulate the activity request into a actor news , Delivered asynchronously to route actor Follow up task processing .Route actor According to the specific type of message received, routing distribution will be carried out , Namely User Insight Actor（ Search for crowd information ） - Build Push Task Actor（ Inquire about db assemble task） - Checkpoint Actor（ Storage task Information ）- Publish Task Actor（ Release task To kafka）."}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/2c/2c30538b5e3ebcbfeb15ab113bf243b6.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" The message driven design of the system has achieved some benefits ："}]},{"type":"numberedlist","attrs":{"start":null,"normalizeStart":1},"content":[{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":1,"align":null,"origin":null},"content":[{"type":"text","text":" Less coupling between programs , Every actor You just need to maintain a little bit of logic ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":2,"align":null,"origin":null},"content":[{"type":"text","text":" The whole process is asynchronous , Good user experience ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":3,"align":null,"origin":null},"content":[{"type":"text","text":" Messages can be produced and consumed across servers , Scale out becomes simple ."}]}]}]},{"type":"heading","attrs":{"align":null,"level":1},"content":[{"type":"text","text":" from Actor To Stream"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" As mentioned above, we created a route actor Will all business actor Get together , This can serve as a supervisory role , You can also know the global logical view . But there's a problem with this implementation , The overall arrangement is more complicated . It is more difficult to describe the flow with branching and merging logic , There are no constraints on the subsequent new processes , Only a sequence can be agreed artificially , For example, the one on the top is closer to the front , Poor maintainability ."}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/84/84ad974f4fcf767e5bb72c651be33d60.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" And because of the whole process User Insight Actor Partial dependence on external data query system , It is easier to become the bottleneck of the whole system . With the load changing , External queries may fail , This will affect the overall usability of the system . To solve this problem, we need to design the corresponding current limiting mechanism and the retrial mechanism . above-mentioned Actor Of mailbox It's a queue in itself , If the load is too high, the message can be discarded , Just specify actor Of maibox The type is bounded queue . If the message cannot be discarded , The token bucket algorithm can be used to realize the current limiting function . For retry mechanism ,User Insight Actor Itself is stateless , It's natural to think of sending the trial message again to User Insight Actor Try again on your own ."}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/80/80a49ee25a7ee223d3d14bae33e59271.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" This plan is relatively simple , If you want to meet the needs of some special scenarios , For example, set the number of retries , Delay execution of the retry request , Specify the demotion policy after failed retries , Only by customizing some logic , But it takes a lot of time to design code flexibly ."}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" All of the above solutions can meet the business needs , Generally speaking, through actor The model can quickly realize asynchronous encapsulation of lightweight business , But in the face of relatively complex business logic, there are still some limitations ："}]},{"type":"numberedlist","attrs":{"start":null,"normalizeStart":1},"content":[{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":1,"align":null,"origin":null},"content":[{"type":"text","text":" It is difficult to implement multi asynchronous task arrangement simply and elegantly , The routing scheme is too complex , Is not intuitive ."}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":2,"align":null,"origin":null},"content":[{"type":"text","text":" Retry mechanism 、 The reusability of service independent functions such as current limiting mechanism is not high ."}]}]}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" That's why... Was adopted later akka-stream To reconstruct the process flow .Akka-stream Is based on akka-actor Of Reactive Streams Specification implementation , It has the following characteristics ："}]},{"type":"bulletedlist","content":[{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" The ability to handle an infinite number of elements "}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" Processing elements asynchronously and sequentially "}]}]},{"type":"listitem","content":[{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" Achieve non blocking back pressure "}]}]}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" It also provides a more abstract and flexible DSL encapsulation , namely source、sink、flow Components ."}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"Source That is the source of the response flow , The source has a data outlet . We can create a Source："}]},{"type":"codeblock","attrs":{"lang":"text"},"content":[{"type":"text","text":"val sourceFromRange = Source(1 to 10)\nval sourceFromIterable = Source(List(1, 2, 3))\nval sourceFromFuture = Source.fromFuture(Future.successful(\"hello\"))"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"Sink It's the ultimate destination of the flow , Contains a data entry , We can create Sink："}]},{"type":"codeblock","attrs":{"lang":"text"},"content":[{"type":"text","text":"val sinkPrintingOutElements = Sink.foreach[String](println(_))\nval sinkCalculatingASumOfElements = Sink.fold[Int, Int](0)(_ + _)"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"Flow It's the middle component of the stream , Contains a data entry and data exit . We can create Flow："}]},{"type":"codeblock","attrs":{"lang":"text"},"content":[{"type":"text","text":"val flowDoublingElements = Flow[Int].map(_ * 2)\nval flowFilteringOutOddElements = Flow[Int].filter(_ % 2 == 0)\nval flowBatchingElements = Flow[Int].grouped(10)"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" The whole business flow can be represented by the graph and network of basic components ："}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/96/968594b61d9d7345a80f84399ac3e1ee.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" Flow operation can be compared to pipeline operation , Every operator is a processing procedure , The data source is the processing of raw materials , After several processes, the final output of a finished product ."}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" As mentioned above, in order to control the system rate , The logic of current limiting is introduced , Such as the implementation of token bucket algorithm , Only when the program gets the token can it enter the next processing logic , This implementation is essentially synchronous blocking , And in reality, the downstream node may be able to carry more requests . In order to solve the problem of inconsistent processing speed between data source and downstream nodes , stay Reactive Streams The back pressure mechanism is introduced in the specification of , It is essentially a data request from the processor to the data source , So a way to adjust the speed .Akka-Stream Provides a set of back pressure functions out of the box , How it is realized and Reactive Streams Agreement , The downstream subscriber By sending subscription To the upstream publisher The number of elements that an active request needs to process . This allows rate control from the source of the entire data stream , use pull instead of push The system can maintain the maximum processing capacity on demand , And it doesn't collapse at the same time ."}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/14/149310577035dfe60bab6ab7e8dc5e1b.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" The following is based on akka stream The reconstructed processing flow , Simple comparison akka actor How to implement , The combination code based on operators is clearer and easier to read , Can easily achieve complex task scheduling ."}]},{"type":"image","attrs":{"src":"https://static001.geekbang.org/infoq/45/45ab5438bf067adce89cb89a10fbd6ba.png","alt":null,"title":"","style":[{"key":"width","value":"100%"},{"key":"bordertype","value":"none"}],"href":"","fromPaste":false,"pastePass":false}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" In terms of the underlying implementation ,akka-stream Underlying or based on akka-actor Carrying out the work , It's just providing a higher perspective at the top DSL encapsulation . This flexible programming method can greatly improve code reusability and maintainability ."}]},{"type":"heading","attrs":{"align":null,"level":1},"content":[{"type":"text","text":" summary "}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" This article records GrowingIO The service team summarizes the practice of reactive system design for specific business scenarios , From asynchronous programming to using actor Model building based on message driven system , In order to reduce the complexity of the system and improve the maintainability, we introduce akka-stram As a choreography framework for reactive flow . Last , I hope to communicate with students who are interested in reactive technology , Make a small advertisement ： Our engineering team continues to recruit ~ Server side 、 front end 、 Big data, all kinds of siege lions are missing , Interested friends welcome to smash resume "},{"type":"link","attrs":{"href":"https://www.growingio.com/joinus","title":"https://www.growingio.com/joinus"},"content":[{"type":"text","text":"https://www.growingio.com/joinus"}]},{"type":"text","text":"."}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" "}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":" Reference material ："}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"link","attrs":{"href":"https://info.lightbend.com/rs/558-NCX-702/images/COLL-ebook-Reactive-Microservices-Architecture.pdf","title":"https://info.lightbend.com/rs/558-NCX-702/images/COLL-ebook-Reactive-Microservices-Architecture.pdf"},"content":[{"type":"text","text":"https://info.lightbend.com/rs/558-NCX-702/images/COLL-ebook-Reactive-Microservices-Architecture.pdf"}]}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"link","attrs":{"href":"https://learning.oreilly.com/library/view/applied-akka-patterns","title":"https://learning.oreilly.com/library/view/applied-akka-patterns"},"content":[{"type":"text","text":"https://learning.oreilly.com/library/view/applied-akka-patterns"}]}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"link","attrs":{"href":"https://freecontent.manning.com/akka-in-action-why-use-clustering/","title":"https://freecontent.manning.com/akka-in-action-why-use-clustering/"},"content":[{"type":"text","text":"https://freecontent.manning.com/akka-in-action-why-use-clustering/"}]}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"link","attrs":{"href":"https://doc.akka.io/","title":"https://doc.akka.io/"},"content":[{"type":"text","text":"https://doc.akka.io/"}]}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","marks":[{"type":"strong"}],"text":" About GrowingIO"}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null},"content":[{"type":"text","text":"GrowingIO It is a leading one-stop digital growth solution service provider in China . For the product 、 operating 、 market 、 Data team and manager provide customer data platform 、 Advertising analysis 、 Product analysis 、 Intelligent operation and other products and consulting services , Help enterprises on the way to digital transformation , Improve data-driven capabilities , Achieving better growth ."}]},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}},{"type":"paragraph","attrs":{"indent":0,"number":0,"align":null,"origin":null}}]}