一、概述
- 大多数运用中，可通过消息中间件来提升系统异步通讯、扩展解耦能力。
- 消息服务中的两个重要概念：
消息代理（message broker）和目的地（destination）
当消息发送者发送消息以后，将由消息代理接管。消息代理保证消息传递到指定目的地。 - 消息代理主要有两种形式的目的地：
1、队列（queue）：点对点通讯（point-to-point）
2、主题（topic）：发布（publish）/ 订阅（subscribe）消息通讯 - JMS（Java Message Service）消息服务
-- 基于JVM消息代理的规范。如ActivateMQ - AMQP（Advanced Message Queuing Protocol）
-- 高级消息队列，也是一种消息代理的规范。如RabbitMQ。
二、消息队列的使用场景
1、异步处理
注册用户后需要发送邮件和短信提示
（1）串行方式：将注册用户信息写入数据库后，发送注册邮件、再发送注册短信
（2）并行方式：将注册信息写入数据库后,发送邮件的同时,发送短信。
（3）消息队列：引入消息队列后，把发送邮件、短信等不是必须的业务逻辑异步处理。
2、应用解耦
在网上购物时，用户下单后,订单系统需要通知库存系统
（1）传统方式
缺点：
- 库存系统出故障，订单就会失败。
-
订单系统与库存系统高度耦合。
（2）消息队列
用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单。就算库存系统出现故障，消息系统也能保证订单消息的可靠传递。
3、流量削峰
秒杀活动，可能会因为流量太大而导致应用挂掉。一般在应用前端加入消息队列。
作用：
1、控制活动人数，超过一定阈值的订单直接丢弃。
2、可以缓解短时间的高流量压垮应用。
三、RabbitMQ简介
RabbitMQ是一个开源用Erlang语言编写的AMQP实现。
- Message（消息）
消息由消息头和消息体组成，消息体是不透明的；消息头由一系列属性组成，如routing-key（路由键）、priority（优先级）、delivery-mode（指出该消息可能需要持久化）等。 - Exchange（交换机）
用于接收生产者（publisher）发送的消息，并根据条件转发给服务器中的队列。
Exchange有四种类型：direct（默认）、fanout、topic、headers - Queue（消息队列）
保存消息直到被消费者（consumer）取走。 - Binding（绑定）
用于交换器与消息队列之间的关联。
Exchange和Queue之间的关联可以是多对多的。 - Publisher（生产者）
- Consumer（消费者）
- Connection（连接）
- Channel（信道）
- Virtual host（虚拟主机）
- Broker（消息队列服务器实体）
rabbitMQ
四、RabbitMQ运行机制
1、Direct Exchange
消息中的路由键（routing-key）如果和Binding中的binding-key一致，交换器就将消息发送到对应的队列中。
2、Fanout Exchange
每个发送到fanout类型的交换器的消息都会分到所有绑定的队列上去。
3、Topic Exchange
Topic Exchange增加了一个模糊匹配的功能，让routing-key与某个模式相匹配，符合匹配规则的队列便于交换机绑定。
通配符#匹配0个或多个字符，*匹配一个单词
五、spring boot集成RabbitMQ
1、 准备
-
创建交换器（Exchange）
smallbear.direct、smallbear.fanout、smallbear.topic
-
创建消息队列（Queue）
qq.music、qq.vedio、163.music、163.vedio、com
-
交换机绑定消息队列
（1） smallbear.direct
（2）smallbear.fanout
（3）smallbear.topic
2. 构建项目
- 添加依赖
<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>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
- 配置
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
- RabbitMQ配置
@Configuration
public class RabbitConfig {
/**
* 消息默认是通过序列化传到rabbitmq中的，这里设置成json格式
*/
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
}
3. 测试
- direct测试
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void directTest() {
Map<String, Object> map = new HashMap<>();
map.put("sender", "小明");
map.put("msg", "今晚出去玩不？");
rabbitTemplate.convertAndSend("smallbear.direct", "163.vedio", map);
}
@Test
public void directReceiveTest() {
Object o = rabbitTemplate.receiveAndConvert("163.vedio");
System.out.println(o); // {msg=今晚出去玩不？, sender=小明}
}
运行directTest()后可在163.vedio队列中看到以下结果
运行
directReceiveTest()获取163.vedio队列中的消息。
- fanout测试
@Data
@NoArgsConstructor
@AllArgsConstructor
public class Book {
private String name;
private String author;
private Double price;
}
@Test
public void fanoutTest() {
Book book = new Book("龙族", "江南", 133.32);
rabbitTemplate.convertAndSend("smallbear.fanout", "", book);
}
运行fanoutTest()，四个消息队列都接收到以下消息
- topic测试
@Test void topicTest() {
rabbitTemplate.convertAndSend("smallbear.topic", "google.com", "Google Chrome");
rabbitTemplate.convertAndSend("smallbear.topic", "baidu.com", "百度一下，你就知道");
}
运行topicTest()，com队列获取到两条消息
4. @RabbitListner监听
- 开启RabbitMQ
@EnableRabbit
@SpringBootApplication
public class RabbitmqDemoApplication {
public static void main(String[] args) {
SpringApplication.run(RabbitmqDemoApplication.class, args);
}
}
- 添加Rabbit监听器
@Data
public class User {
private String name;
private String telephone;
private String email;
}
@Service
public class RegistryService {
@Autowired
private MessageConverter messageConverter;
@RabbitListener(queues = "send.telephone")
public void sendTelephone(User user) {
System.out.println("新用户为" + user.getName() + "发送短信到" + user.getTelephone());
}
@RabbitListener(queues = "send.email")
public void sendEmail(Message message) {
System.out.println("获取头部信息:" + message.getMessageProperties());
User user = (User) messageConverter.fromMessage(message);
System.out.println("新用户为" + user.getName() + "发送邮件到" + user.getEmail());
}
}
- 编写Controller
@RestController
@RequestMapping("user")
public class UserController {
@Autowired
private RabbitTemplate rabbitTemplate;
@PostMapping("registry")
public String registry(User user) {
System.out.println("用户" + user.getName() + "注册成功");
// 新增队列send.email和send.telephone，并与smallbear.direct交换器绑定
rabbitTemplate.convertAndSend("smallbear.direct", "send.telephone", user);
rabbitTemplate.convertAndSend("smallbear.direct", "send.email", user);
return "SUCCESS";
}
}
-
测试
输出结果
用户小李注册成功
获取头部信息:MessageProperties [headers={__TypeId__=com.smallbear.rabbitmqdemo.entity.User}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=smallbear.direct, receivedRoutingKey=send.email, deliveryTag=1, consumerTag=amq.ctag-Hxt7ShNqlFOu6Ww-qlb1bA, consumerQueue=send.email]
新用户为小李发送邮件到12345678@qq.com
新用户为小李发送短信到13111111111
5. AmqpAdmin设置Queue、Exchange和Biinding
@Autowired
private AmqpAdmin amqpAdmin;
@Test void amqpAdminTest() {
amqpAdmin.declareExchange(new DirectExchange("exchange-demo"));
amqpAdmin.declareQueue(new Queue("queue-demo", true));
amqpAdmin.declareBinding(new Binding("queue-demo", Binding.DestinationType.QUEUE, "exchange-demo", "queue-demo", null));
}
