Project Efficiency Doubled with Request Merging!

01IntroductionWhat is the significance of request merging? Let’s take a look at the diagram below.Assuming we have three users (user IDs 1, 2, and 3), and they all want to query their basic information. Each sends a request to the server, which in turn queries the database, resulting in three requests. We all know that database connection resources are quite precious, so how can we save connection resources as much as possible?Here, if we replace the database with a remote service being called, the same principle applies.Let’s change our approach, as shown in the diagram below.We merge the requests on the server side and only send one SQL query to the database. After the database returns the data, the server processes the returned data and groups it based on a unique request ID, returning it to the corresponding user.02Technical Methods

• <span><span>LinkedBlockQueue</span></span>Blocking Queue
• <span><span>ScheduledThreadPoolExecutor</span></span>Scheduled Task Thread Pool
• <span><span>CompleteableFuture</span> <span>future</span></span>Blocking Mechanism (Java 8’s CompletableFuture does not have a timeout mechanism; later optimizations used a queue instead)

Code Implementation

• Code for querying users

public interface UserService {

    Map<String, Users> queryUserByIdBatch(List<UserWrapBatchService.Request> userReqs);
}
@Service
public class UserServiceImpl implements UserService {

    @Resource
    private UsersMapper usersMapper;

    @Override
    public Map<String, Users> queryUserByIdBatch(List<UserWrapBatchService.Request> userReqs) {
        // All parameters
        List<Long> userIds = userReqs.stream().map(UserWrapBatchService.Request::getUserId).collect(Collectors.toList());
        QueryWrapper<Users> queryWrapper = new QueryWrapper<>();
        // Use in statement to merge into one SQL to avoid multiple database IO requests
        queryWrapper.in("id", userIds);
        List<Users> users = usersMapper.selectList(queryWrapper);
        Map<Long, List<Users>> userGroup = users.stream().collect(Collectors.groupingBy(Users::getId));
        HashMap<String, Users> result = new HashMap<>();
        userReqs.forEach(val -> {
            List<Users> usersList = userGroup.get(val.getUserId());
            if (!CollectionUtils.isEmpty(usersList)) {
                result.put(val.getRequestId(), usersList.get(0));
            } else {
                // Indicates no data
                result.put(val.getRequestId(), null);
            }
        });
        return result;
    }
}

• Implementation of request merging

/***
 * zzq
 * Wrapper for batch execution
 * */
@Service
public class UserWrapBatchService {
    @Resource
    private UserService userService;

    /**
     * Maximum number of tasks
     **/
    public static int MAX_TASK_NUM = 100;

    /**
     * Request class, code is the common feature of the query, for example, querying products, distinguished by different IDs
     * CompletableFuture will return the processing result
     */
    public class Request {
        // Unique request ID
        String requestId;
        // Parameter
        Long userId;
        // TODO Java 8's CompletableFuture does not have a timeout mechanism
        CompletableFuture<Users> completableFuture;

        public String getRequestId() {
            return requestId;
        }

        public void setRequestId(String requestId) {
            this.requestId = requestId;
        }

        public Long getUserId() {
            return userId;
        }

        public void setUserId(Long userId) {
            this.userId = userId;
        }

        public CompletableFuture getCompletableFuture() {
            return completableFuture;
        }

        public void setCompletableFuture(CompletableFuture completableFuture) {
            this.completableFuture = completableFuture;
        }
    }

    /*
    LinkedBlockingQueue is a blocking queue, internally using a linked list, ensuring thread safety with two ReentrantLocks
    The difference between LinkedBlockingQueue and ArrayBlockingQueue
    ArrayBlockingQueue has a default length, while LinkedBlockingQueue has a default length of Integer.MAX_VALUE, which is an unbounded queue, and can easily cause OOM when the removal speed is less than the addition speed.
    ArrayBlockingQueue's storage container is an array, while LinkedBlockingQueue's storage container is a linked list.
    The locks for adding or removing elements in the two implementations are different; ArrayBlockingQueue uses the same ReentrantLock for both operations, while LinkedBlockingQueue uses separate locks for adding (putLock) and removing (takeLock), greatly improving throughput and allowing producers and consumers to operate in parallel in high concurrency scenarios, thus enhancing overall queue performance.
    */
    private final Queue<Request> queue = new LinkedBlockingQueue();

    @PostConstruct
    public void init() {
        // Scheduled task thread pool, creating a thread pool with a limited number of threads (1 in this case) that supports scheduled, periodic, or delayed tasks
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);

        scheduledExecutorService.scheduleAtFixedRate(() -> {
            int size = queue.size();
            // If the queue has no data, it means there are no requests during this time, return directly
            if (size == 0) {
                return;
            }
            List<Request> list = new ArrayList<>();
            System.out.println("Merged [