diff --git a/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/pom.xml b/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/pom.xml new file mode 100644 index 00000000..289cd507 --- /dev/null +++ b/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/pom.xml @@ -0,0 +1,58 @@ + + + + labx-10 + cn.iocoder.springboot.labs + 1.0-SNAPSHOT + + 4.0.0 + + labx-10-sc-stream-rabbitmq-producer-transaction + + + 1.8 + 1.8 + 2.2.4.RELEASE + Hoxton.SR1 + + + + + + + org.springframework.boot + spring-boot-starter-parent + ${spring.boot.version} + pom + import + + + org.springframework.cloud + spring-cloud-dependencies + ${spring.cloud.version} + pom + import + + + + + + + + org.springframework.boot + spring-boot-starter-web + + + + + org.springframework.cloud + spring-cloud-starter-stream-rabbit + + + + diff --git a/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/producerdemo/ProducerApplication.java b/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/producerdemo/ProducerApplication.java new file mode 100644 index 00000000..c128d2f4 --- /dev/null +++ b/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/producerdemo/ProducerApplication.java @@ -0,0 +1,16 @@ +package cn.iocoder.springcloud.labx10.rabbitmqdemo.producerdemo; + +import cn.iocoder.springcloud.labx10.rabbitmqdemo.producerdemo.message.MySource; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.cloud.stream.annotation.EnableBinding; + +@SpringBootApplication +@EnableBinding(MySource.class) +public class ProducerApplication { + + public static void main(String[] args) { + SpringApplication.run(ProducerApplication.class, args); + } + +} diff --git a/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/producerdemo/config/TransactionConfig.java b/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/producerdemo/config/TransactionConfig.java new file mode 100644 index 00000000..bd2d2a75 --- /dev/null +++ b/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/producerdemo/config/TransactionConfig.java @@ -0,0 +1,18 @@ +package cn.iocoder.springcloud.labx10.rabbitmqdemo.producerdemo.config; + +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +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 TransactionConfig { + + @Bean + public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory) { + return new RabbitTransactionManager(connectionFactory); + } + +} diff --git a/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/producerdemo/controller/Demo01Controller.java b/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/producerdemo/controller/Demo01Controller.java new file mode 100644 index 00000000..007def82 --- /dev/null +++ b/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/producerdemo/controller/Demo01Controller.java @@ -0,0 +1,44 @@ +package cn.iocoder.springcloud.labx10.rabbitmqdemo.producerdemo.controller; + +import cn.iocoder.springcloud.labx10.rabbitmqdemo.producerdemo.message.Demo01Message; +import cn.iocoder.springcloud.labx10.rabbitmqdemo.producerdemo.message.MySource; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +import java.util.Random; + +@RestController +@RequestMapping("/demo01") +public class Demo01Controller { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + @Autowired + private MySource mySource; + + @Transactional + @GetMapping("/send_transaction") + public void sendTransaction() throws InterruptedException { + // 创建 Message + int id = new Random().nextInt(); + Demo01Message message = new Demo01Message() + .setId(id); + // 创建 Spring Message 对象 + Message springMessage = MessageBuilder.withPayload(message) + .build(); + // 发送消息 + mySource.demo01Output().send(springMessage); + logger.info("[syncSend][发送编号：[{}] 发送成功]", id); + + // 等待 + Thread.sleep(10 * 1000L); + } + +} diff --git a/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/producerdemo/message/Demo01Message.java b/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/producerdemo/message/Demo01Message.java new file mode 100644 index 00000000..7f015623 --- /dev/null +++ b/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/producerdemo/message/Demo01Message.java @@ -0,0 +1,29 @@ +package cn.iocoder.springcloud.labx10.rabbitmqdemo.producerdemo.message; + +/** + * 示例 01 的 Message 消息 + */ +public class Demo01Message { + + /** + * 编号 + */ + private Integer id; + + public Demo01Message setId(Integer id) { + this.id = id; + return this; + } + + public Integer getId() { + return id; + } + + @Override + public String toString() { + return "Demo01Message{" + + "id=" + id + + '}'; + } + +} diff --git a/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/producerdemo/message/MySource.java b/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/producerdemo/message/MySource.java new file mode 100644 index 00000000..64d06834 --- /dev/null +++ b/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/producerdemo/message/MySource.java @@ -0,0 +1,11 @@ +package cn.iocoder.springcloud.labx10.rabbitmqdemo.producerdemo.message; + +import org.springframework.cloud.stream.annotation.Output; +import org.springframework.messaging.MessageChannel; + +public interface MySource { + + @Output("demo01-output") + MessageChannel demo01Output(); + +} diff --git a/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/resources/application.yml b/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/resources/application.yml new file mode 100644 index 00000000..2b120690 --- /dev/null +++ b/labx-10/labx-10-sc-stream-rabbitmq-producer-transaction/src/main/resources/application.yml @@ -0,0 +1,34 @@ +spring: + application: + name: demo-producer-application + cloud: + # Spring Cloud Stream 配置项，对应 BindingServiceProperties 类 + stream: + # Binder 配置项，对应 BinderProperties Map + binders: + rabbit001: + type: rabbit # 设置 Binder 的类型 + environment: # 设置 Binder 的环境配置 + # 如果是 RabbitMQ 类型的时候，则对应的是 RabbitProperties 类 + spring: + rabbitmq: + host: 127.0.0.1 # RabbitMQ 服务的地址 + port: 5672 # RabbitMQ 服务的端口 + username: guest # RabbitMQ 服务的账号 + password: guest # RabbitMQ 服务的密码 + # Binding 配置项，对应 BindingProperties Map + bindings: + demo01-output: + destination: DEMO-TOPIC-01 # 目的地。这里使用 RabbitMQ Exchange + content-type: application/json # 内容格式。这里使用 JSON + binder: rabbit001 # 设置使用的 Binder 名字 + # RabbitMQ 自定义 Binding 配置项，对应 RabbitBindingProperties Map + rabbit: + bindings: + demo01-output: + # RabbitMQ Producer 配置项，对应 RabbitProducerProperties 类 + producer: + transacted: true # 是否开启事务功能，默认为 false + +server: + port: 18080 diff --git a/labx-10/pom.xml b/labx-10/pom.xml index d5b46f6f..d81a3be1 100644 --- a/labx-10/pom.xml +++ b/labx-10/pom.xml @@ -35,6 +35,9 @@ labx-10-sc-stream-rabbitmq-consumer-filter + + labx-10-sc-stream-rabbitmq-producer-transaction +