Stress test modification solution

primary :

Introduce

​ The project is xxl-job Next scheduled task Own an actuator Therefore, it can be enabled according to one node Exclusive port

​ The project java Package configuration :

​ jdk8

​ Memory size :2GB

​ Do not use third-party caching and message processing software Pure primary

​ Business content ： Accept messages of Foreign Exchange , Analytic message , Determine the message type , Judge the primary key of the message to be processed , Update if in the database , If not in the database, insert ( The number of message primary keys is small , Mainly update , The operation of inserting is negligible )

​ Processing mode : Adopted LinkedBlockingQueue Put all the received messages into the queue

Then start another thread to process the queue information Then judge whether the message is processed before database operation

Under pressure test :

​ The pressure test time is one hour and the data volume is one million , The amount of data to be processed is 5w about

​ Pressure test 10 Minutes less than Queue memory exceeded 2G The direct stack is full

PLAN1

​ Processing mode : Judge and process the message before inserting it into the queue , Only the messages that need to be processed are added to the blocking queue

Under pressure test :

​ A little smoother than the original way , but 20 Minutes or so, the stack exploded , Check the log and find that the queue has been inserted within one second 100 Data to be processed , But one second only deals with 10 Data , The processing speed is far from keeping up .

PLAN2

​ Processing mode : Put the message into memory , Create a management class , The thread receiving the message calls this managed class to insert the message into the corresponding map in , The thread processing the message calls this managed class to take out the data that needs to be updated and update .

​ The reason for this treatment : The processing speed in memory is faster than that of querying the database to determine whether to process , Many messages correspond to the same primary key, and the operation of the database is more about updating , Will not squash the stack

​ Processing steps :

• Create a message class that is compatible with all message fields ( Take the union of all messages ) And whether to update the field
• Create a management class There is a private map key Primary key value It is a message class There is an add method in the management class , Set the received message update field to true, And then put in map in , There is also a method in the management class to get the message queue that needs to be updated , First new A blocking queue pair map Traversal will update the field to true The message class update field of is set to false Then put it into the blocking queue , Finally, return to the blocking queue
• The thread receiving the message calls the add method The thread processing the message calls the method of obtaining and updating the message queue , Process the data in the queue

Because these two methods are similar to write and read operations , So both methods should be locked

Writing of management class :

  private HashMap<String, BankQuoteNode> quoteMap = new HashMap<>();//k It is your primary key 

    @Override
    public synchronized void addQuote(BankQuoteNode node) {
    
        node.setUpdateFlag(true);
        StringBuilder key = new StringBuilder();
        key.append(node.getMarketInicator()).append('-').append(node.getSecurityCode()).append('-').append(node
                .getSecurityName());
        String quoteMapKey=key.toString();

        quoteMap.put(quoteMapKey,node);
    }


    @Override
    public synchronized Queue<BankQuoteNode> getAllQuote() {
    

        LinkedBlockingQueue<BankQuoteNode> quoteQueue =new LinkedBlockingQueue<>();
        quoteMap.forEach((k,v)->{
    
            quoteQueue.add(v);
        });
        return quoteQueue;// Back to all 
    }

    @Override
    public synchronized Queue<BankQuoteNode> getRefreshedQuote() {
    
        LinkedBlockingQueue<BankQuoteNode> refreshedQuoteQueue = new LinkedBlockingQueue<>();
        // Traverse quoteMap
        // hold UpdateFlag=true Of node Insert the returned queue 
        quoteMap.forEach((k,v)->{
    

            if(v.isUpdateFlag()){
    
                quoteMap.get(k).setUpdateFlag(false);
                refreshedQuoteQueue.add(v);
            }
        });
        return refreshedQuoteQueue;
    }