lywsvip/SpringBoot-Labs
1
0
Fork 0
mirror of https://gitee.com/yudaocode/SpringBoot-Labs.git synced 2026-09-03 05:53:54 +08:00
Code Issues Packages Projects Releases Wiki Activity

增加 rabbitmq 并发消费

Browse Source
This commit is contained in:
YunaiV
2019-12-12 23:35:44 +08:00
parent 64280ad775
commit c3cce54f23
27 changed files with 508 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
lab-04/lab-04-rabbitmq-demo-concurrency/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo09Consumer.java
View File

@@ -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 {

6
lab-04/lab-04-rabbitmq-demo-concurrency/src/main/resources/application.yaml
View File

@@ -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 。

2
lab-04/lab-04-rabbitmq-demo-concurrency/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo09ProducerTest.java
View File

@@ -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);
}
// 阻塞等待，保证消费

10
lab-04/lab-04-rabbitmq-demo-concurrency/target/classes/application.yaml
View File

@@ -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 。

30
lab-04/lab-04-rabbitmq-demo-orderly/pom.xml Normal file
View File

@@ -0,0 +1,30 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.1.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>lab-04-rabbitmq-demo-orderly</artifactId>
<dependencies>
<!-- 实现对 RabbitMQ 的自动化配置 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!-- 方便等会写单元测试 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

18
lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java Normal file
View File

@@ -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);
}
}

65
lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java Normal file
View File

@@ -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");
}
}
}

30
lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo10Consumer.java Normal file
View File

@@ -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<Demo10Message> message) {
logger.info("[onMessage][线程编号:{} Queue:{} 消息编号：{}]", Thread.currentThread().getId(), getQueue(message),
message.getPayload().getId());
}
private static String getQueue(Message<Demo10Message> message) {
return message.getHeaders().get("amqp_consumerQueue", String.class);
}
}

38
lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo10Message.java Normal file
View File

@@ -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 +
'}';
}
}

26
lab-04/lab-04-rabbitmq-demo-orderly/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo10Producer.java Normal file
View File

@@ -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);
}
}

7
lab-04/lab-04-rabbitmq-demo-orderly/src/main/resources/application.yaml Normal file
View File

@@ -0,0 +1,7 @@
spring:
# RabbitMQ 配置项，对应 RabbitProperties 配置类
rabbitmq:
host: 127.0.0.1 # RabbitMQ 服务的地址
port: 5672 # RabbitMQ 服务的端口
username: guest # RabbitMQ 服务的账号
password: guest # RabbitMQ 服务的密码

1
lab-04/lab-04-rabbitmq-demo-orderly/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java Normal file
View File

@@ -0,0 +1 @@
package cn.iocoder.springboot.lab04.rabbitmqdemo;

36
lab-04/lab-04-rabbitmq-demo-orderly/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo10ProducerTest.java Normal file
View File

@@ -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();
}
}

BIN
lab-04/lab-04-rabbitmq-demo-orderly/target/classes/META-INF/lab-04-rabbitmq-demo-concurrency.kotlin_module Normal file
View File

Binary file not shown.

7
lab-04/lab-04-rabbitmq-demo-orderly/target/classes/application.yaml Normal file
View File

@@ -0,0 +1,7 @@
spring:
# RabbitMQ 配置项，对应 RabbitProperties 配置类
rabbitmq:
host: 127.0.0.1 # RabbitMQ 服务的地址
port: 5672 # RabbitMQ 服务的端口
username: guest # RabbitMQ 服务的账号
password: guest # RabbitMQ 服务的密码

BIN
lab-04/lab-04-rabbitmq-demo-orderly/target/test-classes/META-INF/lab-04-rabbitmq-demo-concurrency.kotlin_module Normal file
View File

Binary file not shown.

30
lab-04/lab-04-rabbitmq-demo-transaction/pom.xml Normal file
View File

@@ -0,0 +1,30 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.1.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>lab-04-rabbitmq-demo-transaction</artifactId>
<dependencies>
<!-- 实现对 RabbitMQ 的自动化配置 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!-- 方便等会写单元测试 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

15
lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/Application.java Normal file
View File

@@ -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);
}
}

61
lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java Normal file
View File

@@ -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);
}
}

21
lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo11Consumer.java Normal file
View File

@@ -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);
}
}

34
lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo11Message.java Normal file
View File

@@ -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 +
'}';
}
}

32
lab-04/lab-04-rabbitmq-demo-transaction/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo11Producer.java Normal file
View File

@@ -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);
}
}

7
lab-04/lab-04-rabbitmq-demo-transaction/src/main/resources/application.yaml Normal file
View File

@@ -0,0 +1,7 @@
spring:
# RabbitMQ 配置项，对应 RabbitProperties 配置类
rabbitmq:
host: 127.0.0.1 # RabbitMQ 服务的地址
port: 5672 # RabbitMQ 服务的端口
username: guest # RabbitMQ 服务的账号
password: guest # RabbitMQ 服务的密码

1
lab-04/lab-04-rabbitmq-demo-transaction/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/package-info.java Normal file
View File

@@ -0,0 +1 @@
package cn.iocoder.springboot.lab04.rabbitmqdemo;

28
lab-04/lab-04-rabbitmq-demo-transaction/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo11ProducerTest.java Normal file
View File

@@ -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();
}
}

7
lab-04/lab-04-rabbitmq-demo-transaction/target/classes/application.yaml Normal file
View File

@@ -0,0 +1,7 @@
spring:
# RabbitMQ 配置项，对应 RabbitProperties 配置类
rabbitmq:
host: 127.0.0.1 # RabbitMQ 服务的地址
port: 5672 # RabbitMQ 服务的端口
username: guest # RabbitMQ 服务的账号
password: guest # RabbitMQ 服务的密码

2
lab-04/pom.xml
View File

@@ -21,6 +21,8 @@
<module>lab-04-rabbitmq-demo-delay</module>
<module>lab-04-rabbitmq-demo-message-model</module>
<module>lab-04-rabbitmq-demo-concurrency</module>
<module>lab-04-rabbitmq-demo-orderly</module>
<module>lab-04-rabbitmq-demo-transaction</module>
</modules>
<dependencies>