7.Spring Boot中使用RabbitMQ_spring boot 使用rabbitmq

目录

---

RabbitMQ专栏目录（点击进入…）

---

---

Spring Boot中使用RabbitMQ

1.导入启动器依赖

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

2.配置文件（application.yml）

server:
  port: 80
  
spring:
  #给项目取个名字
  application:
    name: rabbitmq-provider
  #配置rabbitMq 服务器
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: root
    password: root
    #虚拟主机host 可以不设置,使用server默认host
    virtual-host: TestHost
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

3.初始化队列、交换机。并将队列绑定至交换机（DirectRabbitConfig.java）

使用direct exchange（直连型交换机）创建。

交换机：exchange
路由key：routingkey
队列：queue

exchange和queue是需要绑定在一起的，然后消息发送到exchange再由exchange通过routingkey发送到对应的队列中

@Configuration
public class DirectRabbitConfig {
	// 队列名：TestDirectQueue
	@Bean
	public Queue TestDirectQueue() {
		// 一般设置一下队列的持久化就好,其余三个就是默认是false、false、null
		return new Queue("TestDirectQueue", true);
	}

	// Direct交换机名：TestDirectExchange
	@Bean
	DirectExchange TestDirectExchange() {
		return new DirectExchange("TestDirectExchange", true, false);
	}

	// 绑定：将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
	@Bean
	Binding bindingDirect() {
		return BindingBuilder.bind(TestDirectQueue()) // 队列
						.to(TestDirectExchange())  // 交换机
						.with("TestDirectRouting"); // 绑定键
	}

	@Bean
	DirectExchange lonelyDirectExchange() {
		return new DirectExchange("lonelyDirectExchange");
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

Queue队列

// 创建一个队列是持久的、非独占的和非自动删除的
public Queue(String name) {
	this(name, true, false, false);
}

// 创建一个新的队列，给定一个名称和持久性标志。队列非独占和非自动删除
public Queue(String name, boolean durable) {
	this(name, durable, false, false, null);
}

// 构建一个新的队列，给定队列名称、持久性、独占和自动删除标志
public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete) {
	this(name, durable, exclusive, autoDelete, null);
}

// 构建一个新的队列，给定一个名称、持久性标志和自动删除标志和参数
public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete,
			@Nullable Map<String, Object> arguments) {
	super(arguments);
	Assert.notNull(name, "'name' cannot be null");
	this.name = name;
	this.actualName = StringUtils.hasText(name) ? name : (Base64UrlNamingStrategy.DEFAULT.generateName() + "_awaiting_declaration");
	this.durable = durable;
	this.exclusive = exclusive;
	this.autoDelete = autoDelete;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

AbstractExchange（抽象类）

package org.springframework.amqp.core;

public abstract class AbstractExchange extends AbstractDeclarable implements Exchange {

    // 构建一个新的持久的、非自动删除的与提供的名称的交换机
	public AbstractExchange(String name) {
		this(name, true, false);
	}
	
    // 构建一个新的交换。给定一个名称、持久性标志、自动删除标志
	public AbstractExchange(String name, boolean durable, boolean autoDelete) {
		this(name, durable, autoDelete, null);
	}
	
    // 构建一个新的交换。给定一个名称、持久性标志和自动删除标志和参数
	public AbstractExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
		super(arguments);
		this.name = name;
		this.durable = durable;
		this.autoDelete = autoDelete;
	}
	
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

FanoutExchange（扇形交换机）、DirectExchange（直连交换机）、TopicExchange（主题交换机）、HeadersExchange（头交换机）都继承至AbstractExchange抽象类

这几个交换机都是使用super调用器父类构造方法构建交换机

public FanoutExchange(String name) {
	super(name);
}

public FanoutExchange(String name, boolean durable, boolean autoDelete) {
	super(name, durable, autoDelete);
}

public FanoutExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
	super(name, durable, autoDelete, arguments);
}
1
2
3
4
5
6
7
8
9
10
11

Binding

Binding在Spring-AMQP下core中，生产者创建消息发送至exchange，消费者从队列中消费消息，队列与交换器的绑定关系便是由Binding来表示的

public class Binding extends AbstractDeclarable {

	public Binding(String destination, DestinationType destinationType, String exchange, String routingKey,@Nullable Map<String, Object> arguments) {
		super(arguments);
		this.destination = destination; // 队列
		this.destinationType = destinationType; // 交换机类型
		this.exchange = exchange; // 交换机
		this.routingKey = routingKey; // 路由键
	}

}
1
2
3
4
5
6
7
8
9
10
11

队列、交换机绑定（BindingBuilder）

使用BindingBuilder实例化一个Binding，其主要作用是：在配置时更加快速的创建绑定关系

Binding binding = BindingBuilder.bind(queue).to(exchange).with(routing_key);
1

将一个queue绑定到exchange并用routing_key作为绑定键
bind()返回的是一个DestinationConfigurer，根据参数类型是Queue还是Exchange选择调用方法

public static DestinationConfigurer bind(Queue queue) {
	return new DestinationConfigurer(queue.getName(), DestinationType.QUEUE);
}

public static DestinationConfigurer bind(Exchange exchange) {
	return new DestinationConfigurer(exchange.getName(), DestinationType.EXCHANGE);
}
1
2
3
4
5
6
7

DestinationConfigurer类实现了不同Exchange的绑定，用to(Exchange_Type)用参数exchange类型来创建对应的DirectExchangeRoutingKeyConfigurer，最后使用with绑定routing_key，返回一个Binding实例

4.Controller（控制器）

@RestController
public class SendMessageController {
	//使用RabbitTemplate,这提供了接收/发送等等方法
    @Autowired
    private RabbitTemplate rabbitTemplate;  
 
    @GetMapping("/sendDirectMessage")
    public String sendDirectMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "test message, hello!";
        String createTime = LocalDateTime.now()
				.format(DateTimeFormatter
				.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String,Object> map=new HashMap<>();
        map.put("messageId",messageId);
        map.put("messageData",messageData);
        map.put("createTime",createTime);
        //将消息携带绑定键值：TestDirectRouting发送到交换机TestDirectExchange
        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);
        return "ok";
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

@RestController
public class DirectRabbitController {

    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    @GetMapping("/sendDirectMessage/{message}")
    public void sendDirectMessage(@PathVariable("message") String message){
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = message;
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String,Object> map=new HashMap<>();
        map.put("messageId",messageId);
        map.put("messageData",messageData);
        map.put("createTime",createTime);
        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);
    }
 
    @RabbitListener(queues = "TestDirectQueue")
    public void consumerDirectMessage(Map map){
        System.out.println("DirectReceiver消费者收到消息  : " + map.toString());
    }
    
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24