Shulou(Shulou.com)05/31 Report--

This article mainly introduces the relevant knowledge of "inter-process communication in python and how to set the state quantity to control another process". The editor shows you the operation process through an actual case. The operation method is simple, fast and practical. I hope that this article "inter-process communication in python and how to set the state quantity to control another process" can help you solve the problem.

I. Inter-process communication in python

Business scenario: in the current business scenario, we need to start an interval task, which runs an algorithm, and then processes the results of the algorithm and stores them in the database. The current interval of the task is one hour, and the running time of the algorithm is more than 50 minutes. There is not much time left for the result to be processed, so there may be a timeout. At present, the optimization direction will be more reasonable algorithmically, because the result processing does not take a lot of time. However, in this business scenario, if you want to compress the processing time of the result infinitely and compress it to 0, it is also possible to say that the compression is 0. In fact, after the completion of the execution of the algorithm, you start a process to process it. In this way, the operation of the algorithm will not be affected due to the need for data processing. The algorithm and the result processing are divided into two separate processes. At the beginning of the program, the algorithm operation and result processing are regarded as one cycle, but now the algorithm operation and result processing are divided into two cycles.

Technical implementation scheme:

Start two processes, one of which runs the algorithm, sends a status value to another process after the algorithm is finished, and the other process starts data processing after receiving the state quantity. As long as the two processes do not affect each other. In fact, it is also equivalent to the algorithm process controlling the data processing process.

Test scenario construction code:

From multiprocessing import Process Pipeimport timeimport sysimport osdef send_message (conn): for i in range (1000): print ('send_message:%d'%i) print (os.getpid ()) conn.send (I) time.sleep (3) def send_message1 (conn): # for i in range (1000): print (conn.recv ()) while True: if conn.recv ()% 5 = 0: Print ('today is nice day') time.sleep (1) if _ _ name__ = =' _ main__': # create a process communication pipeline left Right = Pipe () T1 = Process (target=send_message,args= (left,)) T2 = Process (target=send_message1,args= (right,)) t1.start () t2.start ()

There are some points to note in this case scenario:

First, the problem of time.sleep (), sleep for a specified time, will always go wrong, the specific cause of the error has not been found, this is the original problem, there is no long-term test here, so it may not occur, but we should pay attention to it.

Second, there are some differences between the code implementation and the above description, such as not enabling scheduling tasks, only starting a task that runs at intervals.

Third, the data processing process has been dealing with the empty running state, which will lead to a waste of resources (it is more reasonable to form a blocking state, but there is a lack of awareness of the construction of the blocking state, so the resources are sacrificed first.

4. Among the requirements described above, there is a scheduling task under control at the last node of algorithm operation and data processing, which is not reflected here. In fact, timing tasks and data processing should be separated as two cycles to better meet the requirements of the above description.

Second, set the state quantity to control another process

Business scenario: in the current business scenario, we need to start an interval task, which runs an algorithm, and then processes the results of the algorithm and stores them in the database. The current interval of the task is one hour, and the running time of the algorithm is more than 50 minutes. There is not much time left for the result to be processed, so there may be a timeout. At present, the optimization direction will be more reasonable algorithmically, because the result processing does not take a lot of time. However, in this business scenario, if you want to compress the processing time of the result infinitely and compress it to 0, it is also possible to say that the compression is 0. In fact, after the completion of the execution of the algorithm, you start a process to process it. In this way, the operation of the algorithm will not be affected due to the need for data processing. The algorithm and the result processing are divided into two separate processes. At the beginning of the program, the algorithm operation and result processing are regarded as one cycle, but now the algorithm operation and result processing are divided into two cycles.

The above solution only involves enabling two processes to run two tasks, and does not involve enabling the timed task framework, so it may appear inconsistent with the above business scenario, so re-solve it here. There is no problem with the above, but the timed task framework can be dealt with as a task. Then, after the scheduled task has finished running, pass a parameter to another process as the state quantity to start another process. Of course, here, the two processes are completely full, that is, dealing with the blocking state. The use of resources is still not at its best. Later, consider the way to use the process pool to see if you can let one of the processes run and release resources directly.

The technical solutions are as follows:

From multiprocessing import Process Pipeimport timefrom apscheduler.schedulers.background import BackgroundSchedulerfrom apscheduler.schedulers.blocking import BlockingSchedulerfrom apscheduler.schedulers.asyncio import AsyncIOScheduler# schedule = BackgroundScheduler () schedule = BlockingScheduler (timezone= "Asia/Shanghai") # schedule = AsyncIOScheduler (timezone= "Asia/Shanghai") def algorithm (conn): print ('start_run') conn.send (' please run') # time.sleep (5) def worth_result (conn): while True: if conn.recv () = 'please run': Print (conn.recv () + 'very nicetries') def time_job (conns): schedule.add_job (func=algorithm) Trigger='interval',seconds=5,args= (conns,) schedule.start () if _ _ name__ = ='_ _ main__': left,right = Pipe () T1 = Process (target=time_job,args= (left,)) T2 = Process (target=worth_result,args= (right,)) t1.start () t2.start ()

There are also some points to note here. It doesn't matter which type of scheduled task you choose, blocking and non-blocking actually have nothing to do with it, because we have started two processes directly here, and each process is independent of each other. It is not two processes that are enabled under scheduled tasks, so it will not be affected.

There are still questions about this solution:

First, as mentioned above, the two processes are full, so for resources, the utilization rate of the two processes has been very high.

Second, the scalability is insufficient, if there are other processes in this program that need to be dealt with, you need to add a process, or add it to the current process, code refactoring will be more troublesome.

Third, the control of the whole task is insufficient and needs to be improved. For example, for some control and viewing of the running status, if the general program runs for a long time, we should add such an interface, otherwise if there is no result after startup, we do not know its running status, which is a little passive.

Fourth, with regard to the third, using the logging library, it should be possible to output its logs directly, but the log library, as a third-party library, is equivalent to monitoring the entire running status, and whether it will take up another process needs to be tested.

5. Completeness and disaster recovery. If the program fails due to other problems such as resources, some data will be redundant, that is, some algorithms have not been processed. At this time, we need to consider how to supplement the data. What if the original file is not preserved? And what if these data are extremely important data? If the program dies, how to deal with it quickly? Reboot directly?

6. What if the data processing process takes more time than the algorithm? From the point of view of the current business, it is far lower, but what if it is much higher than? Can you distribute the processing work, use multiple machines to process it, and then combine the results?

The idea of distributed processing is getting stronger and stronger.

This is the end of the content about "Inter-process Communication in python and how to set the state quantity to control another process". Thank you for your reading. If you want to know more about the industry, you can follow the industry information channel. The editor will update different knowledge points for you every day.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.