Java系統(tǒng)中拆分同步和異步詳解

更新時間：2019年06月14日 10:18:16 作者：Petr Bouda

這篇文章主要給大家介紹了關于Java系統(tǒng)中拆分同步和異步的相關資料，文中通過示例代碼介紹的非常詳細，對大家學習或者使用Java具有一定的參考學習價值，需要的朋友們下面來一起學習學習吧

前言

很多開發(fā)人員說,將應用程序切換到異步處理很復雜。因為他們有一個天然需要同步通信的Web應用程序。在這篇文章中，我想介紹一種方法來達到異步通信的目的：使用一些眾所周知的庫和工具來設計他們的系統(tǒng)。下面的例子是用Java編寫的，但我相信它更多的是基本原理，同一個應用程序可以用任何語言來重新寫。

所需的工具和庫:

Spring Boot
RabbitMQ

1.Web應用程序

一個用Spring MVC編寫的Web應用程序并運行在Tomcat上。它所做的只是將一個字符串發(fā)送到一個隊列中 (異步通信的開始) 并等待另一個隊列中的消息作為HTTP響應發(fā)送回來。

首先，我們需要定義幾個依賴項，然后等待Spring Boot執(zhí)行所有必要的自動配置。

<dependencies>
 <dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-web</artifactId>
 </dependency>
 <dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
 </dependency>
 <dependency>
  <groupId>com.thedeanda</groupId>
  <artifactId>lorem</artifactId>
 </dependency>
</dependencies>
@SpringBootApplication
public class BlockingApplication {
 public static void main(String[] args) {
  SpringApplication.run(BlockingApplication.class, args);
 }
 @RestController
 public static class MessageController {
  private final RabbitTemplate rabbitTemplate;
  public MessageController(CachingConnectionFactory connectionFactory) {
   this.rabbitTemplate = new RabbitTemplate(connectionFactory);
  }
  @GetMapping("invoke")
  public String sendMessage() {
   Message response = rabbitTemplate.sendAndReceive("uppercase", null, request());
   return new String(response.getBody());
  }
  private static Message request() {
   Lorem LOREM = LoremIpsum.getInstance();
   String name = LOREM.getFirstName() + " " + LOREM.getLastName();
   return new Message(name.getBytes(), new MessageProperties());
  }
 }
 @Bean
 public CachingConnectionFactory connectionFactory() {
  CachingConnectionFactory factory = new CachingConnectionFactory();
  factory.setAddresses("localhost:5672");
  factory.setUsername("admin");
  factory.setPassword("admin");
  return factory;
 }
}

2.消費端應用程序

第二個應用程序僅僅是一個等待消息的RabbitMQ的消費端,將拿到的字符串轉換為大寫,然后將此結果發(fā)送到輸出隊列中。

<dependencies>
 <dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
 </dependency>
</dependencies>
@SpringBootApplication
public class ServiceApplication {
 public static void main(String[] args) {
  SpringApplication.run(ServiceApplication.class, args);
 }
 public static class MessageListener {
  public String handleMessage(byte[] message) {
   Random rand = new Random();
   // Obtain a number between [0 - 49] + 50 = [50 - 99]
   int n = rand.nextInt(50) + 50;
   String content = new String(message);
   try {
    Thread.sleep(n);
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
   return content.toUpperCase();
  }
 }
 @Bean
 public CachingConnectionFactory connectionFactory() {
  CachingConnectionFactory factory = new CachingConnectionFactory();
  factory.setAddresses("localhost:5672");
  factory.setUsername("admin");
  factory.setPassword("admin");
  return factory;
 }
 @Bean
 public SimpleMessageListenerContainer serviceListenerContainer() {
  SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
  container.setConnectionFactory(connectionFactory());
  container.setConcurrentConsumers(20);
  container.setMaxConcurrentConsumers(40);
  container.setQueueNames("uppercase_messages");
  container.setMessageListener(new MessageListenerAdapter(new MessageListener()));
  return container;
 }
}

3.底層如何執(zhí)行的?

程序啟動并首次調用sendMessage()方法后，我們可以看到Spring AMQP支持自動創(chuàng)建了一個新的回復隊列并等待來自我們的服務應用程序的響應。

2019-05-12 17:23:21.451 INFO 4574 --- [nio-8080-exec-1] .l.DirectReplyToMessageListenerContainer : Container initialized for queues: [amq.rabbitmq.reply-to]
2019-05-12 17:23:21.457 INFO 4574 --- [nio-8080-exec-1] .l.DirectReplyToMessageListenerContainer : SimpleConsumer [queue=amq.rabbitmq.reply-to, consumerTag=amq.ctag-VF-iqD9rLEuljIBstbCI1A identity=10e58093] started

如果我們在消費端應用程序中查看消息，我們可以看到Spring自動傳播有關回復隊列的信息以及**相關ID，**用于將其傳遞回Web應用程序以便能夠將請求和響應配對在一起。

這就是發(fā)生魔術的地方。當然，如果您想使其更復雜，您可以在協(xié)作中包含更多服務，然后將Web應用程序的最終響應放入與自動生成的隊列不同的隊列中，該隊列只具有正確的關聯(lián)ID。另外，不要忘記設置合理的超時。

這個解決方案還有一個很大的缺點 - 應用程序吞吐量。我故意這樣做，以便我可以跟進這篇文章，進一步深入調查AsyncProfiler！但是目前，我們使用Tomcat作為主HTTP服務器，默認為200個線程，這意味著我們的應用程序無法同時處理200多條消息，因為我們的服務器線程正在等待RabbitMQ 回復隊列的響應，直到有消息進入或發(fā)生超時。

感謝您閱讀本文，敬請關注后續(xù)內容！如果您想自己嘗試一下，請查看我的GitHub存儲庫。

原文鏈接:dzone.com/articles/ho…

作者：Petr Bouda

譯者：KeepGoingPawn

總結

以上就是這篇文章的全部內容了，希望本文的內容對大家的學習或者工作具有一定的參考學習價值，謝謝大家對腳本之家的支持。

您可能感興趣的文章: