Plusieurs méthodes de programmation asynchrone, combien en connaissez-vous ?

L'exécution asynchrone n'est pas étrangère aux développeurs. Dans le processus de développement réel, l'exécution asynchrone est souvent utilisée dans de nombreux scénarios. Par rapport à l'exécution synchrone, l'exécution asynchrone peut considérablement réduire le temps de liaison des requêtes.

Par exemple : 「发送短信、邮件、异步更新等」, ce sont des scénarios typiques qui peuvent être mis en œuvre de manière asynchrone.

8 façons d'implémenter l'asynchrone

1, Thread

2, Future

3, framework asynchrone CompletableFuture

4, annotation Spring @Async

5, événement Spring ApplicationEvent

6 , message queue

7, framework asynchrone tiers, tel que ThreadUtil

8 de Hutool, Guava asynchronous

Qu'est-ce que l'asynchrone ?

Tout d'abord, examinons un scénario courant dans lequel les utilisateurs passent des commandes :

Scénario commercial

Qu'est-ce que l'asynchrone ?

Dans l'opération de synchronisation, lorsque nous exécutons envoyer un message texte, nous devons attendre que cette méthode soit complètement exécutée avant de pouvoir effectuer l'opération points cadeaux Si l'action points cadeaux prend du temps. pour s'exécuter, l'envoi d'un message texte prendra beaucoup de temps. Attendez, il s'agit d'un scénario de synchronisation typique.

En fait, il n'y a aucune dépendance entre l'envoi de SMS et l'attribution de points. Grâce à l'asynchrone, nous pouvons réaliser que les deux opérations d'attribution de points et d'envoi de SMS peuvent être effectuées en même temps, comme :

Asynchrone

C'est ce qu'on appelle asynchrone. N'est-ce pas très simple ? Parlons de plusieurs façons d'implémenter l'asynchrone.

Implémentation spécifique asynchrone

1. Thread asynchrone

public class AsyncThread extends Thread {

    @Override
    public void run() {
        System.out.println("Current thread name:" + Thread.currentThread().getName() + " Send email success!");
    }

    public static void main(String[] args) {
        AsyncThread asyncThread = new AsyncThread();
        asyncThread.start();
    }
}

Bien sûr, si un thread Thread est créé à chaque fois, il est fréquemment créé et détruit, ce qui gaspille des ressources système, nous pouvons utilisez un pool de threads :

private ExecutorService executorService = Executors.newCachedThreadPool();

public void fun() {
    executorService.submit(new Runnable() {
        @Override
        public void run() {
            log.info("执行业务逻辑...");
        }
    });
}

Vous pouvez encapsuler la logique métier dans Runnable ou Callable et la transmettre au pool de threads pour exécution.

2. Future asynchrone

@Slf4j
public class FutureManager {


    public String execute() throws Exception {


        ExecutorService executor = Executors.newFixedThreadPool(1);
        Future<String> future = executor.submit(new Callable<String>() {
            @Override
            public String call() throws Exception {


                System.out.println(" --- task start --- ");
                Thread.sleep(3000);
                System.out.println(" --- task finish ---");
                return "this is future execute final result!!!";
            }
        });


        //这里需要返回值时会阻塞主线程
        String result = future.get();
        log.info("Future get result: {}", result);
        return result;
    }


    @SneakyThrows
    public static void main(String[] args) {
        FutureManager manager = new FutureManager();
        manager.execute();
    }
}

Résultat de sortie :

--- task start --- 
 --- task finish ---
 Future get result: this is future execute final result!!!

(1) Les lacunes de Future

Les lacunes de Future incluent les points suivants :

Impossible de recevoir passivement le calcul des tâches asynchrones Résultat : Bien que nous puissions soumettre activement des tâches asynchrones aux threads du pool de threads pour exécution, une fois l'exécution de la tâche asynchrone terminée, le thread principal ne peut pas être informé si la tâche est terminée ou non. via la méthode get .

Les pièces futures sont isolées les unes des autres : Parfois, après l'exécution d'une tâche asynchrone de longue durée, vous souhaitez utiliser le résultat renvoyé par celle-ci pour effectuer d'autres opérations. L'opération sera également une tâche asynchrone, et la relation entre. les deux relations nécessitent que les développeurs de programmes les lient et les attribuent manuellement. Les Futures ne peuvent pas former un flux de tâches (pipeline). Chaque Future est isolé l'un de l'autre, CompleteableFuture peut combiner plusieurs Futures connectés en série pour former un flux de tâches. .

