Организация обработки асинхронных событий с Spring Events и Spring AMQP

Привет, Хабр!

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

Spring Events

Spring Events – это встроенный механизм для публикации и обработки событий внутри приложения. Spring Events основаны на паттерне проектирования «издатель-подписчик«.

Для работы с событиями потребуется три основных компонента:

1. Класс события: описывает само событие.

2. Издатель события: компонент, который генерирует и публикует события.

3. Подписчик на событие: компонент, который обрабатывает опубликованные события.

Начнем с создания класса события. Этот класс должен наследоваться от ApplicationEvent:

public class CustomSpringEvent extends ApplicationEvent {     private String message;      public CustomSpringEvent(Object source, String message) {         super(source);         this.message = message;     }      public String getMessage() {         return message;     } }

Для публикации события потребуется класс-издатель, который будет использовать ApplicationEventPublisher для публикации событий:

import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.stereotype.Component;  @Component public class CustomSpringEventPublisher implements ApplicationEventPublisherAware {      private ApplicationEventPublisher applicationEventPublisher;      @Override     public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {         this.applicationEventPublisher = applicationEventPublisher;     }      public void publishEvent(final String message) {         System.out.println("Publishing custom event. ");         CustomSpringEvent customSpringEvent = new CustomSpringEvent(this, message);         applicationEventPublisher.publishEvent(customSpringEvent);     } }

Теперь создадим компонент, который будет обрабатывать события. Для этого можно использовать аннотацию @EventListener:

import org.springframework.context.event.EventListener; import org.springframework.stereotype.Component;  @Component public class CustomSpringEventListener {      @EventListener     public void handleCustomSpringEvent(CustomSpringEvent event) {         System.out.println("Received spring custom event - " + event.getMessage());     } }

Чтобы события обрабатывались асинхронно нужно включить поддержку асинхронности в конфигурации Spring и пометить метод-обработчик аннотацией @Async.

Для этого необходимо добавить аннотацию @EnableAsync в конфигурационный класс:

import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.annotation.EnableAsync;  @Configuration @EnableAsync public class AsyncConfig { }

Теперь изменим слушатель событий, чтобы он обрабатывал события асинхронно:

