ホームページ >Java >&#&チュートリアル >SpringBoot が RabbitMq カスタム メッセージ リスニング コンテナを統合してメッセージ バッチ処理を実装する方法
RabbitMQ は、一般的に使用されるメッセージ キューです。Spring Boot はそれを深く統合しており、送信と送信を迅速に実現できます。メッセージの受信。 RabbitMQ では、メッセージの送受信は非同期であるため、メッセージの到着をリッスンするリスナーが必要です。 Spring Boot はデフォルトのリスナー コンテナを提供しますが、データをバッチで取得するなどの特別なニーズを満たすためにリスナー コンテナをカスタマイズする必要がある場合があります。
今回はSpring Bootを利用してRabbitMQを統合し、リスナーコンテナをカスタマイズしてデータを一括取得する機能を実現します。
前提条件:
開始する前に、次の条件を満たしている必要があります:
RabbitMQ サーバーがインストールされ、起動していること。
環境の準備: 開始する前に、次の環境を準備する必要があります:
<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 サイトの他の関連記事を参照してください。