Futrue ne dispose pas d'un bon mécanisme de gestion des erreurs : Jusqu'à présent, si une exception se produit lors de l'exécution d'une tâche asynchrone, l'appelant ne peut pas la détecter passivement. Il doit capturer l'exception de la méthode get pour savoir si la tâche est exécutée. une tâche asynchrone est exécutée. Une erreur s'est produite et un jugement et un traitement supplémentaires sont nécessaires.

3. CompletableFuture implémente l'asynchrone

public class CompletableFutureCompose {
    /**
     * thenAccept子任务和父任务公用同一个线程
     */
    @SneakyThrows
    public static void thenRunAsync() {
        CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " cf1 do something....");
            return 1;
        });
        CompletableFuture<Void> cf2 = cf1.thenRunAsync(() -> {
            System.out.println(Thread.currentThread() + " cf2 do something...");
        });
        //等待任务1执行完成
        System.out.println("cf1结果->" + cf1.get());
        //等待任务2执行完成
        System.out.println("cf2结果->" + cf2.get());
    }

    public static void main(String[] args) {
        thenRunAsync();
    }
}

Nous n'avons pas besoin d'utiliser explicitement ExecutorService. CompletableFuture utilise ForkJoinPool en interne pour gérer les tâches asynchrones. Si nous souhaitons personnaliser notre propre pool de threads asynchrones dans certains scénarios commerciaux, c'est également possible. .

4. @Async asynchrone de Spring

(1) Pool de threads asynchrones personnalisé

/**
 * 线程池参数配置，多个线程池实现线程池隔离，@Async注解，默认使用系统自定义线程池，可在项目中设置多个线程池，在异步调用的时候，指明需要调用的线程池名称，比如：@Async("taskName")
@EnableAsync
@Configuration
public class TaskPoolConfig {
    /**
     * 自定义线程池
     *
     **/
    @Bean("taskExecutor")
    public Executor taskExecutor() {
        //返回可用处理器的Java虚拟机的数量 12
        int i = Runtime.getRuntime().availableProcessors();
        System.out.println("系统最大线程数  ：" + i);
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //核心线程池大小
        executor.setCorePoolSize(16);
        //最大线程数
        executor.setMaxPoolSize(20);
        //配置队列容量，默认值为Integer.MAX_VALUE
        executor.setQueueCapacity(99999);
        //活跃时间
        executor.setKeepAliveSeconds(60);
        //线程名字前缀
        executor.setThreadNamePrefix("asyncServiceExecutor -");
        //设置此执行程序应该在关闭时阻止的最大秒数，以便在容器的其余部分继续关闭之前等待剩余的任务完成他们的执行
        executor.setAwaitTerminationSeconds(60);
        //等待所有的任务结束后再关闭线程池
        executor.setWaitForTasksToCompleteOnShutdown(true);
        return executor;
    }
}

(2) AsyncService

public interface AsyncService {

    MessageResult sendSms(String callPrefix, String mobile, String actionType, String content);

    MessageResult sendEmail(String email, String subject, String content);
}

@Slf4j
@Service
public class AsyncServiceImpl implements AsyncService {
    @Autowired
    private IMessageHandler mesageHandler;

    @Override
    @Async("taskExecutor")
    public MessageResult sendSms(String callPrefix, String mobile, String actionType, String content) {
        try {


            Thread.sleep(1000);
            mesageHandler.sendSms(callPrefix, mobile, actionType, content);


        } catch (Exception e) {
            log.error("发送短信异常 -> ", e)
        }
    }
    
    @Override
    @Async("taskExecutor")
    public sendEmail(String email, String subject, String content) {
        try {

            Thread.sleep(1000);
            mesageHandler.sendsendEmail(email, subject, content);

        } catch (Exception e) {
            log.error("发送email异常 -> ", e)
        }
    }
}

Dans les projets réels, utilisez @Async pour appeler le pool de threads, les méthodes recommandées et autres. Il s'agit du mode d'utilisation d'un pool de threads personnalisé. Il n'est pas recommandé d'utiliser directement @Async pour implémenter directement l'asynchrone.

5. L'événement Spring ApplicationEvent est asynchrone

（1）定义事件

public class AsyncSendEmailEvent extends ApplicationEvent {
    /**
     * 邮箱
     **/
    private String email;
   /**
     * 主题
     **/
    private String subject;
    /**
     * 内容
     **/
    private String content;
  
    /**
     * 接收者
     **/
    private String targetUserId;

}

（2）定义事件处理器

@Slf4j
@Component
public class AsyncSendEmailEventHandler implements ApplicationListener<AsyncSendEmailEvent> {

    @Autowired
    private IMessageHandler mesageHandler;
    