import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Component; import org.springframework.context.event.EventListener;  @Component public class CustomSpringEventListener {      @Async     @EventListener     public void handleCustomSpringEvent(CustomSpringEvent event) {         System.out.println("Received spring custom event - " + event.getMessage());         // доп. логика обработки события     } }

Примеры использования

Кэширование

Предположим, есть приложение, которое кэширует данные. Когда данные изменяются, нужно обновить кэш. Можно юзать событие для уведомления всех компонентов о необходимости обновления кэша:

// событие изменения данных public class DataChangeEvent extends ApplicationEvent {     private String dataId;      public DataChangeEvent(Object source, String dataId) {         super(source);         this.dataId = dataId;     }      public String getDataId() {         return dataId;     } }  // публикация события изменения данных @Component public class DataService {     @Autowired     private ApplicationEventPublisher eventPublisher;      public void updateData(String dataId) {         // логика обновления данных         eventPublisher.publishEvent(new DataChangeEvent(this, dataId));     } }  // обработчик события изменения данных @Component public class CacheService {      @EventListener     @Async     public void handleDataChangeEvent(DataChangeEvent event) {         System.out.println("Updating cache for dataId: " + event.getDataId());         // логика обновления кэша     } }

Уведомление о завершении задач

Для этой задачи можно использовать событие для уведомления других компонентов о завершении длительных задач:

// событие завершения задачи public class TaskCompleteEvent extends ApplicationEvent {     private String taskId;      public TaskCompleteEvent(Object source, String taskId) {         super(source);         this.taskId = taskId;     }      public String getTaskId() {         return taskId;     } }  // публикация события завершения задачи @Component public class TaskService {     @Autowired     private ApplicationEventPublisher eventPublisher;      @Async     public void executeTask(String taskId) {         // логика выполнения задачи         eventPublisher.publishEvent(new TaskCompleteEvent(this, taskId));     } }  // обработчик события завершения задачи @Component public class NotificationService {      @EventListener     @Async     public void handleTaskCompleteEvent(TaskCompleteEvent event) {         System.out.println("Task completed: " + event.getTaskId());         // логика уведомления     } }

Организация обработки сообщений с помощью Spring AMQP

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

RabbitMQ — это брокер сообщений, реализующий AMQP. Он позволяет обмениваться сообщениями и управлять их очередями.

Для старта работ с Spring AMQP в проекте Spring Boot необходимо добавить соответствующую зависимость в файл pom.xml:

<dependency>     <groupId>org.springframework.boot</groupId>     <artifactId>spring-boot-starter-amqp</artifactId> </dependency>

Далее необходимо настроить параметры подключения к RabbitMQ в файле application.properties.

spring.rabbitmq.host=localhost spring.rabbitmq.port=5672 spring.rabbitmq.username=guest spring.rabbitmq.password=guest

Создание и отправка сообщений с помощью AmqpTemplate

Для отправки сообщений используется AmqpTemplate.

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

import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;  @Configuration public class RabbitConfig {      static final String queueName = "myQueue";      @Bean     Queue queue() {         return new Queue(queueName, false);     }      @Bean     RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {         return new RabbitTemplate(connectionFactory);     }      @Bean     SimpleMessageListenerContainer container(ConnectionFactory connectionFactory,             MessageListenerAdapter listenerAdapter) {         SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();         container.setConnectionFactory(connectionFactory);         container.setQueueNames(queueName);         container.setMessageListener(listenerAdapter);         return container;     }      @Bean     MessageListenerAdapter listenerAdapter(Receiver receiver) {         return new MessageListenerAdapter(receiver, "receiveMessage");     } }

Теперь создадим компонент, который будет отправлять сообщения:

import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component;  @Component public class Sender {      private final RabbitTemplate rabbitTemplate;      @Autowired     public Sender(RabbitTemplate rabbitTemplate) {         this.rabbitTemplate = rabbitTemplate;     }      public void send(String message) {         rabbitTemplate.convertAndSend(RabbitConfig.queueName, message);         System.out.println("Sent: " + message);     } }

Для обработки сообщений, поступающих в очередь, используется аннотация @RabbitListener.

Добавим компонент, который будет обрабатывать входящие сообщения:

import org.springframework.stereotype.Component;  @Component public class Receiver {      public void receiveMessage(String message) {         System.out.println("Received: " + message);     } }

С аннтоацией @RabbitListener можно декларативно указать методы, которые будут обрабатывать сообщения из очередей:

import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Service;  @Service public class MessageListener {      @RabbitListener(queues = RabbitConfig.queueName)     public void processMessage(String message) {         System.out.println("Received message: " + message);     } }

Пример

Полный пример с отправкой и получением сообщений:

import org.springframework.amqp.rabbit.annotation.EnableRabbit; import org.springframework.boot.CommandLineRunner; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean;  @SpringBootApplication @EnableRabbit public class SpringAmqpApplication {      public static void main(String[] args) {         SpringApplication.run(SpringAmqpApplication.class, args);     }      @Bean     CommandLineRunner runner(Sender sender) {         return args -> {             System.out.println("Sending message...");             sender.send("Hello, RabbitMQ!");         };     } }

Примеры использования двух инструментов вместе

Обработка заказов с локальными событиями и асинхронной передачей

В этом примере система обрабатывает заказы локально, публикует событие о новом заказе с помощью Spring Events и затем отправляет заказ на дальнейшую обработку в другую службу через RabbitMQ:

// событие заказа public class OrderCreatedEvent extends ApplicationEvent {     private String orderId;      public OrderCreatedEvent(Object source, String orderId) {         super(source);         this.orderId = orderId;     }      public String getOrderId() {         return orderId;     } }  // публикация события заказа @Component public class OrderService {     @Autowired     private ApplicationEventPublisher eventPublisher;      public void createOrder(String orderId) {         // логика создания заказа         System.out.println("Order created: " + orderId);         eventPublisher.publishEvent(new OrderCreatedEvent(this, orderId));     } }  // обработка локального события и отправка в RabbitMQ @Component public class OrderCreatedListener {     @Autowired     private AmqpTemplate amqpTemplate;      @EventListener     @Async     public void handleOrderCreatedEvent(OrderCreatedEvent event) {         System.out.println("Handling order created event for order: " + event.getOrderId());         // доп. логика обработки         amqpTemplate.convertAndSend("orderQueue", event.getOrderId());     } }  // конфигурация RabbitMQ @Configuration public class RabbitConfig {     @Bean     public Queue orderQueue() {         return new Queue("orderQueue", false);     }      @Bean     public AmqpTemplate amqpTemplate(ConnectionFactory connectionFactory) {         return new RabbitTemplate(connectionFactory);     } }

Обновление состояния задач и уведомление через RabbitMQ

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

// событие обновления задачи public class TaskStatusUpdateEvent extends ApplicationEvent {     private String taskId;     private String status;      public TaskStatusUpdateEvent(Object source, String taskId, String status) {         super(source);         this.taskId = taskId;         this.status = status;     }      public String getTaskId() {         return taskId;     }      public String getStatus() {         return status;     } }  // публикация события обновления задачи @Component public class TaskService {     @Autowired     private ApplicationEventPublisher eventPublisher;      public void updateTaskStatus(String taskId, String status) {         System.out.println("Updating task status: " + taskId + " to " + status);         eventPublisher.publishEvent(new TaskStatusUpdateEvent(this, taskId, status));     } }  // обработка события обновления задачи и отправка в RabbitMQ @Component public class TaskStatusUpdateListener {     @Autowired     private AmqpTemplate amqpTemplate;      @EventListener     @Async     public void handleTaskStatusUpdateEvent(TaskStatusUpdateEvent event) {         System.out.println("Sending task status update to RabbitMQ: " + event.getTaskId() + " - " + event.getStatus());         amqpTemplate.convertAndSend("taskStatusQueue", event.getTaskId() + " - " + event.getStatus());     } }  // конфигурация RabbitMQ @Configuration public class RabbitConfig {     @Bean     public Queue taskStatusQueue() {         return new Queue("taskStatusQueue", false);     }      @Bean     public AmqpTemplate amqpTemplate(ConnectionFactory connectionFactory) {         return new RabbitTemplate(connectionFactory);     } }

---

В заключение напоминаю об открытых уроках, которые пройдут в рамках курса «Разработчик на Spring Framework» в Otus:

• 1 июля: Тестирование Spring приложений. Интеграционные тесты с контекстом. Тестирование слоя репозиториев и сервисов. Запись по ссылке

• 16 июля: Тестирование Spring приложений. Интеграционные тесты контроллеров, интеграций с внешними API и безопасности. Запись по ссылке