1. Principle and mechanism

Use one-way pool to ensure orderly submission and execution.

explain:

1) Alternate execution between threads is not supported.

2) Multithreading is not synchronous, only to ensure the order of execution, multithreading is concurrent execution.

2. Code examples

package com.thread;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class TasksWithoutLockTest extends Thread {

    private static int num = 0;
    private int id;

    public TasksWithoutLockTest(int id) {
        this.id = id;
    }


    @Override
    public void run() {
        System.out.println("Thread" + id + "output num:" + num++);
        try {
            Thread.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        //ExecutorService exec = Executors.newFixedThreadPool(1);
        ExecutorService exec = Executors.newSingleThreadScheduledExecutor();
        for (int i = 0; i < 12; i++) {
            exec.submit(new TasksWithoutLockTest(i));
        }

        exec.shutdown();

    }
}

3. Output results

1. Principle analysis

Atomicinteger is thread safe, and autoincrement is thread safe.

2. Code examples

package com.thread;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

public class TasksTest extends Thread {
    private static AtomicInteger atomic = new AtomicInteger(1);
    private int id;

    public TasksTest(int id) {
        this.id = id;
    }

    @Override
    public void run() {
        while (atomic.get() <= 12) {
            while (atomic.get() % 3 == id) {
                System.out.println("thread_" + id + " output:" + atomic.get());
                atomic.incrementAndGet();
            }
        }
    }

    public static void main(String[] args) {
        Thread thread0 = new TasksTest(0);
        Thread thread1 = new TasksTest(1);
        Thread thread2 = new TasksTest(2);

        ExecutorService exec = Executors.newFixedThreadPool(3);

        exec.submit(thread0);
        exec.submit(thread1);
        exec.submit(thread2);

        exec.shutdown();

    }
}

Keywords: thread, thread pool, queue, task scheduling

1. Retrying tool class

package com.iretry;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * Retry Tools Class
 */
@Slf4j
public class RetryUtils {

    private final static int DEFAULT_RETRY_TIMES = 6;
    private final static int[] DEFAULT_DELAY_SECONDS = {3, 30, 180, 600, 1800, 3600};

    private static Queue<RetryRunnable> TASK_QUEUE = new ConcurrentLinkedQueue<>();

    public static ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    public static ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();

    static {
        //Configure the number of core threads
        executor.setCorePoolSize(10);
        // Configure the maximum number of threads
        executor.setMaxPoolSize(20);
        //configure queue size
        executor.setQueueCapacity(1000);
        //configure the name prefix of the threads in the thread pool
        executor.setThreadNamePrefix("async_");
        // rejection-policy: how to handle new tasks when the pool has reached max size
        // CALLER_RUNS: instead of executing the task in a new thread, the caller's thread will execute it
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //Perform initialization
        executor.initialize();

        scheduler.setThreadNamePrefix("scheduler_");
        scheduler.setPoolSize(5);
        scheduler.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        scheduler.initialize();
    }

    public RetryUtils() {
        //Check every second: iterate through the task queue, if needed, the thread pool schedules execution
        scheduler.scheduleAtFixedRate(() -> {
            for (RetryRunnable task : TASK_QUEUE) {
                long nextRetryMillis = task.nextRetryMillis;
                if (nextRetryMillis != -1 && nextRetryMillis <= System.currentTimeMillis()) {
                    task.nextRetryMillis = -1;
                    executor.execute(task);
                }
            }
        }, 1000);
    }

    public void doRetry(Task task) {
        doRetry(DEFAULT_RETRY_TIMES, DEFAULT_DELAY_SECONDS, task);
    }

    public void doRetry(int maxRetryTime, Task task) {
        doRetry(maxRetryTime, DEFAULT_DELAY_SECONDS, task);
    }

    public void doRetry(int[] retryDelaySeconds, Task task) {
        doRetry(retryDelaySeconds.length, retryDelaySeconds, task);
    }

    /**
     * @param maxRetryTime      Maximum number of retries
     * @param retryDelaySeconds array of retry interval times
     * @param task task
     */
    public void doRetry(final int maxRetryTime, final int[] retryDelaySeconds, final Task task) {
        Runnable runnable = new RetryRunnable(maxRetryTime, retryDelaySeconds, task);
        executor.execute(runnable);
    }

    /**
     * Custom Threads Class
     */
    private static class RetryRunnable implements Runnable {

        private final Task task;
        private final int maxRetryTimes;
        private final int[] retryDelaySeconds;

        private int retryTimes;
        private volatile long nextRetryMillis;

        //Constructors
        public RetryRunnable(final int maxRetryTimes, final int[] retryDelaySeconds, final Task task) {
            this.task = task;
            if (maxRetryTimes <= 0) {
                this.maxRetryTimes = DEFAULT_RETRY_TIMES;
            } else {
                this.maxRetryTimes = maxRetryTimes;
            }
            if (retryDelaySeconds == null || retryDelaySeconds.length == 0) {
                this.retryDelaySeconds = DEFAULT_DELAY_SECONDS;
            } else {
                this.retryDelaySeconds = retryDelaySeconds;
            }
        }

        //Implementation of business methods
        @Override
        public void run() {
            try {

                task.run();

            } catch (Throwable e) {
                int sleepSeconds = retryTimes < retryDelaySeconds.length ?retryDelaySeconds[retryTimes] : retryDelaySeconds[retryDelaySeconds.length - 1];

                if (retryTimes < maxRetryTimes) {
                    if (retryTimes == 0) {
                        TASK_QUEUE.add(this);
                        log.error("task executed error, " + sleepSeconds + " seconds do next... ", e);
                    } else {
                        log.error("retry " + retryTimes + " times error, " + sleepSeconds + " seconds do next... ", e);
                    }
                    nextRetryMillis = System.currentTimeMillis() + sleepSeconds * 1000;

                } else {
                    log.error("retry " + retryTimes + " times error", e);
                    log.error("retry snapshot: {}", JSON.toJSONStringWithDateFormat(task.snapshot(), "yyyy-MM-dd HH:mm:ss"));

                    TASK_QUEUE.remove(this);
                    task.retryFailed(e);

                }

                retryTimes++;

            }
        }
    }

}

2. Business class interface

package com.iretry;

public interface Task {

    void run() throws Exception;

    default void retryFailed(Throwable e) {
    }

    default Object snapshot() {
        return null;
    }
}

3. Test class

package com.iretry;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
public class IRetryTest {

    public static void main(String[] args) {
        RetryUtils retryUtils = new RetryUtils();

        IRetryTest iRetryTest = new IRetryTest();
        iRetryTest.executeBusiness(retryUtils, new IRetryTest().new UserInfo("Jack"));

    }

    //Business methods that require retries
    private void executeBusiness(RetryUtils retryUtils, UserInfo user) {

        retryUtils.doRetry(new int[]{6, 12}, new Task() {
            @Override
            public void run() throws Exception {
                user.setName("Henry");
                log.info("Execute the operation, change the name of the employee to: " + user.getName());
                Integer a = 1/0;

            }

            // Support rollback operations after final failure
            @Override
            public void retryFailed(Throwable e) {
                user.setName("Jack");
                log.info("Retry failed, business data rolled back...") ;
                log.info("Username is called." + user.getName());
            }

            @Override
            public Object snapshot() {
                return user;
            }
        });
    }


    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    private class UserInfo {
        private String name;
    }


}

4. Implementation results

1. POM is introduced. Note: the version number should be consistent with XXL job admin.

       <dependency>
            <groupId>com.xuxueli</groupId>
            <artifactId>xxl-job-core</artifactId>
            <version>1.9.2-SNAPSHOT</version>
        </dependency>

2、application.yml

xxl:
  job:
    admin:
      addresses: http://47.102.168.36:9500
    executor:
      log-path: ${catalina.home:/usr}/logs/job
      log-retention-days: 30
      ip: 127.0.0.1
      port: 10001

3. Configure executor bean

import com.xxl.job.core.executor.XxlJobExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


/**
 * xxl-job config
 */
@Configuration
public class XxlJobConfig {
    private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);

    @Value("${xxl.job.admin.addresses}")
    private String adminAddresses;

    @Value("${spring.application.name}")
    private String appName;

    @Value("${xxl.job.executor.ip}")
    private String ip;

    @Value("${xxl.job.executor.port}")
    private Integer port;

    @Value("${xxl.job.executor.log-path}")
    private String logPath;

    @Value("${xxl.job.executor.log-retention-days}")
    private int logRetentionDays;


    @Bean(initMethod = "start", destroyMethod = "destroy")
    public XxlJobExecutor xxlJobExecutor() {
        logger.info(">>>>>>>>>>> xxl-job config init.");
        XxlJobExecutor xxlJobExecutor = new XxlJobExecutor();
        xxlJobExecutor.setAdminAddresses(adminAddresses);
        xxlJobExecutor.setAppName(appName);
        xxlJobExecutor.setIp(ip);
        xxlJobExecutor.setPort(port);
        xxlJobExecutor.setPort(-1);
        xxlJobExecutor.setAccessToken(null);
        xxlJobExecutor.setLogPath(logPath);
        xxlJobExecutor.setLogRetentionDays(logRetentionDays);
        return xxlJobExecutor;
    }
}

4. Task code

import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.IJobHandler;
import com.xxl.job.core.handler.annotation.JobHandler;
import com.xxl.job.core.log.XxlJobLogger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * JOB
 */
@JobHandler(value = "iFlyTenantJob")
@Component
public class IFlyTenantJob extends IJobHandler {

    private static final Logger log = LoggerFactory.getLogger(IFlyTenantJob.class);

    @Override
    public ReturnT<String> execute(String s) {
        XxlJobLogger.log("The mission begins...") ;
        log.info("Task started...");

        iFlyAiService.queryTenants();
        try {
            Thread.sleep(6000);
            //int a = 1/0;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        XxlJobLogger.log("End of mission...");
        log.info("End of mission...");
        return ReturnT.SUCCESS;

    }


}

1. Why

The application class is directly placed in the default package, such as Src/main/Java directory.

2. Solutions

It should be placed in the package path created under Src/main/Java. The details are as follows