From c3cce54f233a4b7609147e4fea8d947bd07c4677 Mon Sep 17 00:00:00 2001 From: YunaiV <> Date: Thu, 12 Dec 2019 23:35:44 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=20rabbitmq=20=E5=B9=B6?= =?UTF-8?q?=E5=8F=91=E6=B6=88=E8=B4=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../rabbitmqdemo/consumer/Demo09Consumer.java | 4 +- .../src/main/resources/application.yaml | 6 +- .../producer/Demo09ProducerTest.java | 2 +- .../target/classes/application.yaml | 10 +-- lab-04/lab-04-rabbitmq-demo-orderly/pom.xml | 30 ++++++++ .../lab04/rabbitmqdemo/Application.java | 18 +++++ .../rabbitmqdemo/config/RabbitConfig.java | 65 ++++++++++++++++++ .../rabbitmqdemo/consumer/Demo10Consumer.java | 30 ++++++++ .../rabbitmqdemo/message/Demo10Message.java | 38 ++++++++++ .../rabbitmqdemo/producer/Demo10Producer.java | 26 +++++++ .../src/main/resources/application.yaml | 7 ++ .../lab04/rabbitmqdemo/package-info.java | 1 + .../producer/Demo10ProducerTest.java | 36 ++++++++++ ...04-rabbitmq-demo-concurrency.kotlin_module | Bin 0 -> 16 bytes .../target/classes/application.yaml | 7 ++ ...04-rabbitmq-demo-concurrency.kotlin_module | Bin 0 -> 16 bytes .../lab-04-rabbitmq-demo-transaction/pom.xml | 30 ++++++++ .../lab04/rabbitmqdemo/Application.java | 15 ++++ .../rabbitmqdemo/config/RabbitConfig.java | 61 ++++++++++++++++ .../rabbitmqdemo/consumer/Demo11Consumer.java | 21 ++++++ .../rabbitmqdemo/message/Demo11Message.java | 34 +++++++++ .../rabbitmqdemo/producer/Demo11Producer.java | 32 +++++++++ .../src/main/resources/application.yaml | 7 ++ .../lab04/rabbitmqdemo/package-info.java | 1 + .../producer/Demo11ProducerTest.java | 28 ++++++++ .../target/classes/application.yaml | 7 ++ lab-04/pom.xml | 2 + 27 files changed, 508 insertions(+), 10 deletions(-) create mode 100644 lab-04/lab-04-rabbitmq-demo-orderly/pom.xml create mode 100644 lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java create mode 100644 lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java create mode 100644 lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo10Consumer.java create mode 100644 lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo10Message.java create mode 100644 lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo10Producer.java create mode 100644 lab-04/lab-04-rabbitmq-demo-orderly/src/main/resources/application.yaml create mode 100644 lab-04/lab-04-rabbitmq-demo-orderly/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java create mode 100644 lab-04/lab-04-rabbitmq-demo-orderly/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo10ProducerTest.java create mode 100644 lab-04/lab-04-rabbitmq-demo-orderly/target/classes/META-INF/lab-04-rabbitmq-demo-concurrency.kotlin_module create mode 100644 lab-04/lab-04-rabbitmq-demo-orderly/target/classes/application.yaml create mode 100644 lab-04/lab-04-rabbitmq-demo-orderly/target/test-classes/META-INF/lab-04-rabbitmq-demo-concurrency.kotlin_module create mode 100644 lab-04/lab-04-rabbitmq-demo-transaction/pom.xml create mode 100644 lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java create mode 100644 lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java create mode 100644 lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo11Consumer.java create mode 100644 lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo11Message.java create mode 100644 lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo11Producer.java create mode 100644 lab-04/lab-04-rabbitmq-demo-transaction/src/main/resources/application.yaml create mode 100644 lab-04/lab-04-rabbitmq-demo-transaction/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java create mode 100644 lab-04/lab-04-rabbitmq-demo-transaction/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo11ProducerTest.java create mode 100644 lab-04/lab-04-rabbitmq-demo-transaction/target/classes/application.yaml diff --git a/lab-04/lab-04-rabbitmq-demo-concurrency/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo09Consumer.java b/lab-04/lab-04-rabbitmq-demo-concurrency/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo09Consumer.java index a794546d..268a0b82 100644 --- a/lab-04/lab-04-rabbitmq-demo-concurrency/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo09Consumer.java +++ b/lab-04/lab-04-rabbitmq-demo-concurrency/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo09Consumer.java @@ -8,7 +8,9 @@ import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component -@RabbitListener(queues = Demo09Message.QUEUE) +//@RabbitListener(queues = Demo09Message.QUEUE) +@RabbitListener(queues = Demo09Message.QUEUE, + concurrency = "2") //@RabbitListener(queues = {"QUEUE_DEMO_01", "QUEUE_DEMO_02"}) public class Demo09Consumer { diff --git a/lab-04/lab-04-rabbitmq-demo-concurrency/src/main/resources/application.yaml b/lab-04/lab-04-rabbitmq-demo-concurrency/src/main/resources/application.yaml index e9f4ed65..51cd27ad 100644 --- a/lab-04/lab-04-rabbitmq-demo-concurrency/src/main/resources/application.yaml +++ b/lab-04/lab-04-rabbitmq-demo-concurrency/src/main/resources/application.yaml @@ -6,9 +6,9 @@ spring: username: guest # RabbitMQ 服务的账号 password: guest # RabbitMQ 服务的密码 listener: - type: direct + type: simple # 选择的 ListenerContainer 的类型。默认为 direct 类型 simple: - concurrency: 2 - max-concurrency: 10 + concurrency: 2 # 每个 @ListenerContainer 的并发消费的线程数 + max-concurrency: 10 # 每个 @ListenerCon 允许的并发消费的线程数 # direct: # consumers-per-queue: 2 # 对于每一个 @RabbitListener ，一个 Queue ，对应创建几个 Consumer 。 diff --git a/lab-04/lab-04-rabbitmq-demo-concurrency/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo09ProducerTest.java b/lab-04/lab-04-rabbitmq-demo-concurrency/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo09ProducerTest.java index 4620db3b..5797ac0f 100644 --- a/lab-04/lab-04-rabbitmq-demo-concurrency/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo09ProducerTest.java +++ b/lab-04/lab-04-rabbitmq-demo-concurrency/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo09ProducerTest.java @@ -25,7 +25,7 @@ public class Demo09ProducerTest { for (int i = 0; i < 10; i++) { int id = (int) (System.currentTimeMillis() / 1000); producer.syncSend(id); - logger.info("[testSyncSend][发送编号：[{}] 发送成功]", id); +// logger.info("[testSyncSend][发送编号：[{}] 发送成功]", id); } // 阻塞等待，保证消费 diff --git a/lab-04/lab-04-rabbitmq-demo-concurrency/target/classes/application.yaml b/lab-04/lab-04-rabbitmq-demo-concurrency/target/classes/application.yaml index 16fc5bab..51cd27ad 100644 --- a/lab-04/lab-04-rabbitmq-demo-concurrency/target/classes/application.yaml +++ b/lab-04/lab-04-rabbitmq-demo-concurrency/target/classes/application.yaml @@ -6,9 +6,9 @@ spring: username: guest # RabbitMQ 服务的账号 password: guest # RabbitMQ 服务的密码 listener: - type: direct + type: simple # 选择的 ListenerContainer 的类型。默认为 direct 类型 simple: - concurrency: 2 - max-concurrency: 10 - direct: - consumers-per-queue: 2 + concurrency: 2 # 每个 @ListenerContainer 的并发消费的线程数 + max-concurrency: 10 # 每个 @ListenerCon 允许的并发消费的线程数 +# direct: +# consumers-per-queue: 2 # 对于每一个 @RabbitListener ，一个 Queue ，对应创建几个 Consumer 。 diff --git a/lab-04/lab-04-rabbitmq-demo-orderly/pom.xml b/lab-04/lab-04-rabbitmq-demo-orderly/pom.xml new file mode 100644 index 00000000..5b36a2f9 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-orderly/pom.xml @@ -0,0 +1,30 @@ + + + + org.springframework.boot + spring-boot-starter-parent + 2.2.1.RELEASE + + + 4.0.0 + + lab-04-rabbitmq-demo-orderly + + + + + org.springframework.boot + spring-boot-starter-amqp + + + + + org.springframework.boot + spring-boot-starter-test + test + + + + diff --git a/lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java b/lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java new file mode 100644 index 00000000..dc3526f6 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java @@ -0,0 +1,18 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.scheduling.annotation.EnableAsync; + +/** + * TODO RabbitMQ 顺序消息的示例，暂时未提供 + */ +@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-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java b/lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java new file mode 100644 index 00000000..dff437da --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java @@ -0,0 +1,65 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.config; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo10Message; +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.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration +public class RabbitConfig { + + /** + * Direct Exchange 示例的配置类 + */ + public static class DirectExchangeDemoConfiguration { + + // 创建 Queue + @Bean + public Queue demo10Queue0() { + return new Queue(Demo10Message.QUEUE_0); + } + @Bean + public Queue demo10Queue1() { + return new Queue(Demo10Message.QUEUE_1); + } + @Bean + public Queue demo10Queue2() { + return new Queue(Demo10Message.QUEUE_2); + } + @Bean + public Queue demo10Queue3() { + return new Queue(Demo10Message.QUEUE_3); + } + + // 创建 Direct Exchange + @Bean + public DirectExchange demo10Exchange() { + return new DirectExchange(Demo10Message.EXCHANGE, + true, // durable: 是否持久化 + false); // exclusive: 是否排它 + } + + // 创建 Binding + @Bean + public Binding demo10Binding0() { + return BindingBuilder.bind(demo10Queue0()).to(demo10Exchange()).with("0"); + } + @Bean + public Binding demo10Binding1() { + return BindingBuilder.bind(demo10Queue1()).to(demo10Exchange()).with("1"); + } + @Bean + public Binding demo10Binding2() { + return BindingBuilder.bind(demo10Queue2()).to(demo10Exchange()).with("2"); + } + @Bean + public Binding demo10Binding3() { + return BindingBuilder.bind(demo10Queue3()).to(demo10Exchange()).with("3"); + } + + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo10Consumer.java b/lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo10Consumer.java new file mode 100644 index 00000000..2162b7cd --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo10Consumer.java @@ -0,0 +1,30 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.consumer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo10Message; +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.messaging.Message; +import org.springframework.stereotype.Component; + +@Component +@RabbitListener(queues = Demo10Message.QUEUE_0) +@RabbitListener(queues = Demo10Message.QUEUE_1) +@RabbitListener(queues = Demo10Message.QUEUE_2) +@RabbitListener(queues = Demo10Message.QUEUE_3) +public class Demo10Consumer { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + @RabbitHandler(isDefault = true) + public void onMessage(Message message) { + logger.info("[onMessage][线程编号:{} Queue:{} 消息编号：{}]", Thread.currentThread().getId(), getQueue(message), + message.getPayload().getId()); + } + + private static String getQueue(Message message) { + return message.getHeaders().get("amqp_consumerQueue", String.class); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo10Message.java b/lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo10Message.java new file mode 100644 index 00000000..2e7348c7 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo10Message.java @@ -0,0 +1,38 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.message; + +import java.io.Serializable; + +public class Demo10Message implements Serializable { + + private static final String QUEUE_BASE = "QUEUE_DEMO_10-"; + public static final String QUEUE_0 = QUEUE_BASE + "0"; + public static final String QUEUE_1 = QUEUE_BASE + "1"; + public static final String QUEUE_2 = QUEUE_BASE + "2"; + public static final String QUEUE_3 = QUEUE_BASE + "3"; + + public static final int QUEUE_COUNT = 4; + + public static final String EXCHANGE = "EXCHANGE_DEMO_10"; + + /** + * 编号 + */ + private Integer id; + + public Demo10Message setId(Integer id) { + this.id = id; + return this; + } + + public Integer getId() { + return id; + } + + @Override + public String toString() { + return "Demo10Message{" + + "id=" + id + + '}'; + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo10Producer.java b/lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo10Producer.java new file mode 100644 index 00000000..3817d065 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo10Producer.java @@ -0,0 +1,26 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.producer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo10Message; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +@Component +public class Demo10Producer { + + @Autowired + private RabbitTemplate rabbitTemplate; + + public void syncSend(Integer id) { + // 创建 Demo10Message 消息 + Demo10Message message = new Demo10Message(); + message.setId(id); + // 同步发送消息 + rabbitTemplate.convertAndSend(Demo10Message.EXCHANGE, this.getRoutingKey(id), message); + } + + private String getRoutingKey(Integer id) { + return String.valueOf(id % Demo10Message.QUEUE_COUNT); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-orderly/src/main/resources/application.yaml b/lab-04/lab-04-rabbitmq-demo-orderly/src/main/resources/application.yaml new file mode 100644 index 00000000..59cf3bfb --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-orderly/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-orderly/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java b/lab-04/lab-04-rabbitmq-demo-orderly/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-orderly/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-orderly/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo10ProducerTest.java b/lab-04/lab-04-rabbitmq-demo-orderly/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo10ProducerTest.java new file mode 100644 index 00000000..e26df613 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-orderly/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo10ProducerTest.java @@ -0,0 +1,36 @@ +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 Demo10ProducerTest { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + @Autowired + private Demo10Producer producer; + + @Test + public void testSyncSend() throws InterruptedException { + for (int i = 0; i < 2; i++) { + for (int id = 0; id < 4; id++) { + producer.syncSend(id); +// logger.info("[testSyncSend][发送编号：[{}] 发送成功]", id); + } + } + + // 阻塞等待，保证消费 + new CountDownLatch(1).await(); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-orderly/target/classes/META-INF/lab-04-rabbitmq-demo-concurrency.kotlin_module b/lab-04/lab-04-rabbitmq-demo-orderly/target/classes/META-INF/lab-04-rabbitmq-demo-concurrency.kotlin_module new file mode 100644 index 0000000000000000000000000000000000000000..2983af70661ad375cc499ebc4da5a68ca46c532e GIT binary patch literal 16 RcmZQzU|?ooU|@t|egFVe02KfL literal 0 HcmV?d00001 diff --git a/lab-04/lab-04-rabbitmq-demo-orderly/target/classes/application.yaml b/lab-04/lab-04-rabbitmq-demo-orderly/target/classes/application.yaml new file mode 100644 index 00000000..59cf3bfb --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-orderly/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-orderly/target/test-classes/META-INF/lab-04-rabbitmq-demo-concurrency.kotlin_module b/lab-04/lab-04-rabbitmq-demo-orderly/target/test-classes/META-INF/lab-04-rabbitmq-demo-concurrency.kotlin_module new file mode 100644 index 0000000000000000000000000000000000000000..2983af70661ad375cc499ebc4da5a68ca46c532e GIT binary patch literal 16 RcmZQzU|?ooU|@t|egFVe02KfL literal 0 HcmV?d00001 diff --git a/lab-04/lab-04-rabbitmq-demo-transaction/pom.xml b/lab-04/lab-04-rabbitmq-demo-transaction/pom.xml new file mode 100644 index 00000000..107f8811 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-transaction/pom.xml @@ -0,0 +1,30 @@ + + + + org.springframework.boot + spring-boot-starter-parent + 2.2.1.RELEASE + + + 4.0.0 + + lab-04-rabbitmq-demo-transaction + + + + + org.springframework.boot + spring-boot-starter-amqp + + + + + org.springframework.boot + spring-boot-starter-test + test + + + + diff --git a/lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java b/lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java new file mode 100644 index 00000000..fbfbdc73 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-transaction/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.transaction.annotation.EnableTransactionManagement; + +@SpringBootApplication +@EnableTransactionManagement // 开启事务 +public class Application { + + public static void main(String[] args) { + SpringApplication.run(Application.class, args); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java b/lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java new file mode 100644 index 00000000..b03756fd --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java @@ -0,0 +1,61 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.config; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo11Message; +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.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.transaction.annotation.EnableTransactionManagement; + +@Configuration +@EnableTransactionManagement +public class RabbitConfig { + + /** + * Direct Exchange 示例的配置类 + */ + public static class DirectExchangeDemoConfiguration { + + // 创建 Queue + @Bean + public Queue demo11Queue() { + return new Queue(Demo11Message.QUEUE, // Queue 名字 + true, // durable: 是否持久化 + false, // exclusive: 是否排它 + false); // autoDelete: 是否自动删除 + } + + // 创建 Direct Exchange + @Bean + public DirectExchange demo11Exchange() { + return new DirectExchange(Demo11Message.EXCHANGE, + true, // durable: 是否持久化 + false); // exclusive: 是否排它 + } + + // 创建 Binding + // Exchange：Demo11Message.EXCHANGE + // Routing key：Demo11Message.ROUTING_KEY + // Queue：Demo11Message.QUEUE + @Bean + public Binding demo11Binding() { + return BindingBuilder.bind(demo11Queue()).to(demo11Exchange()).with(Demo11Message.ROUTING_KEY); + } + + } + + @Bean + public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory, RabbitTemplate rabbitTemplate) { + // 设置 RabbitTemplate 支持事务 + rabbitTemplate.setChannelTransacted(true); + + // 创建 RabbitTransactionManager 对象 + return new RabbitTransactionManager(connectionFactory); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo11Consumer.java b/lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo11Consumer.java new file mode 100644 index 00000000..d883bac0 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo11Consumer.java @@ -0,0 +1,21 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.consumer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo11Message; +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 = Demo11Message.QUEUE) +public class Demo11Consumer { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + @RabbitHandler + public void onMessage(Demo11Message message) { + logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo11Message.java b/lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo11Message.java new file mode 100644 index 00000000..644f262f --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo11Message.java @@ -0,0 +1,34 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.message; + +import java.io.Serializable; + +public class Demo11Message implements Serializable { + + public static final String QUEUE = "QUEUE_DEMO_11"; + + public static final String EXCHANGE = "EXCHANGE_DEMO_11"; + + public static final String ROUTING_KEY = "ROUTING_KEY_11"; + + /** + * 编号 + */ + private Integer id; + + public Demo11Message setId(Integer id) { + this.id = id; + return this; + } + + public Integer getId() { + return id; + } + + @Override + public String toString() { + return "Demo11Message{" + + "id=" + id + + '}'; + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo11Producer.java b/lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo11Producer.java new file mode 100644 index 00000000..602705b7 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo11Producer.java @@ -0,0 +1,32 @@ +package cn.iocoder.springboot.lab04.rabbitmqdemo.producer; + +import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo11Message; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; +import org.springframework.transaction.annotation.Transactional; + +@Component +public class Demo11Producer { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + @Autowired + private RabbitTemplate rabbitTemplate; + + @Transactional + public void syncSend(Integer id) throws InterruptedException { + // 创建 Demo11Message 消息 + Demo11Message message = new Demo11Message(); + message.setId(id); + // 同步发送消息 + rabbitTemplate.convertAndSend(Demo11Message.EXCHANGE, Demo11Message.ROUTING_KEY, message); + logger.info("[syncSend][发送编号：[{}] 发送成功]", id); + + // 等待 + Thread.sleep(10 * 1000L); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-transaction/src/main/resources/application.yaml b/lab-04/lab-04-rabbitmq-demo-transaction/src/main/resources/application.yaml new file mode 100644 index 00000000..59cf3bfb --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-transaction/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-transaction/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java b/lab-04/lab-04-rabbitmq-demo-transaction/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-transaction/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-transaction/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo11ProducerTest.java b/lab-04/lab-04-rabbitmq-demo-transaction/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo11ProducerTest.java new file mode 100644 index 00000000..fc771b73 --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-transaction/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo11ProducerTest.java @@ -0,0 +1,28 @@ +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.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 Demo11ProducerTest { + + @Autowired + private Demo11Producer producer; + + @Test + public void testSyncSend() throws InterruptedException { + int id = (int) (System.currentTimeMillis() / 1000); + producer.syncSend(id); + + // 阻塞等待，保证消费 + new CountDownLatch(1).await(); + } + +} diff --git a/lab-04/lab-04-rabbitmq-demo-transaction/target/classes/application.yaml b/lab-04/lab-04-rabbitmq-demo-transaction/target/classes/application.yaml new file mode 100644 index 00000000..59cf3bfb --- /dev/null +++ b/lab-04/lab-04-rabbitmq-demo-transaction/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/pom.xml b/lab-04/pom.xml index 50c636db..dddc7b07 100644 --- a/lab-04/pom.xml +++ b/lab-04/pom.xml @@ -21,6 +21,8 @@ lab-04-rabbitmq-demo-delay lab-04-rabbitmq-demo-message-model lab-04-rabbitmq-demo-concurrency + lab-04-rabbitmq-demo-orderly + lab-04-rabbitmq-demo-transaction