The article directories

• introduce
• Singleton class
• • The main role
  • Demand reason
  • Implementation method
• The asynchronous solution
• • Producer-consumer design patterns
  • Celery is introduced and used
  • • introduce
    • The actual use
    • Creating an environment Directory
    • Starting an Asynchronous Task
    • Calling an asynchronous task

introduce

This blog introduces the singleton and producer-consumer design patterns. Both development patterns are available in most Python frameworks. My personal projects are mostly Django.

Singleton class

The main role

Reduce class instantiation times and memory space address usage, reduce memory usage.

Demand reason

Implementation method

The current requirement is that if we instantiate a class, if the class is already created, we can use it instead of instantiating it in general. To create a class, we need to use a magic method in Python class __new__(CLS, *args, **kwargs) is a staticmethod that creates class instances without the @staticmethod decorator and takes precedence over the __init__(self) initializer. In human language, __new__(CLS, *args, **kwargs) is called when a class is created. The class is created before initializing __init__(self), so there is no self(the instantiated self of the class) but CLS (the class itself). So when we create a class, we make a judgment to determine whether the class already exists, and if it does, we return the existing class directly, rather than instantiating it again.

classThe test class:
    def __new__(cls, *args, **kwargs) :
        The # hasattr() method returns whether the object has an attribute with the given name.
        Check whether the class attribute is an empty object
        if not hasattr(cls, Instance '_') :If empty, the parent method is called to create memory space for the first objectCLS. = _ instancesuper().__new__(cls, *args, **kwargs)
            # return the class attribute holding the object reference to __init__
        returnCls. _ Instances without optimization Instantiation I = Test class () Without optimization instantiation II = Test Class ()print('Example 1 :'.id(No optimization instantiation I)print('Example 2 :'.id(No optimization instantiation ii))Copy the code

Note: because__new__()Will be preferred__init__()The initialization method is called, so it is best not to initialize attributes in __init__ after using the singleton class. If you need to initialize a value, you can do so in the__new__()If you need to pass a value to a class, pass it to a method.

classThe test class:
    def __new__(cls, *args, **kwargs) :
        The # hasattr() method returns whether the object has an attribute with the given name.
        Check whether the class attribute is an empty object
        if not hasattr(cls, Instance '_') :If empty, the parent method is called to create memory space for the first objectCLS. = _ instancesuper().__new__(cls, *args, **kwargs)
            # return the class attribute holding the object reference to __init__CLS. = test123
        returnCLS. _ instancedefTests using initialization values (self) :
        print(self test)defTest the incoming value (The self, the value) :
        print(Value) Singleton 1 = Test class () Singleton 1 = Test class ()print('Example 1 :'.id(Singleton 1)print('Example 2 :'.id(Singleton 1)) Singleton 1. The test uses the initialization value () singleton one. Test the incoming value (321)
Copy the code

The asynchronous solution

Producer-consumer design patterns

There is a mature design pattern in asynchronous solutions: the producer-consumer design pattern.

This design pattern divides asynchrony into four modules:task,producers,consumers,A middleman.



If you want to learn more about this pattern, check this outA blog written by a big manVery graphic.

Celery is introduced and used

introduce

pip install Celery
Copy the code

The actual use

In the introduction above we mentioned the message queue, this message queue is used to store the output data generated by the producer to send to the consumer, so he needs to be a database, here I used Redis, celery, currently also supports RabbitMQ, Redis, Amazon SQS etc database. You can go to the official documents for details.

Suppose we are currently acting as a Django back end and need to send an email to the user. Tasks: Send mail producers: Django backend consumers: celery intermediaries: message queues

Creating an environment Directory

# main.py
from celery import Celery

Entrance to # celery

Create celery instanceCelery = celery ('celery_app')

Load the configuration fileCelery asynchronous application. Config_from_object ('celery_tasks.config')

# register taskAutomatic tasks(['Celery_tasks. Task module'])
"" Linux runs celery -a celery_tasks. Main worker-l info • -a denotes the corresponding application whose argument is the position of the celery instance in the project. • Worker refers to the worker to start here. • -l indicates the log level, such as info level. Celery_tasks. Main worker-l info --pool=solo ""
Copy the code

# config.py
Celery configuration file
from Django_Module.settings importRedis database# specify the middleman, message queue, task queue redis

Redis ://[: password] @host address: port/database number
# such as
# broker_url = f "redis: / / : [email protected]/4"
broker_url = f"redis://:{Redis database ['password']}@{Redis database ['address']}:{Redis database ['port']}/ 4"
Copy the code

Tasks. py: Here is the start page of the task module. The file name must be tasks.py where the task you want to perform asynchronously must be decorated as shown below.

Tasks.py
# Define tasks
fromCelery_tasks. Task module. Mail sending module. Mail sending libraryimportSend E-mailfrom celery_tasks.main importCelery asynchronous applicationCelery tasks, name = task name
@celery asynchronous application. task(name="Send asynchronous mail")
defSending asynchronous mail (Recipient email address, sender =Experimental Management Platform of Wuchang Shouyi University, email subject =Experimental Management Platform of Wuchang Shouyi University, mail type =1, access link =None, custom content =None) :Return message = Send message (recipient email, sender, subject, type, access link, custom content)returnReturn informationCopy the code

Task module > Mail sending module > Mail sending library. py: This is the file you need to execute asynchronously, here I just write a simple mail sending module.

import random
import smtplib
from email.mime.text import MIMEText
from email.utils import formataddr
# Introduce global variables
from Django_Module.settings importemailSend validation module
defGenerate captcha (Median =4) :
    Generate a random captcha with uppercase and lowercase letters + digits: Param bits: Captcha bits (default =4) :return: Captcha """Verification code =' '
    # 48-58 is ASCII 0~9,65-90 is A~Z, and 97-122 is A~ZRandom number ASCII =list(range(48.58)) * 3 + list(range(65.91)) + list(range(97.123))
    for i in range(bits): random character =chr(random. Choice (ASCII))# 65~90 is ASCii code A~ZVerification code + =str(Random character)returnVerification codedefSend mail (Recipient email address, sender =College Experiment Management Platform, email subject =College Experiment Management Platform, mail type =1, access link =None, custom content =None) :
    "" Call the official email address to send an email :param Recipient Email address: Param Sender: User name of the sender: Param Email title: Email title :param Email type: 1: sends an authentication email with a six-digit verification code. 2: the operation of mailbox binding or password retrieval. 3: custom email content: param access links: take effect only when 2, specify the access url links: param custom content: only in for 3: return: send the success (if the mail type is automatically after 1 sent successfully returned authentication code) "" "
    ifMail type ==1: Verification code = Generate verification code (6)
        ifEmail subject ==College Experiment Management Platform: Email subject = Email subject +'Registration Verification Code Mail'Email content =
      

        '
      + Verification code +'
 The verification code is valid for five minutes, please use it as soon as possible. 
'Return value = verification codeelifMail type ==2: Email content =<p style=" margin: 0px; padding: 0px; text-align:center; font-size: 2rem; color: #333333; background-color: #CCFFFF; width: 20%;" > <a href="{access link}" style="font-weight: bold; color: #2c3e50; text-decoration:none;" > Confirm that the binding </a> </div> <p> is valid within 10 minutes, please resend the binding if it expires. </p> <p> If the link fails please click:{access link}</p> <p> If this operation is not performed by myself, change the password as soon as possible. </p> """Return value =True
    elifMail type ==3: Mail content = User-defined content Return value =None
    try: body = MIMEText(message content,'html'.'utf-8') the text ['From'] = formatAddr ([sender, email]'Use email'] ['account'[text]])'To'] = Recipient mailbox body ['Subject'] = Email header server = smtplib.smtp_ssl (email ['Use server'] ['address'], mailbox ['Use server'] ['port']) server.login(email ['Use email'] ['account'], mailbox ['Use email'] ['password']) server.sendmail(email ['Use email'] ['account'], [recipient mailbox,], body.as_string ()) server.quit()except Exception as e:
        tag = False
    else: tag = Return valuereturn tag
Copy the code

Starting an Asynchronous Task

Main worker-l info run celery -A celery_tasks. Main worker-l info –pool=solo

• -A refers to the corresponding application whose argument is the position of the Celery instance in the project.
• Worker refers to the producer to be started here.
• -l indicates the log level, such as info level.

Note: Make sure you are in the task path when starting the task. If not, either CD to the task path or change celery_tasks.main to the absolute path of the task

Calling an asynchronous task