How to use buffer queue to realize high concurrent order business (glory Collection Edition)

 

Preface

Mainly in the project ( Small and medium-sized projects ) There is a payment order business ( Just deal with VIP, There's no inventory involved ), At present, the number of users has not come up , There is no problem at present , But think of if the number of users becomes larger , The number of concurrent orders becomes larger , There may be a series of problems , In my spare time , Did this demo Test related questions .

The possible problems are as follows ：

1. The order is repeated

2. High and low , Slower performance

Solution ：ThreadPoolExecutor Thread pool + Queue queue

1. First of all springBoot The framework of the project is as follows ：

2. The classes involved in the business test process , as follows

• BusinessThread class

package com.springboot.demo.Threads;

import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

@Component
@Scope("prototype")//spring  Multiple cases 
public class BusinessThread implements Runnable{

    private String acceptStr;

    public BusinessThread(String acceptStr) {
        this.acceptStr = acceptStr;
    }

    public String getAcceptStr() {
        return acceptStr;
    }

    public void setAcceptStr(String acceptStr) {
        this.acceptStr = acceptStr;
    }

    @Override
    public void run() {
        // Business operations 
        System.out.println(" Multithreading has processed the order insertion system , The order number ："+acceptStr);

        // Thread blocking 
        /*try {
            Thread.sleep(1000);
            System.out.println(" Multithreading has processed the order insertion system , The order number ："+acceptStr);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }*/
    }
}

• TestThreadPoolManager class

package com.springboot.demo.Threads;

import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Queue;
import java.util.concurrent.*;

@Component
public class TestThreadPoolManager implements BeanFactoryAware {

    // For from IOC Take the object inside 
    private BeanFactory factory; // If you realize Runnable The class of is through spring Of application.xml File Injection , It can be done by  factory.getBean() obtain , It's just a mention here 

    //  Minimum number of threads maintained by thread pool 
    private final static int CORE_POOL_SIZE = 2;
    //  Maximum number of thread pool maintenance threads 
    private final static int MAX_POOL_SIZE = 10;
    //  The thread pool maintains the free time allowed for threads 
    private final static int KEEP_ALIVE_TIME = 0;
    //  The size of the buffer queue used by the thread pool 
    private final static int WORK_QUEUE_SIZE = 50;

    @Override
    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
        factory = beanFactory;
    }

    /**
     *  For orders stored in the queue , Prevent duplicate submissions , In the real world , You can use redis Instead of   Duplicate verification 
     */
    Map<String, Object> cacheMap = new ConcurrentHashMap<>();


    /**
     *  Buffer queue for orders , When the thread pool is full , Then put the order in this buffer queue 
     */
    Queue<Object> msgQueue = new LinkedBlockingQueue<Object>();


    /**
     *  When the thread pool is full , Execute the following code , Put the order in the buffer queue 
     */
    final RejectedExecutionHandler handler = new RejectedExecutionHandler() {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            // Order added to buffer queue 
            msgQueue.offer(((BusinessThread) r).getAcceptStr());
            System.out.println(" The system task is too busy , Give this order to ( Scheduling thread pool ) One by one processing , The order number ：" + ((BusinessThread) r).getAcceptStr());
        }
    };


    /** Creating a thread pool */
   final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS, new ArrayBlockingQueue(WORK_QUEUE_SIZE), this.handler);


    /** Add tasks to the order thread pool */
    public void addOrders(String orderId){
        System.out.println(" This order is ready to be added to the thread pool , The order number ：" + orderId);
        // Verify that the currently entered order already exists 
        if (cacheMap.get(orderId) == null) {
            cacheMap.put(orderId, new Object());
            BusinessThread businessThread = new BusinessThread(orderId);
            threadPool.execute(businessThread);
        }
    }

    /**
     *  Scheduled tasks of thread pool ---->  be called ( Scheduling thread pool ). This thread pool supports   The need to perform tasks regularly and periodically .
     */
    final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(5);


    /**
     *  Check ( Scheduling thread pool ), Once a second , Check whether there is a buffer queue for the order   Order records , Then rejoin the thread pool 
     */
    final ScheduledFuture scheduledFuture = scheduler.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            // Determine whether the buffer queue has records 
            if(!msgQueue.isEmpty()){
                // When the queue capacity of thread pool is less than WORK_QUEUE_SIZE, Then start buffering the queued orders   Add to   Thread pool 
                if (threadPool.getQueue().size() < WORK_QUEUE_SIZE) {
                    String orderId = (String) msgQueue.poll();
                    BusinessThread businessThread = new BusinessThread(orderId);
                    threadPool.execute(businessThread);
                    System.out.println("( Scheduling thread pool ) Order business appears in buffer queue , Re add to thread pool , The order number ："+orderId);
                }
            }
        }
    }, 0, 1, TimeUnit.SECONDS);


    /** Get message buffer queue */
    public Queue<Object> getMsgQueue() {
        return msgQueue;
    }

    /** Terminate order thread pool + Scheduling thread pool */
    public void shutdown() {
        //true Indicates that if a scheduled task is executing , To suspend immediately ,false Then wait for the task to finish before stopping 
        System.out.println(" Terminate order thread pool + Scheduling thread pool ："+scheduledFuture.cancel(false));
        scheduler.shutdown();
        threadPool.shutdown();

    }
}

• TestController class

package com.springboot.demo;

import com.springboot.demo.Threads.TestThreadPoolManager;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import java.util.Queue;
import java.util.UUID;

/**
 * Created by Administrator on 2018/5/9.
 */
@RestController
public class TestController {

    @Autowired
    TestThreadPoolManager testThreadPoolManager;

    /**
     *  Test simulation order request   entrance 
     * @param id
     * @return
     */
    @GetMapping("/start/{id}")
    public String start(@PathVariable Long id) {
        // Simulated random number 
        String orderNo = System.currentTimeMillis() + UUID.randomUUID().toString();

        testThreadPoolManager.addOrders(orderNo);

        return "Test ThreadPoolExecutor start";
    }

    /**
     *  Out of Service 
     * @param id
     * @return
     */
    @GetMapping("/end/{id}")
    public String end(@PathVariable Long id) {

        testThreadPoolManager.shutdown();

        Queue q = testThreadPoolManager.getMsgQueue();
        System.out.println(" Thread service is turned off , And the number of unprocessed messages ：" + q.size());
        return "Test ThreadPoolExecutor start";
    }
}

3. Use JMeter Simulate concurrent order requests

4. result

Printed log description , The start order is executed directly into the system , When the thread pool is full , Then use RejectedExecutionHandler Method to add subsequent orders to Queue Buffer queue , Use ScheduledFuture Timing method ( I'm here once a second ) Check Queue queue , Add the orders in the queue to the thread pool again , Perform the Insert task later .

Some of the logs are as follows