    @Async("taskExecutor")
    @Override
    public void onApplicationEvent(AsyncSendEmailEvent event) {
        if (event == null) {
            return;
        }

        String email = event.getEmail();
        String subject = event.getSubject();
        String content = event.getContent();
        String targetUserId = event.getTargetUserId();
        mesageHandler.sendsendEmailSms(email, subject, content, targerUserId);
      }
}

另外，可能有些时候采用ApplicationEvent实现异步的使用，当程序出现异常错误的时候，需要考虑补偿机制，那么这时候可以结合Spring Retry重试来帮助我们避免这种异常造成数据不一致问题。

6、消息队列

（1）回调事件消息生产者

@Slf4j
@Component
public class CallbackProducer {

    @Autowired
    AmqpTemplate amqpTemplate;

    public void sendCallbackMessage(CallbackDTO allbackDTO, final long delayTimes) {

        log.info("生产者发送消息，callbackDTO，{}", callbackDTO);

        amqpTemplate.convertAndSend(CallbackQueueEnum.QUEUE_GENSEE_CALLBACK.getExchange(), CallbackQueueEnum.QUEUE_GENSEE_CALLBACK.getRoutingKey(), JsonMapper.getInstance().toJson(genseeCallbackDTO), new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //给消息设置延迟毫秒值，通过给消息设置x-delay头来设置消息从交换机发送到队列的延迟时间
                message.getMessageProperties().setHeader("x-delay", delayTimes);
                message.getMessageProperties().setCorrelationId(callbackDTO.getSdkId());
                return message;
            }
        });
    }
}

（2）回调事件消息消费者

@Slf4j
@Component
@RabbitListener(queues = "message.callback", containerFactory = "rabbitListenerContainerFactory")
public class CallbackConsumer {

    @Autowired
    private IGlobalUserService globalUserService;

    @RabbitHandler
    public void handle(String json, Channel channel, @Headers Map<String, Object> map) throws Exception {

        if (map.get("error") != null) {
            //否认消息
            channel.basicNack((Long) map.get(AmqpHeaders.DELIVERY_TAG), false, true);
            return;
        }

        try {
        
            CallbackDTO callbackDTO = JsonMapper.getInstance().fromJson(json, CallbackDTO.class);
            //执行业务逻辑
            globalUserService.execute(callbackDTO);
            //消息消息成功手动确认，对应消息确认模式acknowledge-mode: manual
            channel.basicAck((Long) map.get(AmqpHeaders.DELIVERY_TAG), false);

        } catch (Exception e) {
            log.error("回调失败 -> {}", e);
        }
    }
}

7、ThreadUtil异步工具类

@Slf4j
public class ThreadUtils {

    public static void main(String[] args) {
        for (int i = 0; i < 3; i++) {
            ThreadUtil.execAsync(() -> {
                ThreadLocalRandom threadLocalRandom = ThreadLocalRandom.current();
                int number = threadLocalRandom.nextInt(20) + 1;
                System.out.println(number);
            });
            log.info("当前第：" + i + "个线程");
        }

        log.info("task finish!");
    }
}

8、Guava异步

Guava的ListenableFuture顾名思义就是可以监听的Future，是对java原生Future的扩展增强。我们知道Future表示一个异步计算任务，当任务完成时可以得到计算结果。如果我们希望一旦计算完成就拿到结果展示给用户或者做另外的计算，就必须使用另一个线程不断的查询计算状态。这样做，代码复杂，而且效率低下。使用「Guava ListenableFuture」可以帮我们检测Future是否完成了，不需要再通过get()方法苦苦等待异步的计算结果，如果完成就自动调用回调函数，这样可以减少并发程序的复杂度。

ListenableFuture是一个接口，它从jdk的Future接口继承，添加了void addListener(Runnable listener, Executor executor)方法。

我们看下如何使用ListenableFuture。首先需要定义ListenableFuture的实例：

ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());
        final ListenableFuture<Integer> listenableFuture = executorService.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                log.info("callable execute...")
                TimeUnit.SECONDS.sleep(1);
                return 1;
            }
        });

首先，通过MoreExecutors类的静态方法listeningDecorator方法初始化一个ListeningExecutorService的方法，然后，使用此实例的submit方法即可初始化ListenableFuture对象。

ListenableFuture要做的工作，在Callable接口的实现类中定义，这里只是休眠了1秒钟然后返回一个数字1，有了ListenableFuture实例，可以执行此Future并执行Future完成之后的回调函数。

Futures.addCallback(listenableFuture, new FutureCallback<Integer>() {
    @Override
    public void onSuccess(Integer result) {
        //成功执行...
        System.out.println("Get listenable future&#39;s result with callback " + result);
    }

    @Override
    public void onFailure(Throwable t) {
        //异常情况处理...
        t.printStackTrace();
    }
});

那么，以上就是本期介绍的实现异步的8种方式了。

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!