一. AMQP协议
什么是AMQP协议
AMQP(Advanced Message Queuing Protocol,高级消息队列协议):它是进程之间传递异步消息的网络协议
AMQP工作过程
发布者通过发布消息,通过交换机,交换机根据路由规则将收到的消息分发交换机绑定的下消息队列,最后AMQP代理将消息推送给订阅了此队列的消费者
或消费者按照需求自行获取。
二. RabbitMQ简介
RabbitMQ是通过Erlang语言基于AMQP协议编写的消息中间件,它在分布式系统中可以解应用耦合、流量削峰、异步消息等问题。它有两个特性
队列排队和异步
- 应用解耦:多个个应用程序之间可通过RabbitMQ作为媒介,两个应用不再粘连,实现解耦;
- 异步消息:多个应用可通过RabbitMQ进行消息传递;
- 流量削峰:在高并发情况下,可以通过RabbitMQ的队列特性实现流量削峰;
- 应用场景:
- 应用到队列特性的应用场景: 排序算法、秒杀活动。
- 应用到异步特性的应用场景: 消息分发、异步处理、数据同步、处理耗时任务。
三.springBoot整合RabbitMQ
生产者端发送消息
pom文件
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId><version>2.6.3</version></dependency>
yml文件
spring:application:name: producerrabbitmq:host: xxxusername: adminpassword: admin
配置类,需要返回一个Queue,org.springframework.amqp.core.Queue下的Queue对象
@Configuration
public class RabbitMqConfig {@Beanprotected Queue queue(){return new Queue("myQueue");}
}
使用RabbitMQ发送消息,注入AmqpTemplate,调用convertAndSend()方法
class ProducerApplicationTests {@Autowiredprivate AmqpTemplate amqpTemplate;@Testvoid send() {for (int i = 0; i < 10; i++) {amqpTemplate.convertAndSend("myQueue","这是发送的消息");System.out.println("发送成功!");}}}
消费端接收消息
配置同生产端,不需要配置RabbitMqConfig,接收消息时只需要使用注解RabbitMqConfig,queues属性绑定相应的队列即可。
@Component
public class ReceiveService {@RabbitListener(queues = "myQueue")public void test01(String msg){System.out.println("接收到消息1" + msg);}@RabbitListener(queues = "myQueue")public void test02(String msg){System.out.println("接收到消息2" + msg);}@RabbitListener(queues = "myQueue")public void test03(String msg){System.out.println("接收到消息3" + msg);}
}
四.交换器(四种)
Direct Exchange:直连交换器
它是RabbitMQ的默认交换器,给指定队列发消息,绑定该消息队列的消费者一次获取消息
实战:
/** 生产者发送消息,发送10个消息*/
@SpringBootTest
class ProducerApplicationTests {@Autowiredprivate AmqpTemplate amqpTemplate;@Testvoid send() {for (int i = 0; i < 10; i++) {amqpTemplate.convertAndSend("myQueue","这是发送的消息");System.out.println("发送成功!");}}}
/** 接收消息*/
@Component
public class ReceiveService {@RabbitListener(queues = "myQueue")public void test01(String msg){System.out.println("接收到消息1" + msg);}@RabbitListener(queues = "myQueue")public void test02(String msg){System.out.println("接收到消息2" + msg);}@RabbitListener(queues = "myQueue")public void test03(String msg){System.out.println("接收到消息3" + msg);}
}
结果:可以看到1、2、3依次接收消息
接收到消息1这是发送的消息
接收到消息2这是发送的消息
接收到消息3这是发送的消息
接收到消息2这是发送的消息
接收到消息3这是发送的消息
接收到消息1这是发送的消息
接收到消息3这是发送的消息
接收到消息1这是发送的消息
接收到消息2这是发送的消息
接收到消息1这是发送的消息
Fanout Exchange:扇形交换器
绑定该交换器的所有队列都可以接收到消息,扇形交换机将消息广播到所有与之绑定的队列。无论消息的路由键是什么,扇形交换机都会将消息发送到所有绑定的队列中。这种类型的交换机常用于实现发布-订阅模式,将消息广播给多个消费者。
实战
/** 绑定*/
/** Fanout Exchange*/
@Bean
public Queue FanoutExchangeQueue1(){return new Queue("fanoutExchangeQueue1");}
@Bean
public Queue FanoutExchangeQueue2(){return new Queue("fanoutExchangeQueue2");}
@Bean
public FanoutExchange fanoutExchange(){return new FanoutExchange("amq.fanout");}
@Bean
public Binding FanoutExchangeBinding1(Queue FanoutExchangeQueue1,FanoutExchange fanoutExchange){return BindingBuilder.bind(FanoutExchangeQueue1).to(fanoutExchange);}
@Bean
public Binding FanoutExchangeBinding2(Queue FanoutExchangeQueue2,FanoutExchange fanoutExchange){return BindingBuilder.bind(FanoutExchangeQueue2).to(fanoutExchange);}
/** 生产者发送消息*/@Testvoid sendByFanoutExchange() {amqpTemplate.convertAndSend("amq.fanout","key","这是发送到的消息");System.out.println("发送成功!");}
/** 消费者 Direct Exchange*/@RabbitListener(queues = "fanoutExchangeQueue1")public void test04(String msg){System.out.println("接收到消息4" + msg);}@RabbitListener(queues = "fanoutExchangeQueue2")public void test05(String msg){System.out.println("接收到消息5" + msg);}
结果:每一个绑定到Fanout Exchange上的队列都可以接收到消息
接收到消息4这是发送到的消息
接收到消息5这是发送到的消息
Topic Exchange:主题交换器
允许在路由键中设置匹配规则:'*‘代表一个字母两个’.'之间的内容;‘#’代表0或多个字符;
实战
/** 绑定*/@Beanpublic Queue topicExchangeQueue1(){return new Queue("topicExchangeQueue1");}@Beanpublic Queue topicExchangeQueue2(){return new Queue("topicExchangeQueue2");}@Beanpublic TopicExchange topicExchange(){return new TopicExchange("amq.topic");}@Beanpublic Binding TopicExchangeToQueue1(Queue topicExchangeQueue1,TopicExchange topicExchange){return BindingBuilder.bind(topicExchangeQueue1).to(topicExchange).with("com.shaoby.*");}@Beanpublic Binding TopicExchangeToQueue2(Queue topicExchangeQueue2,TopicExchange topicExchange){return BindingBuilder.bind(topicExchangeQueue2).to(topicExchange).with("com.shaoby.test.#");}
/**生产者发送消息*//** key为com.shaoby.test*/@Testvoid sendByTopicExchange() {amqpTemplate.convertAndSend("amq.topic","com.shaoby.test","这是发送到的消息");System.out.println("发送成功!");}/** key为com.shaoby.test.a*/@Testvoid sendByTopicExchange() {amqpTemplate.convertAndSend("amq.topic","com.shaoby.test.a.b","这是发送到的消息");System.out.println("发送成功!");}
/**消费者接收消息*//**Topic Exchange*/@RabbitListener(queues = "topicExchangeQueue1")public void test06(String msg){System.out.println("接收到消息6" + msg);}@RabbitListener(queues = "topicExchangeQueue2")public void test07(String msg){System.out.println("接收到消息7" + msg);}
结果:
路由key为com.shaoby.test都能接收到消息,com.shaoby.test.a.b只有topicExchangeQueue2能接收到消息
Header Exchange:首部交换器
绑定:
/** Header Exchange*/
@Bean
public Queue headerExchangeQueue1(){return new Queue("headerExchangeQueue1");}@Bean
public Queue headerExchangeQueue2(){return new Queue("headerExchangeQueue2");}
@Bean
public HeadersExchange headersExchange(){return new HeadersExchange("amp.header");}
@Bean
public Binding headExchangeToQueue1(Queue headerExchangeQueue1,HeadersExchange headersExchange){HashMap<String, Object> map = new HashMap<>();map.put("type","OK");map.put("status","200");return BindingBuilder.bind(headerExchangeQueue1).to(headersExchange).whereAll(map).match();}
@Bean
public Binding headExchangeToQueue2(Queue headerExchangeQueue2,HeadersExchange headersExchange){HashMap<String, Object> map = new HashMap<>();map.put("type","error");map.put("status","500");return BindingBuilder.bind(headerExchangeQueue2).to(headersExchange).whereAll(map).match();}
/** 生产者发送消息*/
@Testvoid sendByHeadExchange() {Map<String, Object> headers = new HashMap<>();headers.put("type","OK");headers.put("status","200");String message = "这是发送到的消息";MessageProperties messageProperties = new MessageProperties();headers.forEach(messageProperties::setHeader);Message msg = new Message(message.getBytes(), messageProperties);amqpTemplate.convertAndSend("amp.header",null, msg);System.out.println("发送成功!");}
@RabbitListener(queues = "headerExchangeQueue1")public void test08(Message msg){System.out.println("接收到消息8:" + msg.toString());}@RabbitListener(queues = "headerExchangeQueue2")public void test09(Message msg){System.out.println("接收到消息9:" + msg.toString());}
结果:只有匹配上header才能收到消息
接收到消息8:(Body:'[B@a7b38a8(byte[24])' MessageProperties [headers={type=OK, status=200}, contentType=application/octet-stream, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=amp.header, receivedRoutingKey=, deliveryTag=2, consumerTag=amq.ctag-1WTdKW4n_rAEdJUosQD7bg, consumerQueue=headerExchangeQueue1])