关于rabbitmq,你想知道的都在这里(上)
关于rabbitmq,你想知道的都在这里(上)#### rabbitmq在window上安装 依赖erlang环境,首先现在erlang
然后下载rabbitmq的windows版本
首先我们了解下消息队列是由交换机exchange和队列组合构成的,有三种形式
- 直连型:一个交换机关联一个队列,指定一个路由key,消息通过交换机名称和路由key发送到指定队列,发送一个,队列里面就多一个消息。
- 扇型:一个交换机关联多个队列。消息通过交换机名称发送,所有关联了这个交换机的队列都将收到消息,发送一个消息再N个消息队列产生N个一模一样的消息数据。
- 主题型:一个交换机根据规则关联多个队列。这种类型与扇型的很像,但是主题型会根据动态路由key决定消息的投递到哪个队列。这里的路由规则很像正则表达式。会根据事先设定的路由规则动态将消息投递到队列,可能投递到一个队列也可能投递到多个队列。
首先添加依赖:
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
yml文件指定rabbitmq连接信息
server:
port: 8021
spring:
#给项目来个名字
application:
name: rabbitmq-provider
#配置rabbitMq 服务器
rabbitmq:
host: localhost
port: 5672
# username: skywalker
# password: skywalker
# #虚拟host 可以不设置,使用server默认host
# virtual-host: skywalker-virtualhost
#确认消息已发送到交换机(Exchange)
publisher-confirm-type: correlated
#确认消息已发送到队列(Queue)
publisher-returns: true
注意:我们需要创建两个工程,一个生产者
producer
、一个消费者comsumer
,生产者用来生产消息,消费者用来消费生产者将消息投递到rabbitmq中的消息。两个工程中的pom依赖一样,yml也一样,只需要将
server.port
设置成不同的端口即可。这里我们将生产者设置为8021
端口,消费者设置为8022
端口。
直连型:
从上面的讲解中我们知道,有交换机exchange
,有队列queue
,有路由routing
,因此我们需要在生产者端将三者关联起来,然后发送消息,这样消费端才能收到消息。
绑定关联
@Configuration
public class DirectRabbitConfig {
public static String directRouting = "directRouting";
public static String directQueue = "directQueue";
public static String directExchange = "directExchange";
@Bean
public Queue DirectQueue() {
return new Queue(DirectRabbitConfig.directQueue,true); //true 是否持久
}
@Bean
DirectExchange DirectExchange() {
return new DirectExchange(DirectRabbitConfig.directExchange);
}
@Bean
Binding bindingDirect() {
// BindingBuilder.bind(队列A).to(交换机B).with(路由) 将队列A绑定到交换机B,使用路由C传递消息
return BindingBuilder.bind(DirectQueue()).to(DirectExchange()).with(directRouting);
}
}
发送消息
@Autowired
private RabbitTemplate rabbitTemplate; //使用RabbitTemplate,这提供了接收/发送等等方法
@GetMapping("/sendDirectMsg")
public String sendDirectMsg() {
Map<String,Object> map=new HashMap<String,Object>();
map.put("id",UUID.randomUUID().toString());
map.put("data","hello,i am direct msg!");
map.put("datetime",System.currentTimeMillis());
//交换机 路由 消息(发送消息的时候不需要管队列,因为队列已经在DirectRabbitConfig中配置了,队列应该是消费者关心的事情)
rabbitTemplate.convertAndSend(DirectRabbitConfig.directExchange, DirectRabbitConfig.directRouting, map);
return "ok";
}
注意:在发送完消息后,通过控制台
http://localhost:15672
(默认用户名密码都是guest
),查看是否产生了消息,如下界面:
接收消息
@Component
@RabbitListener(queues = "directQueue")//监听的队列名称 directQueue,不需要管路由和交换机,因为这些是生产者管理的事情。消费者只需要关心队列即可
public class DirectReceiver {
@RabbitHandler
public void handler(Map testMessage) {
System.out.println("directReceiver消费者收到消息 : " + testMessage.toString());
}
}
扇型
配置关联
@Configuration
public class FanoutRabbitConfig {
public static String fanoutQueue1="fanoutQueue1";
public static String fanoutQueue2="fanoutQueue2";
public static String fanoutQueue3="fanoutQueue3";
public static String fanoutExchange = "fanoutExchange";
@Bean
public Queue queue1() {
return new Queue(FanoutRabbitConfig.fanoutQueue1);
}
@Bean
public Queue queue2() {
return new Queue(FanoutRabbitConfig.fanoutQueue2);
}
@Bean
public Queue queue3() {
return new Queue(FanoutRabbitConfig.fanoutQueue3);
}
@Bean
FanoutExchange fanoutExchange() {
return new FanoutExchange(FanoutRabbitConfig.fanoutExchange);
}
@Bean
Binding bindingExchange1() {
//将队列fanoutQueue1 绑定到 fanoutExchange
return BindingBuilder.bind(queue1()).to(fanoutExchange());
}
@Bean
Binding bindingExchange2() {
return BindingBuilder.bind(queue2()).to(fanoutExchange());
}
@Bean
Binding bindingExchange3() {
return BindingBuilder.bind(queue3()).to(fanoutExchange());
}
}
发送消息
@GetMapping("/sendFanoutMsg")
public String sendFanoutMsg() {
Map<String, Object> map = new HashMap<String,Object>();
map.put("id",UUID.randomUUID().toString());
map.put("data","hello,i am fanout msg!");
map.put("datetime",System.currentTimeMillis());
rabbitTemplate.convertAndSend(FanoutRabbitConfig.fanoutExchange, null, map);//扇型不需要路由key,设置了也无效
return "ok";
}
接收消息
@Component
@RabbitListener(queues = "fanoutQueue1")
public class FanoutReceiver1 {
@RabbitHandler
public void handler(Map testMessage) {
System.out.println("FanoutReceiver1消费者收到消息 : " +testMessage.toString());
}
}
注意:扇型我们配置了3个队列,此处指列出了一个队列的名称,自行加2和3即可,使用postman发送消息后再控制台可以看到:
三个队列都接收到了消息,消费没有先后顺序。
主题型
配置
@Configuration
public class TopicRabbitConfig {
//绑定键
public static String topicQueue1 = "topicQueue1";
public static String topicQueue2 = "topicQueue2";
public static String topicExchange = "topicExchange";
public static String topicRoutingApple = "fruit.apple";
// * 表示1~n个字符 (必须出现的)
// # 表示0~n个字符 (可能不出现)
// 若队列绑定为#,则无视消息路由,接收所有消息
// 当*和#都未出现时,就相当于直连direct
public static String topicRoutingFruit = "fruit.#";
@Bean
public Queue topicQueue1() {
return new Queue(TopicRabbitConfig.topicQueue1);
}
@Bean
public Queue topicQueue2() {
return new Queue(TopicRabbitConfig.topicQueue2);
}
@Bean
TopicExchange exchange() {
return new TopicExchange(TopicRabbitConfig.topicExchange);
}
//只有携带路由key 为fruit.apple,才会分发到该队列
@Bean
Binding bindingExchangeMessage() {
return BindingBuilder.bind(topicQueue1()).to(exchange()).with(TopicRabbitConfig.topicRoutingApple);
}
//只要是消息携带的路由键是以fruit.开头,都会分发到该队列
@Bean
Binding bindingExchangeMessage2() {
return BindingBuilder.bind(topicQueue2()).to(exchange()).with(TopicRabbitConfig.topicRoutingFruit);
}
}
发送消息
@GetMapping("/sendTopicMsg1")
public String sendTopicMsg1() {
Map<String, Object> map = new HashMap<String,Object>();
map.put("id",UUID.randomUUID().toString());
map.put("data","hello,i am topic msg from apple!");
map.put("datetime",System.currentTimeMillis());
rabbitTemplate.convertAndSend(TopicRabbitConfig.topicExchange, TopicRabbitConfig.topicRoutingApple, map);
return "ok";
}
@GetMapping("/sendTopicMsg2")
public String sendTopicMsg2() {
Map<String, Object> map = new HashMap<String,Object>();
map.put("id",UUID.randomUUID().toString());
map.put("data","hello,i am topic msg from fruit!");
map.put("datetime",System.currentTimeMillis());
rabbitTemplate.convertAndSend(TopicRabbitConfig.topicExchange, "fruit.orange", map);
return "ok";
}
接收消息
@Component
@RabbitListener(queues = "topicQueue1")
public class Topic1Receiver {
@RabbitHandler
public void handler(Map testMessage) {
System.out.println("Topic1Receiver消费者收到消息 : " + testMessage.toString());
}
}
@Component
@RabbitListener(queues = "topicQueue2")
public class Topic2Receiver {
@RabbitHandler
public void handler(Map testMessage) {
System.out.println("Topic2Receiver消费者收到消息 : " + testMessage.toString());
}
}
好了,这边基本介绍就到这里了,回顾下,我们讲解了消息队列的三种形式,及各自的应用。另外topic主题型队列是一种特殊的队列。下回我们扩展讲解下消息的持久化,消息回调(生产端和消费端)