Введение в проект реакторного автобуса

1. Обзор

В этой быстрой статье мы представим реактор-автобус путем создания реального сценария для реактивного приложения, управляемого событиями.

2. Основы проекта реактора

2.1. Почему реактор?

Современным приложениям необходимо обрабатывать огромное количество одновременных запросов и обрабатывать значительный объем данных. Стандартный блокирующий код уже не достаточен для выполнения этих требований.

Реактивный дизайн шаблон является архитектурный подход, основанный на событиях, для асинхронной обработки большого объема одновременных запросов на обслуживание из одного или нескольких обработчиков услуг.

Проект реактора основан на этой модели и имеет четкую и амбициозную цель создания не блокировали, реактивные приложения на JVM .

2.2. Примеры сценариев

Прежде чем мы начат, вот несколько интересных сценариев, где использование реактивного архитектурного стиля имело бы смысл, просто чтобы получить представление о том, где мы могли бы применить его:

Услуги уведомлений для такой крупной торговой онлайн-платформы, как Amazon
Огромные услуги по обработке транзакций для банковского сектора
Акции торговых предприятий, где цены акций меняются одновременно

3. Maven зависимостей

Давайте начнем использовать проект реактора автобус, добавив следующую зависимость в наш пом.xml:


    io.projectreactor
    reactor-bus
    2.0.8.RELEASE

Мы можем проверить последнюю версию реактор-автобус в Мавен Центральный .

4. Создание демо-приложения

Чтобы лучше понять преимущества подхода, основанного на реакторах, рассмотрим практический пример.

Мы построим простое приложение, ответственное за отправку уведомлений пользователям платформы интернет-магазинов. Например, если пользователь размещает новый заказ, приложение отправляет подтверждение заказа по электронной почте или SMS.

Типичная синхронная реализация, естественно, будет ограничена пропускной способностью службы электронной почты или SMS. Таким образом, скачки трафика, такие как праздники, как правило, будет проблематичным.

С помощью реактивного подхода мы можем проектировать нашу систему, чтобы быть более гибкой и лучше адаптироваться к сбоям или тайм-аутам, которые могут возникнуть во внешних системах, таких как серверы шлюзов.

Давайте посмотрим на приложение – начиная с более традиционных аспектов и переходя к более реактивным конструкциям.

4.1. Простое POJO

Во-первых, давайте создадим класс POJO для представления данных уведомлений:

public class NotificationData {
	
    private long id;
    private String name;
    private String email;
    private String mobile;
    
    // getter and setter methods
}

4.2. Уровень обслуживания

Теперь определим простой уровень обслуживания:

public interface NotificationService {

    void initiateNotification(NotificationData notificationData) 
      throws InterruptedException;

}

И реализация, имитирующая долгосрочную операцию:

@Service
public class NotificationServiceimpl implements NotificationService {
	
    @Override
    public void initiateNotification(NotificationData notificationData) 
      throws InterruptedException {

      System.out.println("Notification service started for "
        + "Notification ID: " + notificationData.getId());
		
      Thread.sleep(5000);
		
      System.out.println("Notification service ended for "
        + "Notification ID: " + notificationData.getId());
    }
}

Обратите внимание, что, чтобы проиллюстрировать реальный сценарий отправки сообщений через SMS или адрес электронной почты, мы намеренно вводим пяти секунд задержки в инициироватьНотификация метод с Thread.sleep (5000).

Следовательно, когда поток попадает в службу, он будет заблокирован в течение пяти секунд.

4.3. Потребитель

Давайте теперь перейти к более реактивным аспектам нашего применения и реализации потребителя – который мы затем карту реактора событие автобуса:

@Service
public class NotificationConsumer implements 
  Consumer> {

    @Autowired
    private NotificationService notificationService;
	
    @Override
    public void accept(Event notificationDataEvent) {
        NotificationData notificationData = notificationDataEvent.getData();
        
        try {
            notificationService.initiateNotification(notificationData);
        } catch (InterruptedException e) {
            // ignore        
        }	
    }
}

Как видим, потребитель, которого мы создали, реализует Потребительский интерфейс. Основная логика заключается в принять метод.

Это аналогичный подход, с который мы можем встретиться в типичной реализации весеннего слушателя.

4.4. Контроллер

Наконец, теперь, когда мы можем потреблять события, давайте также генерировать их.

Мы собираемся сделать это в простой контроллер:

@Controller
public class NotificationController {

    @Autowired
    private EventBus eventBus;

