SpringCloud之集成RabbitMQ_springcloud中封装一个rabbitmq工具模块

以订单系统和库存系统为例，使用api接口会使得服务之间的耦合增大

通过整合RabbitMQ实现服务之间的解耦

一、引入依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
1
2
3
4

二、配置文件

spring :
	rabbitmq:
	    host: 192.168.184.128
	    port: 5672
	    username: admin
	    password: admin
	    virtual-host: testhost
	    listener:
	      simple:
	        #开启手动确认消息
	        acknowledge-mode: manual
	        concurrency: 3 #消费者最小数量
	        max-concurrency: 10 # 消费者最大数量
	        retry:
	          #开启消费重试
	          enabled: true
	          #重试三次
	          max-attempts: 3
	          #间隔五秒
	          initial-interval: 5000
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

三、消息通信

3.1 绑定队列

1. 生产者

@Component
public class MqSend {
	@Autowired
    private AmqpTemplate amqpTemplate;
    
    public void process(){
        amqpTemplate.convertAndSend("队列名称","内容");
    }
}
1
2
3
4
5
6
7
8
9

2. 消费者

@Component
public class MqReceiver {
	//使用queuesToDeclare，当队列不存在时自动创建队列
    @RabbitListener(queuesToDeclare =@Queue("队列名称"))
    public void process(String message){
        System.out.println(message);
    }
}
1
2
3
4
5
6
7
8

3.2 绑定交换机

注意：rabbitMQ上需要先添加对应的交换机和队列

1. 生产者

public class MqSend {
	@Autowired
    private AmqpTemplate amqpTemplate;
    
    public void process(){
        amqpTemplate.convertAndSend("交换机名称","队列名称","内容");
    }
}
1
2
3
4
5
6
7
8

2. 消费者

```java
@Component
public class MqReceiver {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("队列名称"),
            exchange = @Exchange("交换机名称")
    ))
    public void process(String message){
        System.out.println(message);
    }
}
1
2
3
4
5
6
7
8
9
10
11

3.3 绑定routing Key

注意：rabbitMQ上需要先添加对应的交换机和队列

1. 生产者

public class MqSend {
	@Autowired
    private AmqpTemplate amqpTemplate;
    
    public void process(){
        amqpTemplate.convertAndSend("交换机名称","routing key","内容");
    }
}
1
2
3
4
5
6
7
8

2. 消费者

@Component
public class MqReceiver {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("队列名称"),
            exchange = @Exchange("交换机名称"),
            key = "routing key"
    ))
    public void process(String message){
        System.out.println(message);
    }
}
1
2
3
4
5
6
7
8
9
10
11

四、流量削峰

在消费者注解加上concurrency属性并且设置对应的最大处理消息数

@RabbitListener(bindings = @QueueBinding(
            value = @Queue("goodsQueue"),
            exchange = @Exchange("goodsExchange"),
            key = "buy"
    ),concurrency = "5")
public void process(String message){
}
1
2
3
4
5
6
7

注意：设置的处理消息数需要配置文件设置的范围内