The business scenario

We do is come on business, the user can through the current position and target location in the app for map route and route way all the gas stations, the route query invokes the gold map provided interfaces, way station is according to the return route information query, so when the user input the starting position and target position after clicking the query will do the following steps:

  1. Call the Amap interface to get the route
  2. According to the route information returned by Amap to query all the fuel stations in the path

Trouble spots

With the rapid development of the company, there are many oil stations on the platform. When the starting point and target point are far away from each other, there will be a large number of oil stations through the channel. It will be time-consuming to use ordinary database query alone. The first step is to call the Amap API, which also has some delay. So how do we optimize?

In order to reduce the interface time consumption and improve user experience, we need to optimize the interface implementation. We cannot optimize it by calling Autonavi API, so we can only optimize the oil station in the query path.

Optimization idea

When too much gas station, a query can be very time consuming, so we can consider partial multi-threaded concurrent to the query, will be a long line according to the path length is divided into a number of conditions, such as a path for up to 800 km, we can put the query parameters of 800 km split into several smaller distance parameter set (ps: convenient for everyone to understand, Actual path planning queries are based on multiple parameters such as latitude, longitude and distance. For example, {[0, 50], [50100], [100150]… [750,800]}, then we open multiple threads to query concurrently according to the new query conditions, and finally splicing the results back, so as to achieve the purpose of reducing the query time.

Although the idea is easy to understand, there are two caveats to implementing it, and I’ll list them to see if you’ve considered them.

  • According to the business scenario, it is not a simple asynchronous query, but requires all threads to complete the execution and return the query results after combining, so synchronization control is required here. Here we use the CountDownLatch synchronization component provided by the JDK.
  • An in-thread operation requires a return value, which is implemented using the Callable interface and FutureTask.

The specific implementation

1. Normally, threads need to implement the Runnable interface, but threads that need to return values need to implement the Callable interface.

@Component
@Slf4j
@Scope("protoType") // It is important to note that the beans injected by Spring are singletons by default. In current business scenarios, multiple threads are required to perform query operations, so we declare the component as protoType mode
public class PathPlanTask implements Callable<List<Object>> {
	// Query parameters
    private PathPlanPartQuery pathPlanPartQuery;
    
    private CountDownLatch countDownLatch;
    
    @Override
    public List<Object> call(a) throws Exception {
        try {
            //TODO business query
            List<Object> result = queryList(pathPlanPartQuery);
            // Return the result
            return result;
        }catch (Exception e){
            // Error logs are printed
            log.error("query PathByGasstation error!");
        }finally{
            Countdownlatch.await () will be released when the operation like I -- is reduced to 0, otherwise it will block.countDownLatch.countDown(); }}public void setPathPlanPartQuery(PathPlanPartQuery pathPlanPartQuery){
        this.pathPlanPartQuery = pathPlanPartQuery;
    }

    public void setCountDownLatch(CountDownLatch countDownLatch){
        this.countDownLatch = countDownLatch;
    }

    private List<Object> queryList(PathPlanPartQuery pathPlanPartQuery) {
        // TODO query logic is omitted here
        returnLists.newArrayList(); }}Copy the code

2.Callable is usually used with FutureTask to obtain the return value of the thread through FutureTask’s GET method.

// Usually defined as a utility class for fetching
private static final ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(8.20.1000,
        TimeUnit.SECONDS, new ArrayBlockingQueue<>(50), new ThreadPoolExecutor.AbortPolicy());

// Business code
 private List<Object> queryGasInfoBaseDtoList(List<PathPlanQueryParam> queryParamList) {
        long stMills = System.currentTimeMillis();
     	// Define thread pool for multithreading management, using Util for static thread pool
        
		// define countDownLatch. The constructor passes the size of the set of parameters
     	{[0,50],[50,100],[100,150]... [750800]}
        CountDownLatch countDownLatch = new CountDownLatch(queryParamList.size());
		// Define the FutureTask set
        List<FutureTask<List<GasInfoBaseResponseDto>>> futureTaskList = Lists.newArrayList();
        try {
            // Iterate over the query parameter set
            for (PathPlanQueryParam queryParam : queryParamList) {
                // Use the getBean method here.
                PathPlanTask pathPlanTask =
                    ApplicationContextProvider.getBean("pathPlanTask", PathPlanTask.class);
            	/ / set countDownLatch
                pathPlanTask.setCountDown(countDownLatch);
                // Get query parameters
                PathPlanPartQuery pathPlanPartQuery = getPathPlanPartQuery(queryParam);
                pathPlanTask.setPathPlanPartQuery(pathPlanPartQuery);
                // Define FutureTask, taking the defined Callable implementation class as a construction parameter
                FutureTask<List<GasInfoBaseResponseDto>> futureTask = new FutureTask<>(pathPlanTask);
                // To the thread pool to execute
                poolExecutor.submit(futureTask);
                // Add the futureTask collection
                futureTaskList.add(futureTask);
            }
            // It blocks until the countdownlatch.countdown () method reduces the size argument passed at creation to 0.
            // This block can ensure that multiple threads have completed the execution of the final return.
            countDownLatch.await();
			
            // We concatenate the final result after executing multiple threads
            List<Object> gasInfoDtoList = Lists.newArrayList();
            for (FutureTask<List<Object>> futureTask : futureTaskList) {
                Futuretask.get () is blocked while the thread is still executing and has not returned
                List<Object> baseResponseDtoList = futureTask.get();
                if(CollectionUtils.isNotEmpty(baseResponseDtoList)) { gasInfoDtoList.addAll(baseResponseDtoList); }}return gasInfoDtoList;
        } catch (Exception e) {
            log.error("queryGasInfoBaseDtoList_err", e);
        } finally {
            log.info("queryGasInfoBaseDtoList_requestId:{},batchTimes:{},cost:{}", pointRequestOld.getRequestId(),
                pointRequestOld.getBatchTimes(), System.currentTimeMillis() - stMills);
        }
        return null;
    }
Copy the code

conclusion

The above is the actual application scenario of multi-threading encountered in my work, which can be summarized as starting multiple Callable threads through the thread pool to query data in batches, introducing CountDownLatch component to control the end time of query, and then using FutureTask’s GET method to obtain the final results and assemble and return.

I will update some of the underlying knowledge of multithreading, through the source code to introduce AQS, CountDownLatch, thread pool and so on the working principle, welcome your attention!