    @GetMapping("/startNotification/{param}")
    public void startNotification(@PathVariable Integer param) {
        for (int i = 0; i < param; i++) {
            NotificationData data = new NotificationData();
            data.setId(i);

            eventBus.notify("notificationConsumer", Event.wrap(data));

            System.out.println(
              "Notification " + i + ": notification task submitted successfully");
        }
    }
}

Это вполне объяснимо – мы излучаем события через EventBus здесь.

Например, если клиент попадает в URL с параметрическим значением десять, то десять событий будут отправлены через автобус событий.

4.5. Java Config

Давайте теперь ставим все вместе и создать простое приложение Spring Boot.

Во-первых, мы должны настроить EventBus и Экологические фасоль:

@Configuration
public class Config {

    @Bean
    public Environment env() {
        return Environment.initializeIfEmpty().assignErrorJournal();
    }

    @Bean
    public EventBus createEventBus(Environment env) {
        return EventBus.create(env, Environment.THREAD_POOL);
    }
}

В нашем случае, мы мгновенно будем EventBus с пулом потоков по умолчанию, доступным в среде .

Кроме того, мы можем использовать индивидуальный Диспетчер пример:

EventBus evBus = EventBus.create(
  env, 
  Environment.newDispatcher(
    REACTOR_CAPACITY,REACTOR_CONSUMERS_COUNT,   
    DispatcherType.THREAD_POOL_EXECUTOR));

Теперь мы готовы создать основной код приложения:

import static reactor.bus.selector.Selectors.$;

@SpringBootApplication
public class NotificationApplication implements CommandLineRunner {

    @Autowired
    private EventBus eventBus;

    @Autowired
    private NotificationConsumer notificationConsumer;

    @Override
    public void run(String... args) throws Exception {
        eventBus.on($("notificationConsumer"), notificationConsumer);
    }

    public static void main(String[] args) {
        SpringApplication.run(NotificationApplication.class, args);
    }
}

В нашем запустить метод мы регистрируем уведомлениеКонсумер для срабатывания уведомления соответствует данному селектору .

Обратите внимание, как мы используем статический импорт $ атрибут для создания Селектор объект.

5. Проверить приложение

Давайте создадим тест, чтобы увидеть наши УведомлениеПрименение в действии:

@RunWith(SpringRunner.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
public class NotificationApplicationIntegrationTest {

    @LocalServerPort
    private int port;

    @Test
    public void givenAppStarted_whenNotificationTasksSubmitted_thenProcessed() {
        RestTemplate restTemplate = new RestTemplate();
        restTemplate.getForObject("http://localhost:" + port + "/startNotification/10", String.class);
    }
}

Как мы видим, как только запрос выполняется, все десять задачи получают представлены мгновенно, не создавая никаких . И после того, как представлены, события уведомления обрабатываются параллельно.

Notification 0: notification task submitted successfully
Notification 1: notification task submitted successfully
Notification 2: notification task submitted successfully
Notification 3: notification task submitted successfully
Notification 4: notification task submitted successfully
Notification 5: notification task submitted successfully
Notification 6: notification task submitted successfully
Notification 7: notification task submitted successfully
Notification 8: notification task submitted successfully
Notification 9: notification task submitted successfully
Notification service started for Notification ID: 1
Notification service started for Notification ID: 2
Notification service started for Notification ID: 3
Notification service started for Notification ID: 0
Notification service ended for Notification ID: 1
Notification service ended for Notification ID: 0
Notification service started for Notification ID: 4
Notification service ended for Notification ID: 3
Notification service ended for Notification ID: 2
Notification service started for Notification ID: 6
Notification service started for Notification ID: 5
Notification service started for Notification ID: 7
Notification service ended for Notification ID: 4
Notification service started for Notification ID: 8
Notification service ended for Notification ID: 6
Notification service ended for Notification ID: 5
Notification service started for Notification ID: 9
Notification service ended for Notification ID: 7
Notification service ended for Notification ID: 8
Notification service ended for Notification ID: 9

Важно иметь в виду, что в нашем сценарии нет необходимости обрабатывать эти события в каком-либо определенном порядке.

6. Заключение

В этом быстром учебнике, мы создали простое приложение, управляемое событиями, . Мы также видели, как начать писать более реактивный и не блокирующий код.

Тем не менее, этот сценарий просто царапает поверхность предмета и представляет собой лишь хорошую основу для начала экспериментов с реактивной парадигмой .

Как всегда, исходный код доступен более на GitHub .