ホームページ >Java >&#&チュートリアル >SpringBoot が RabbitMq カスタム メッセージ リスニング コンテナを統合してメッセージ バッチ処理を実装する方法

SpringBoot が RabbitMq カスタム メッセージ リスニング コンテナを統合してメッセージ バッチ処理を実装する方法

WBOY
WBOY転載
2023-05-13 08:52:11995ブラウズ

SpringBoot は、RabbitMq カスタム メッセージ リスニング コンテナを統合して、メッセージ バッチ処理を実装します

はじめに

RabbitMQ は、一般的に使用されるメッセージ キューです。Spring Boot はそれを深く統合しており、送信と送信を迅速に実現できます。メッセージの受信。 RabbitMQ では、メッセージの送受信は非同期であるため、メッセージの到着をリッスンするリスナーが必要です。 Spring Boot はデフォルトのリスナー コンテナを提供しますが、データをバッチで取得するなどの特別なニーズを満たすためにリスナー コンテナをカスタマイズする必要がある場合があります。

今回はSpring Bootを利用してRabbitMQを統合し、リスナーコンテナをカスタマイズしてデータを一括取得する機能を実現します。
前提条件:
開始する前に、次の条件を満たしている必要があります:

  • RabbitMQ サーバーがインストールされ、起動していること。

  • #使用するキューが作成されました。

  • Spring Boot と RabbitMQ の基本的な知識はすでに理解しています。

環境の準備: 開始する前に、次の環境を準備する必要があります:

  • JDK 1.8 または上記のバージョン

  • Spring Boot 2.5.0以降のバージョン

  • RabbitMQ 3.8.0以降のバージョン

依存関係の追加

まず、pom.xml ファイルに次の依存関係を追加します。

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

構成ファイル

次に、次の構成を application.properties に追加します。ファイル:

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/

# 队列名称
spring.rabbitmq.listener.simple.queue-name=myQueue

# 最大并发消费者数量
spring.rabbitmq.listener.simple.concurrency=5

# 最小数量
spring.rabbitmq.listener.simple.min-concurrency=1

# 最大数量
spring.rabbitmq.listener.simple.max-concurrency=10

# 批量处理消息的大小
spring.rabbitmq.listener.simple.batch-size=50

または

spring:
  rabbitmq:
    host: localhost
    listener:
      simple:
        batch-size: 50
        concurrency: 5
        max-concurrency: 10
        min-concurrency: 1
        queue-name: myQueue
    password: guest
    port: 5672
    username: guest
    virtual-host: /

リスナーの作成

次に、キューから受信したメッセージを処理するためにリスナー クラスを作成する必要があります。以下は簡単な例です:

@Component
public class MyListener {
    
    @RabbitListener(queues = "myQueue", containerFactory = "myFactory")
    public void handleMessage(List<MyMessage> messages, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag)
            throws IOException {
        try {
            // 处理消息
            System.out.println("Received " + messages.size() + " messages");
            for (Message message : messages) {
           		// 处理消息
            	System.out.println("Received message: " + new String(message.getBody()));
        	}
        	channel.basicAck(messages.get(messages.size() - 1).getMessageProperties().getDeliveryTag(), true);
        } finally {
            // 手动确认消息
            channel.basicAck(deliveryTag, true);
        }
    }
}

上記のコードでは、@RabbitListener アノテーションを使用してリッスンするキューの名前を指定し、また、リッスンするコンテナーを作成するための myFactory ファクトリの使用も指定します。 。このリスナーでは、受信したメッセージを単に出力します。

SimpleRabbitListenerContainerFactory の作成

次に、リスニング コンテナーの動作をカスタマイズできるように、SimpleRabbitListenerContainerFactory ファクトリを作成する必要があります。簡単な例を次に示します。

@Configuration
public class RabbitMQConfig {

//    @Bean
//    public SimpleRabbitListenerContainerFactory myFactory(ConnectionFactory connectionFactory) {
//        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
//        factory.setConnectionFactory(connectionFactory);
//        factory.setConcurrentConsumers(1);
//        factory.setMaxConcurrentConsumers(10);
//        factory.setBatchListener(true);
//        factory.setBatchSize(50);
//        return factory;
//    }

	@Bean
    public SimpleRabbitListenerContainerFactory myFactory(
            ConnectionFactory connectionFactory,
            PlatformTransactionManager transactionManager,
            MessageConverter messageConverter) {
        
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        
        // 并发消费者数,默认为 1
        factory.setConcurrentConsumers(5);
        
        // 最大并发消费者数,默认为 1
        factory.setMaxConcurrentConsumers(10);
        
        // 拒绝未确认的消息并重新将它们放回队列,默认为 true
        factory.setDefaultRequeueRejected(false);
        
        // 容器启动时是否自动启动,默认为 true
        factory.setAutoStartup(true);
        
        // 消息确认模式,默认为 AUTO
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        
        // 每个消费者在一次请求中预获取的消息数,默认为 1
        factory.setPrefetchCount(5);
        
        // 从队列中接收消息的超时时间,默认为 0,表示没有超时限制
        factory.setReceiveTimeout(1000);
        
        // 与容器一起使用的事务管理器。默认情况下,容器不会使用事务
        factory.setTransactionManager(transactionManager);
        
        // 消息转换器,用于将接收到的消息转换为 Java 对象或将 Java 对象转换为消息
        factory.setMessageConverter(messageConverter);
        
        // 用于异步消息处理的线程池。默认情况下,容器使用一个简单的 SimpleAsyncTaskExecutor
        factory.setTaskExecutor(new SimpleAsyncTaskExecutor());
        
        // 在关闭容器时等待活动线程终止的时间,默认为 5000 毫秒
        factory.setShutdownTimeout(10000);
        
        // 重试失败的消息之前等待的时间,默认为 5000 毫秒
        factory.setRecoveryInterval(5000);
        
        // 如果消息处理器尝试监听不存在的队列,是否抛出异常。默认为 true
        factory.setMissingQueuesFatal(false);
        
        // 监听器容器连接工厂
        factory.setConnectionFactory(connectionFactory);

        return factory;
    }
}

これらのプロパティのほとんどはオプションであり、必要に応じて設定できます。アプリケーションのニーズに応じて、これらのプロパティを自由に調整して、アプリケーションのパフォーマンスと信頼性を向上させます。

メッセージの送信

最後に、いくつかのメッセージをキューに送信するための簡単なメッセージ送信コードを作成できます。簡単な例を次に示します:

@Component
public class MySender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessage(String message) {
        for (int i = 0; i < 100; i++) {
            rabbitTemplate.convertAndSend("myQueue", "message:" + i);
        }
    }
}

以上がSpringBoot が RabbitMq カスタム メッセージ リスニング コンテナを統合してメッセージ バッチ処理を実装する方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。

声明:
この記事はyisu.comで複製されています。侵害がある場合は、admin@php.cn までご連絡ください。