From b196a5f290e497b52215ffe423ca5cec11d7a534 Mon Sep 17 00:00:00 2001 From: YunaiV <> Date: Tue, 10 Dec 2019 21:30:59 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=20rabbitmq=20=E7=A4=BA?= =?UTF-8?q?=E4=BE=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../pom.xml | 30 +++++++ .../lab04/rabbitmqdemo/Application.java | 15 ++++ .../rabbitmqdemo/config/RabbitConfig.java | 64 ++++++++++++++ .../rabbitmqdemo/consumer/Demo06Consumer.java | 29 +++++++ .../rabbitmqdemo/message/Demo06Message.java | 34 ++++++++ .../rabbitmqdemo/producer/Demo06Producer.java | 22 +++++ .../src/main/resources/application.yaml | 7 ++ .../lab04/rabbitmqdemo/package-info.java | 1 + .../producer/Demo06ProducerTest.java | 39 +++++++++ .../target/classes/application.yaml | 7 ++ .../pom.xml | 30 +++++++ .../lab04/rabbitmqdemo/Application.java | 15 ++++ .../rabbitmqdemo/config/RabbitConfig.java | 83 ++++++++++++++++++ .../rabbitmqdemo/consumer/Demo05Consumer.java | 29 +++++++ .../rabbitmqdemo/message/Demo05Message.java | 34 ++++++++ .../rabbitmqdemo/producer/Demo06Producer.java | 22 +++++ .../src/main/resources/application.yaml | 7 ++ .../lab04/rabbitmqdemo/package-info.java | 1 + .../producer/Demo05ProducerTest.java | 39 +++++++++ .../target/classes/application.yaml | 7 ++ lab-04/lab-04-rabbitmq-demo-batch/pom.xml | 30 +++++++ .../lab04/rabbitmqdemo/Application.java | 15 ++++ .../rabbitmqdemo/config/RabbitConfig.java | 70 +++++++++++++++ .../rabbitmqdemo/consumer/Demo05Consumer.java | 21 +++++ .../rabbitmqdemo/message/Demo05Message.java | 34 ++++++++ .../rabbitmqdemo/producer/Demo05Producer.java | 22 +++++ .../src/main/resources/application.yaml | 7 ++ .../lab04/rabbitmqdemo/package-info.java | 1 + .../producer/Demo05ProducerTest.java | 39 +++++++++ .../rabbitmqdemo/config/RabbitConfig.java | 85 +++++++++++++++++++ .../consumer/Demo03ConsumerA.java | 21 +++++ .../consumer/Demo03ConsumerB.java | 21 +++++ .../rabbitmqdemo/consumer/Demo04Consumer.java | 26 ++++++ .../rabbitmqdemo/message/Demo03Message.java | 33 +++++++ .../rabbitmqdemo/message/Demo04Message.java | 35 ++++++++ .../rabbitmqdemo/producer/Demo03Producer.java | 22 +++++ .../rabbitmqdemo/producer/Demo04Producer.java | 27 ++++++ .../producer/Demo03ProducerTest.java | 33 +++++++ .../producer/Demo04ProducerTest.java | 44 ++++++++++ lab-04/pom.xml | 3 + 40 files changed, 1104 insertions(+) create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume-02/pom.xml create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo06Consumer.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo06Message.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo06Producer.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/resources/application.yaml create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo06ProducerTest.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume-02/target/classes/application.yaml create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume/pom.xml create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo05Consumer.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo05Message.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo06Producer.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/resources/application.yaml create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo05ProducerTest.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch-consume/target/classes/application.yaml create mode 100644 lab-04/lab-04-rabbitmq-demo-batch/pom.xml create mode 100644 lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo05Consumer.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo05Message.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo05Producer.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch/src/main/resources/application.yaml create mode 100644 lab-04/lab-04-rabbitmq-demo-batch/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java create mode 100644 lab-04/lab-04-rabbitmq-demo-batch/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo05ProducerTest.java create mode 100644 lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo03ConsumerA.java create mode 100644 lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo03ConsumerB.java create mode 100644 lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo04Consumer.java create mode 100644 lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo03Message.java create mode 100644 lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo04Message.java create mode 100644 lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo03Producer.java create mode 100644 lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo04Producer.java create mode 100644 lab-04/lab-04-rabbitmq-demo/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo03ProducerTest.java create mode 100644 lab-04/lab-04-rabbitmq-demo/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo04ProducerTest.java diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume-02/pom.xml b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/pom.xml new file mode 100644 index 00000000..38eaf880 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/pom.xml @@ -0,0 +1,30 @@ + + + + org.springframework.boot + spring-boot-starter-parent + 2.2.1.RELEASE + + + 4.0.0 + + lab-04-rabbitmq-demo-batch-consume-02 + + + + + org.springframework.boot + spring-boot-starter-amqp + + + + + org.springframework.boot + spring-boot-starter-test + test + + + + diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java new file mode 100644 index 00000000..54968a6a --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java @@ -0,0 +1,15 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.scheduling.annotation.EnableAsync; + +@SpringBootApplication +@EnableAsync // 开启异步 +public class Application { + + public static void main(String[] args) { + SpringApplication.run(Application.class, args); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java new file mode 100644 index 00000000..12fc21a2 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java @@ -0,0 +1,64 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.config; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo06Message; +import org.springframework.amqp.core.Binding; +import org.springframework.amqp.core.BindingBuilder; +import org.springframework.amqp.core.DirectExchange; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration +public class RabbitConfig { + + /** + * Direct Exchange 示例的配置类 + */ + public static class DirectExchangeDemoConfiguration { + + // 创建 Queue + @Bean + public Queue demo06Queue() { + return new Queue(Demo06Message.QUEUE, // Queue 名字 + true, // durable: 是否持久化 + false, // exclusive: 是否排它 + false); // autoDelete: 是否自动删除 + } + + // 创建 Direct Exchange + @Bean + public DirectExchange demo06Exchange() { + return new DirectExchange(Demo06Message.EXCHANGE, + true, // durable: 是否持久化 + false); // exclusive: 是否排它 + } + + // 创建 Binding + // Exchange：Demo06Message.EXCHANGE + // Routing key：Demo06Message.ROUTING_KEY + // Queue：Demo06Message.QUEUE + @Bean + public Binding demo06Binding() { + return BindingBuilder.bind(demo06Queue()).to(demo06Exchange()).with(Demo06Message.ROUTING_KEY); + } + + } + + @Bean(name = "consumerBatchContainerFactory") + public SimpleRabbitListenerContainerFactory consumerBatchContainerFactory( + SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory) { + // 创建 SimpleRabbitListenerContainerFactory 对象 + SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); + configurer.configure(factory, connectionFactory); + // 额外添加批量消费的属性 + factory.setBatchListener(true); + factory.setBatchSize(10); + factory.setReceiveTimeout(10 * 1000L); + factory.setConsumerBatchEnabled(true); + return factory; + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo06Consumer.java b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo06Consumer.java new file mode 100644 index 00000000..91dc7cd3 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo06Consumer.java @@ -0,0 +1,29 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.consumer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo06Message; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.amqp.rabbit.annotation.RabbitHandler; +import org.springframework.amqp.rabbit.annotation.RabbitListener; +import org.springframework.stereotype.Component; + +import java.util.List; + +@Component +@RabbitListener(queues = Demo06Message.QUEUE, + containerFactory = "consumerBatchContainerFactory") +public class Demo06Consumer { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + @RabbitHandler + public void onMessage(List messages) { + logger.info("[onMessage][线程编号:{} 消息数量：{}]", Thread.currentThread().getId(), messages.size()); + } + +// @RabbitHandler(isDefault = true) +// public void onMessageX(List messages) { +// logger.info("[onMessage][线程编号:{} 消息数量：{}]", Thread.currentThread().getId(), messages.size()); +// } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo06Message.java b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo06Message.java new file mode 100644 index 00000000..208d2bc3 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo06Message.java @@ -0,0 +1,34 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.message; + +import java.io.Serializable; + +public class Demo06Message implements Serializable { + + public static final String QUEUE = "QUEUE_DEMO_06"; + + public static final String EXCHANGE = "EXCHANGE_DEMO_06"; + + public static final String ROUTING_KEY = "ROUTING_KEY_06"; + + /** + * 编号 + */ + private Integer id; + + public Demo06Message setId(Integer id) { + this.id = id; + return this; + } + + public Integer getId() { + return id; + } + + @Override + public String toString() { + return "Demo06Message{" + + "id=" + id + + '}'; + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo06Producer.java b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo06Producer.java new file mode 100644 index 00000000..332d498e --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo06Producer.java @@ -0,0 +1,22 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.producer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo06Message; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +@Component +public class Demo06Producer { + + @Autowired + private RabbitTemplate rabbitTemplate; + + public void syncSend(Integer id) { + // 创建 Demo06Message 消息 + Demo06Message message = new Demo06Message(); + message.setId(id); + // 同步发送消息 + rabbitTemplate.convertAndSend(Demo06Message.EXCHANGE, Demo06Message.ROUTING_KEY, message); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/resources/application.yaml b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/resources/application.yaml new file mode 100644 index 00000000..59cf3bfb --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/main/resources/application.yaml @@ -0,0 +1,7 @@ +spring: + # RabbitMQ 配置项，对应 RabbitProperties 配置类 + rabbitmq: + host: 127.0.0.1 # RabbitMQ 服务的地址 + port: 5672 # RabbitMQ 服务的端口 + username: guest # RabbitMQ 服务的账号 + password: guest # RabbitMQ 服务的密码 diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java new file mode 100644 index 00000000..766da5b0 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java @@ -0,0 +1 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo; diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo06ProducerTest.java b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo06ProducerTest.java new file mode 100644 index 00000000..2e80be64 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo06ProducerTest.java @@ -0,0 +1,39 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.producer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.Application; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.context.junit4.SpringRunner; + +import java.util.concurrent.CountDownLatch; + +@RunWith(SpringRunner.class) +@SpringBootTest(classes = Application.class) +public class Demo06ProducerTest { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + @Autowired + private Demo06Producer producer; + + @Test + public void testSyncSend() throws InterruptedException { + for (int i = 0; i < 3; i++) { + // 同步发送消息 + int id = (int) (System.currentTimeMillis() / 1000); + producer.syncSend(id); + + // 故意每条消息之间，隔离 10 秒 + logger.info("[testASyncSend][发送编号：[{}] 发送成功]", id); +// Thread.sleep(10 * 1000L); + } + + // 阻塞等待，保证消费 + new CountDownLatch(1).await(); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume-02/target/classes/application.yaml b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/target/classes/application.yaml new file mode 100644 index 00000000..59cf3bfb --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume-02/target/classes/application.yaml @@ -0,0 +1,7 @@ +spring: + # RabbitMQ 配置项，对应 RabbitProperties 配置类 + rabbitmq: + host: 127.0.0.1 # RabbitMQ 服务的地址 + port: 5672 # RabbitMQ 服务的端口 + username: guest # RabbitMQ 服务的账号 + password: guest # RabbitMQ 服务的密码 diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume/pom.xml b/lab-04/lab-04-rabbitmq-demo-batch-consume/pom.xml new file mode 100644 index 00000000..4d164a4a --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume/pom.xml @@ -0,0 +1,30 @@ + + + + org.springframework.boot + spring-boot-starter-parent + 2.2.1.RELEASE + + + 4.0.0 + + lab-04-rabbitmq-demo-batch-consume + + + + + org.springframework.boot + spring-boot-starter-amqp + + + + + org.springframework.boot + spring-boot-starter-test + test + + + + diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java b/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java new file mode 100644 index 00000000..54968a6a --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java @@ -0,0 +1,15 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.scheduling.annotation.EnableAsync; + +@SpringBootApplication +@EnableAsync // 开启异步 +public class Application { + + public static void main(String[] args) { + SpringApplication.run(Application.class, args); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java b/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java new file mode 100644 index 00000000..926b9231 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java @@ -0,0 +1,83 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.config; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo05Message; +import org.springframework.amqp.core.Binding; +import org.springframework.amqp.core.BindingBuilder; +import org.springframework.amqp.core.DirectExchange; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.batch.BatchingStrategy; +import org.springframework.amqp.rabbit.batch.SimpleBatchingStrategy; +import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate; +import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler; + +@Configuration +public class RabbitConfig { + + /** + * Direct Exchange 示例的配置类 + */ + public static class DirectExchangeDemoConfiguration { + + // 创建 Queue + @Bean + public Queue demo05Queue() { + return new Queue(Demo05Message.QUEUE, // Queue 名字 + true, // durable: 是否持久化 + false, // exclusive: 是否排它 + false); // autoDelete: 是否自动删除 + } + + // 创建 Direct Exchange + @Bean + public DirectExchange demo05Exchange() { + return new DirectExchange(Demo05Message.EXCHANGE, + true, // durable: 是否持久化 + false); // exclusive: 是否排它 + } + + // 创建 Binding + // Exchange：Demo05Message.EXCHANGE + // Routing key：Demo05Message.ROUTING_KEY + // Queue：Demo05Message.QUEUE + @Bean + public Binding demo05Binding() { + return BindingBuilder.bind(demo05Queue()).to(demo05Exchange()).with(Demo05Message.ROUTING_KEY); + } + + } + + @Bean + public BatchingRabbitTemplate batchRabbitTemplate(ConnectionFactory connectionFactory) { + // 创建 BatchingStrategy 对象，代表批量策略 + int batchSize = 16384; // 超过收集的消息数量的最大条数。 + int bufferLimit = 33554432; // 每次批量发送消息的最大内存 + int timeout = 30000; // 超过收集的时间的最大等待时长，单位：毫秒 + BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(batchSize, bufferLimit, timeout); + + // 创建 TaskScheduler 对象，用于实现超时发送的定时器 + TaskScheduler taskScheduler = new ConcurrentTaskScheduler(); + + // 创建 BatchingRabbitTemplate 对象 + BatchingRabbitTemplate batchTemplate = new BatchingRabbitTemplate(batchingStrategy, taskScheduler); + batchTemplate.setConnectionFactory(connectionFactory); + return batchTemplate; + } + + @Bean(name = "consumerBatchContainerFactory") + public SimpleRabbitListenerContainerFactory consumerBatchContainerFactory( + SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory) { + // 创建 SimpleRabbitListenerContainerFactory 对象 + SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); + configurer.configure(factory, connectionFactory); + // 额外添加批量消费的属性 + factory.setBatchListener(true); + return factory; + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo05Consumer.java b/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo05Consumer.java new file mode 100644 index 00000000..acd53fa3 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo05Consumer.java @@ -0,0 +1,29 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.consumer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo05Message; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.amqp.rabbit.annotation.RabbitHandler; +import org.springframework.amqp.rabbit.annotation.RabbitListener; +import org.springframework.stereotype.Component; + +import java.util.List; + +@Component +@RabbitListener(queues = Demo05Message.QUEUE, + containerFactory = "consumerBatchContainerFactory") +public class Demo05Consumer { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + @RabbitHandler + public void onMessage(List messages) { + logger.info("[onMessage][线程编号:{} 消息数量：{}]", Thread.currentThread().getId(), messages.size()); + } + +// @RabbitHandler(isDefault = true) +// public void onMessageX(List messages) { +// logger.info("[onMessage][线程编号:{} 消息数量：{}]", Thread.currentThread().getId(), messages.size()); +// } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo05Message.java b/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo05Message.java new file mode 100644 index 00000000..df6e0636 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo05Message.java @@ -0,0 +1,34 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.message; + +import java.io.Serializable; + +public class Demo05Message implements Serializable { + + public static final String QUEUE = "QUEUE_DEMO_05"; + + public static final String EXCHANGE = "EXCHANGE_DEMO_05"; + + public static final String ROUTING_KEY = "ROUTING_KEY_05"; + + /** + * 编号 + */ + private Integer id; + + public Demo05Message setId(Integer id) { + this.id = id; + return this; + } + + public Integer getId() { + return id; + } + + @Override + public String toString() { + return "Demo05Message{" + + "id=" + id + + '}'; + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo06Producer.java b/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo06Producer.java new file mode 100644 index 00000000..8768ed34 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo06Producer.java @@ -0,0 +1,22 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.producer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo05Message; +import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +@Component +public class Demo06Producer { + + @Autowired + private BatchingRabbitTemplate batchingRabbitTemplate; + + public void syncSend(Integer id) { + // 创建 Demo05Message 消息 + Demo05Message message = new Demo05Message(); + message.setId(id); + // 同步发送消息 + batchingRabbitTemplate.convertAndSend(Demo05Message.EXCHANGE, Demo05Message.ROUTING_KEY, message); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/resources/application.yaml b/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/resources/application.yaml new file mode 100644 index 00000000..52b3c35a --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume/src/main/resources/application.yaml @@ -0,0 +1,7 @@ +spring: + # RabbitMQ 配置项，对应 RabbitProperties 配置类 + rabbitmq: + host: 127.0.0.1 # RabbitMQ 服务的地址 + port: 5672 # RabbitMQ 服务的端口 + username: guest # RabbitMQ 服务的账号 + password: guest # RabbitMQ 服务的密 diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java b/lab-04/lab-04-rabbitmq-demo-batch-consume/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java new file mode 100644 index 00000000..766da5b0 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java @@ -0,0 +1 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo; diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo05ProducerTest.java b/lab-04/lab-04-rabbitmq-demo-batch-consume/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo05ProducerTest.java new file mode 100644 index 00000000..d8c70649 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo05ProducerTest.java @@ -0,0 +1,39 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.producer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.Application; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.context.junit4.SpringRunner; + +import java.util.concurrent.CountDownLatch; + +@RunWith(SpringRunner.class) +@SpringBootTest(classes = Application.class) +public class Demo05ProducerTest { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + @Autowired + private Demo06Producer producer; + + @Test + public void testSyncSend() throws InterruptedException { + for (int i = 0; i < 3; i++) { + // 同步发送消息 + int id = (int) (System.currentTimeMillis() / 1000); + producer.syncSend(id); + + // 故意每条消息之间，隔离 10 秒 + logger.info("[testASyncSend][发送编号：[{}] 发送成功]", id); + Thread.sleep(10 * 1000L); + } + + // 阻塞等待，保证消费 + new CountDownLatch(1).await(); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-batch-consume/target/classes/application.yaml b/lab-04/lab-04-rabbitmq-demo-batch-consume/target/classes/application.yaml new file mode 100644 index 00000000..59cf3bfb --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch-consume/target/classes/application.yaml @@ -0,0 +1,7 @@ +spring: + # RabbitMQ 配置项，对应 RabbitProperties 配置类 + rabbitmq: + host: 127.0.0.1 # RabbitMQ 服务的地址 + port: 5672 # RabbitMQ 服务的端口 + username: guest # RabbitMQ 服务的账号 + password: guest # RabbitMQ 服务的密码 diff --git a/lab-04/lab-04-rabbitmq-demo-batch/pom.xml b/lab-04/lab-04-rabbitmq-demo-batch/pom.xml new file mode 100644 index 00000000..6ea62743 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch/pom.xml @@ -0,0 +1,30 @@ + + + + org.springframework.boot + spring-boot-starter-parent + 2.2.1.RELEASE + + + 4.0.0 + + lab-04-rabbitmq-demo-batch + + + + + org.springframework.boot + spring-boot-starter-amqp + + + + + org.springframework.boot + spring-boot-starter-test + test + + + + diff --git a/lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java b/lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java new file mode 100644 index 00000000..54968a6a --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java @@ -0,0 +1,15 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.scheduling.annotation.EnableAsync; + +@SpringBootApplication +@EnableAsync // 开启异步 +public class Application { + + public static void main(String[] args) { + SpringApplication.run(Application.class, args); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java b/lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java new file mode 100644 index 00000000..4fee3dbc --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java @@ -0,0 +1,70 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.config; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo05Message; +import org.springframework.amqp.core.Binding; +import org.springframework.amqp.core.BindingBuilder; +import org.springframework.amqp.core.DirectExchange; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.batch.BatchingStrategy; +import org.springframework.amqp.rabbit.batch.SimpleBatchingStrategy; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler; + +@Configuration +public class RabbitConfig { + + /** + * Direct Exchange 示例的配置类 + */ + public static class DirectExchangeDemoConfiguration { + + // 创建 Queue + @Bean + public Queue demo05Queue() { + return new Queue(Demo05Message.QUEUE, // Queue 名字 + true, // durable: 是否持久化 + false, // exclusive: 是否排它 + false); // autoDelete: 是否自动删除 + } + + // 创建 Direct Exchange + @Bean + public DirectExchange demo05Exchange() { + return new DirectExchange(Demo05Message.EXCHANGE, + true, // durable: 是否持久化 + false); // exclusive: 是否排它 + } + + // 创建 Binding + // Exchange：Demo05Message.EXCHANGE + // Routing key：Demo05Message.ROUTING_KEY + // Queue：Demo05Message.QUEUE + @Bean + public Binding demo05Binding() { + return BindingBuilder.bind(demo05Queue()).to(demo05Exchange()).with(Demo05Message.ROUTING_KEY); + } + + } + + @Bean + public BatchingRabbitTemplate batchRabbitTemplate(ConnectionFactory connectionFactory) { + // 创建 BatchingStrategy 对象，代表批量策略 + int batchSize = 16384; // 超过收集的消息数量的最大条数。 + int bufferLimit = 33554432; // 每次批量发送消息的最大内存 + int timeout = 30000; // 超过收集的时间的最大等待时长，单位：毫秒 + BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(batchSize, bufferLimit, timeout); + + // 创建 TaskScheduler 对象，用于实现超时发送的定时器 + TaskScheduler taskScheduler = new ConcurrentTaskScheduler(); + + // 创建 BatchingRabbitTemplate 对象 + BatchingRabbitTemplate batchTemplate = new BatchingRabbitTemplate(batchingStrategy, taskScheduler); + batchTemplate.setConnectionFactory(connectionFactory); + return batchTemplate; + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo05Consumer.java b/lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo05Consumer.java new file mode 100644 index 00000000..8fd51335 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo05Consumer.java @@ -0,0 +1,21 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.consumer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo05Message; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.amqp.rabbit.annotation.RabbitHandler; +import org.springframework.amqp.rabbit.annotation.RabbitListener; +import org.springframework.stereotype.Component; + +@Component +@RabbitListener(queues = Demo05Message.QUEUE) +public class Demo05Consumer { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + @RabbitHandler + public void onMessage(Demo05Message message) { + logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo05Message.java b/lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo05Message.java new file mode 100644 index 00000000..df6e0636 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo05Message.java @@ -0,0 +1,34 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.message; + +import java.io.Serializable; + +public class Demo05Message implements Serializable { + + public static final String QUEUE = "QUEUE_DEMO_05"; + + public static final String EXCHANGE = "EXCHANGE_DEMO_05"; + + public static final String ROUTING_KEY = "ROUTING_KEY_05"; + + /** + * 编号 + */ + private Integer id; + + public Demo05Message setId(Integer id) { + this.id = id; + return this; + } + + public Integer getId() { + return id; + } + + @Override + public String toString() { + return "Demo05Message{" + + "id=" + id + + '}'; + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo05Producer.java b/lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo05Producer.java new file mode 100644 index 00000000..bd96fecb --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo05Producer.java @@ -0,0 +1,22 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.producer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo05Message; +import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +@Component +public class Demo05Producer { + + @Autowired + private BatchingRabbitTemplate batchingRabbitTemplate; + + public void syncSend(Integer id) { + // 创建 Demo05Message 消息 + Demo05Message message = new Demo05Message(); + message.setId(id); + // 同步发送消息 + batchingRabbitTemplate.convertAndSend(Demo05Message.EXCHANGE, Demo05Message.ROUTING_KEY, message); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-batch/src/main/resources/application.yaml b/lab-04/lab-04-rabbitmq-demo-batch/src/main/resources/application.yaml new file mode 100644 index 00000000..59cf3bfb --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch/src/main/resources/application.yaml @@ -0,0 +1,7 @@ +spring: + # RabbitMQ 配置项，对应 RabbitProperties 配置类 + rabbitmq: + host: 127.0.0.1 # RabbitMQ 服务的地址 + port: 5672 # RabbitMQ 服务的端口 + username: guest # RabbitMQ 服务的账号 + password: guest # RabbitMQ 服务的密码 diff --git a/lab-04/lab-04-rabbitmq-demo-batch/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java b/lab-04/lab-04-rabbitmq-demo-batch/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java new file mode 100644 index 00000000..766da5b0 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java @@ -0,0 +1 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo; diff --git a/lab-04/lab-04-rabbitmq-demo-batch/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo05ProducerTest.java b/lab-04/lab-04-rabbitmq-demo-batch/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo05ProducerTest.java new file mode 100644 index 00000000..ea3fa8d1 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-batch/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo05ProducerTest.java @@ -0,0 +1,39 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.producer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.Application; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.context.junit4.SpringRunner; + +import java.util.concurrent.CountDownLatch; + +@RunWith(SpringRunner.class) +@SpringBootTest(classes = Application.class) +public class Demo05ProducerTest { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + @Autowired + private Demo05Producer producer; + + @Test + public void testSyncSend() throws InterruptedException { + for (int i = 0; i < 3; i++) { + // 同步发送消息 + int id = (int) (System.currentTimeMillis() / 1000); + producer.syncSend(id); + + // 故意每条消息之间，隔离 10 秒 + logger.info("[testASyncSend][发送编号：[{}] 发送成功]", id); + Thread.sleep(10 * 1000L); + } + + // 阻塞等待，保证消费 + new CountDownLatch(1).await(); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java b/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java index 44bde799..6bf38c7c 100644 --- a/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java +++ b/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java @@ -2,6 +2,8 @@ package cn.iocoder.springboot.lab04.rabbitmqdemo.config; import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo01Message; import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo02Message; +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo03Message; +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo04Message; import org.springframework.amqp.core.*; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -75,4 +77,87 @@ public class RabbitConfig { } + /** + * Fanout Exchange 示例的配置类 + */ + public static class FanoutExchangeDemoConfiguration { + + // 创建 Queue A + @Bean + public Queue demo03QueueA() { + return new Queue(Demo03Message.QUEUE_A, // Queue 名字 + true, // durable: 是否持久化 + false, // exclusive: 是否排它 + false); // autoDelete: 是否自动删除 + } + + // 创建 Queue B + @Bean + public Queue demo03QueueB() { + return new Queue(Demo03Message.QUEUE_B, // Queue 名字 + true, // durable: 是否持久化 + false, // exclusive: 是否排它 + false); // autoDelete: 是否自动删除 + } + + // 创建 Fanout Exchange + @Bean + public FanoutExchange demo03Exchange() { + return new FanoutExchange(Demo03Message.EXCHANGE, + true, // durable: 是否持久化 + false); // exclusive: 是否排它 + } + + // 创建 Binding A + // Exchange：Demo03Message.EXCHANGE + // Queue：Demo03Message.QUEUE_A + @Bean + public Binding demo03BindingA() { + return BindingBuilder.bind(demo03QueueA()).to(demo03Exchange()); + } + + // 创建 Binding B + // Exchange：Demo03Message.EXCHANGE + // Queue：Demo03Message.QUEUE_B + @Bean + public Binding demo03BindingB() { + return BindingBuilder.bind(demo03QueueB()).to(demo03Exchange()); + } + + } + + /** + * Headers Exchange 示例的配置类 + */ + public static class HeadersExchangeDemoConfiguration { + + // 创建 Queue + @Bean + public Queue demo04Queue() { + return new Queue(Demo04Message.QUEUE, // Queue 名字 + true, // durable: 是否持久化 + false, // exclusive: 是否排它 + false); // autoDelete: 是否自动删除 + } + + // 创建 Headers Exchange + @Bean + public HeadersExchange demo04Exchange() { + return new HeadersExchange(Demo04Message.EXCHANGE, + true, // durable: 是否持久化 + false); // exclusive: 是否排它 + } + + // 创建 Binding + // Exchange：Demo04Message.EXCHANGE + // Queue：Demo04Message.QUEUE + // Headers: Demo04Message.HEADER_KEY + Demo04Message.HEADER_VALUE + @Bean + public Binding demo4Binding() { + return BindingBuilder.bind(demo04Queue()).to(demo04Exchange()) + .where(Demo04Message.HEADER_KEY).matches(Demo04Message.HEADER_VALUE); // 配置 Headers 匹配 + } + + } + } diff --git a/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo03ConsumerA.java b/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo03ConsumerA.java new file mode 100644 index 00000000..5cf22a8c --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo03ConsumerA.java @@ -0,0 +1,21 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.consumer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo03Message; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.amqp.rabbit.annotation.RabbitHandler; +import org.springframework.amqp.rabbit.annotation.RabbitListener; +import org.springframework.stereotype.Component; + +@Component +@RabbitListener(queues = Demo03Message.QUEUE_A) +public class Demo03ConsumerA { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + @RabbitHandler + public void onMessage(Demo03Message message) { + logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo03ConsumerB.java b/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo03ConsumerB.java new file mode 100644 index 00000000..d3472f58 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo03ConsumerB.java @@ -0,0 +1,21 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.consumer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo03Message; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.amqp.rabbit.annotation.RabbitHandler; +import org.springframework.amqp.rabbit.annotation.RabbitListener; +import org.springframework.stereotype.Component; + +@Component +@RabbitListener(queues = Demo03Message.QUEUE_B) +public class Demo03ConsumerB { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + @RabbitHandler + public void onMessage(Demo03Message message) { + logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo04Consumer.java b/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo04Consumer.java new file mode 100644 index 00000000..aa39ae48 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo04Consumer.java @@ -0,0 +1,26 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.consumer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo04Message; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.amqp.rabbit.annotation.RabbitHandler; +import org.springframework.amqp.rabbit.annotation.RabbitListener; +import org.springframework.stereotype.Component; + +@Component +@RabbitListener(queues = Demo04Message.QUEUE) +public class Demo04Consumer { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + @RabbitHandler + public void onMessage(Demo04Message message) { + logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message); + } + +// @RabbitHandler(isDefault = true) +// public void onMessage(org.springframework.amqp.core.Message message) { +// logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message); +// } + +} diff --git a/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo03Message.java b/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo03Message.java new file mode 100644 index 00000000..87f963d6 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo03Message.java @@ -0,0 +1,33 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.message; + +import java.io.Serializable; + +public class Demo03Message implements Serializable { + + public static final String QUEUE_A = "QUEUE_DEMO_03_A"; + public static final String QUEUE_B = "QUEUE_DEMO_03_B"; + + public static final String EXCHANGE = "EXCHANGE_DEMO_03"; + + /** + * 编号 + */ + private Integer id; + + public Demo03Message setId(Integer id) { + this.id = id; + return this; + } + + public Integer getId() { + return id; + } + + @Override + public String toString() { + return "Demo03Message{" + + "id=" + id + + '}'; + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo04Message.java b/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo04Message.java new file mode 100644 index 00000000..313488ac --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo04Message.java @@ -0,0 +1,35 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.message; + +import java.io.Serializable; + +public class Demo04Message implements Serializable { + + public static final String QUEUE = "QUEUE_DEMO_04_A"; + + public static final String EXCHANGE = "EXCHANGE_DEMO_04"; + + public static final String HEADER_KEY = "color"; + public static final String HEADER_VALUE = "red"; + + /** + * 编号 + */ + private Integer id; + + public Demo04Message setId(Integer id) { + this.id = id; + return this; + } + + public Integer getId() { + return id; + } + + @Override + public String toString() { + return "Demo04Message{" + + "id=" + id + + '}'; + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo03Producer.java b/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo03Producer.java new file mode 100644 index 00000000..16e07af9 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo03Producer.java @@ -0,0 +1,22 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.producer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo03Message; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +@Component +public class Demo03Producer { + + @Autowired + private RabbitTemplate rabbitTemplate; + + public void syncSend(Integer id) { + // 创建 Demo03Message 消息 + Demo03Message message = new Demo03Message(); + message.setId(id); + // 同步发送消息 + rabbitTemplate.convertAndSend(Demo03Message.EXCHANGE, null, message); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo04Producer.java b/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo04Producer.java new file mode 100644 index 00000000..3a3d5952 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo04Producer.java @@ -0,0 +1,27 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.producer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo04Message; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +@Component +public class Demo04Producer { + + @Autowired + private RabbitTemplate rabbitTemplate; + + public void syncSend(Integer id, String headerValue) { + // 创建 MessageProperties 属性 + MessageProperties messageProperties = new MessageProperties(); + messageProperties.setHeader(Demo04Message.HEADER_KEY, headerValue); // 设置 header + // 创建 Message 消息 + Message message = rabbitTemplate.getMessageConverter().toMessage( + new Demo04Message().setId(id), messageProperties); + // 同步发送消息 + rabbitTemplate.send(Demo04Message.EXCHANGE, null, message); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo03ProducerTest.java b/lab-04/lab-04-rabbitmq-demo/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo03ProducerTest.java new file mode 100644 index 00000000..6a7beeeb --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo03ProducerTest.java @@ -0,0 +1,33 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.producer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.Application; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.context.junit4.SpringRunner; + +import java.util.concurrent.CountDownLatch; + +@RunWith(SpringRunner.class) +@SpringBootTest(classes = Application.class) +public class Demo03ProducerTest { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + @Autowired + private Demo03Producer producer; + + @Test + public void testSyncSend() throws InterruptedException { + int id = (int) (System.currentTimeMillis() / 1000); + producer.syncSend(id); + logger.info("[testSyncSend][发送编号：[{}] 发送成功]", id); + + // 阻塞等待，保证消费 + new CountDownLatch(1).await(); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo04ProducerTest.java b/lab-04/lab-04-rabbitmq-demo/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo04ProducerTest.java new file mode 100644 index 00000000..37ea0cd6 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo04ProducerTest.java @@ -0,0 +1,44 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.producer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.Application; +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo04Message; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.context.junit4.SpringRunner; + +import java.util.concurrent.CountDownLatch; + +@RunWith(SpringRunner.class) +@SpringBootTest(classes = Application.class) +public class Demo04ProducerTest { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + @Autowired + private Demo04Producer producer; + + @Test + public void testSyncSendSuccess() throws InterruptedException { + int id = (int) (System.currentTimeMillis() / 1000); + producer.syncSend(id, Demo04Message.HEADER_VALUE); + logger.info("[testSyncSend][发送编号：[{}] 发送成功]", id); + + // 阻塞等待，保证消费 + new CountDownLatch(1).await(); + } + + @Test + public void testSyncSendFailure() throws InterruptedException { + int id = (int) (System.currentTimeMillis() / 1000); + producer.syncSend(id, "error"); + logger.info("[testSyncSend][发送编号：[{}] 发送成功]", id); + + // 阻塞等待，保证消费 + new CountDownLatch(1).await(); + } + +} diff --git a/lab-04/pom.xml b/lab-04/pom.xml index 479c4119..d7b7a357 100644 --- a/lab-04/pom.xml +++ b/lab-04/pom.xml @@ -14,6 +14,9 @@ lab-04-rabbitmq-native lab-04-rabbitmq-demo + lab-04-rabbitmq-demo-batch + lab-04-rabbitmq-demo-batch-consume + lab-04-rabbitmq-demo-batch-consume-02