概述
-
大多应用中，可通过消息服务中间件来提升系统异步通信、扩展解耦能力
-
消息服务中两个重要概念：消息代理（message broker）和目的地（destination），当消息发送者发送消息以后，将由消息代理接管，消息代理保证消息传递到指定目的地
-
消息队列主要有两种形式的目的地
- 队列（queue）：点对点消息通信（point-to-point）
- 主题（topic）：发布（publish）/订阅（subscribe）消息通信
-
点对点式：
- 消息发送者发送消息，消息代理将其放入一个队列中，消息接收者从队列中获取消息内容，消息读取后被移出队列
- 消息只有唯一的发送者和接受者，但并不是说只能有一个接收者
-
发布订阅式
- 发送者（发布者）发送消息到主题，多个接收者（订阅者）监听（订阅）这个主题，那么就会在消息到达时同时收到消息
-
JMS（Java Message Service）Java消息服务：
- 基于JVM消息代理的规范，ActiveMQ、HornetMQ是JMS的实现
-
AMQP（Advanced Message Queuing Protocol）
- 高级消息队列协议，也是一个消息代理的规范，兼容JMS
- RabbitMQ是AMOP的实现
-
JMS对比AMQP
| JMS | AMQP | |
|---|---|---|
| 定义 | Java API | 网络线级协议 |
| 跨语言 | 否 | 是 |
| 跨平台 | 否 | 是 |
| Model | 提供两种消息模型：1.Peer-2-Peer 2.Pub/Sub | 提供五种消息模型：1.direct exchange 2.fanout exchange 3.topic exchange 4.headers exchange 5.system exchange |
| 支持消息类型 | 多种消息类型：TextMessage、MapMessage、ByteMessage、StreamMessage、ObjectMessage、Message（只有消息头和属性） | byte[]，当实际应用时，有复杂的消息，可以将消息序列化后发送 |
| 综合评价 | JMS定义了Java API层面的标准，在Java体系中，多个client均可以通过JMS进行交互，不需要应用修改代码，但是其对跨平台的支持较差 | AMQP定义了wire-level层的协议标准，天然具有跨平台、跨语言的特性 |
RabbitMQ简介
RabbitMQ是一个由erlang开发的AMQP的开源实现
核心概念
Message
消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）
Publisher
消息的生产者，也是一个向交换器发布消息的客户端程序
Exchange
交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列，Exchange有4种类型：direct（默认，点对点单播）、fanout（广播）、topic（主题订阅）和headers，不同类型的Exchange转发消息的策略有所区别
Queue
消息队列，用来保存消息直到发送给消费者，它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列，消息一直在队列里面，等待消费者连接到这个队列将其取走
Binding
绑定，用于消息队列和交换器之间的关联，一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表，Exchange和Queue的绑定可以是多对多的关系
Connection
网络连接
Channel
信道，多路复用连接中的一条独立的双向数据流通道，通过一条条信道以复用一条TCP连接，节省资源和开销
Consumer
消息的消费者，表示一个从消息队列中取得消息的客户端应用程序
Virtual Host
虚拟主机，表示一批交换器、消息队列和相关对象，每个vhost实质上就是一个mini版的RabbitMQ服务器，拥有自己的队列、交换器、绑定个权限机制。vhost是AMQP概念的基础，必须在连接时指定，RabbitMQ默认的vhost是/
Broker
表示消息队列服务器实体
示意图：
SpringBoot整合RabbitMQ
环境准备
-
Docker下载RabbitMQ镜像启动容器(5672是客户端和rabbitmq服务器通信端口，15672是web页面访问端口)
docker pull rabbitmq docker run -d -p 5672:5672 -p 15672:15672 rabbitmq浏览器访问IP:15672登录(默认账户名密码为guest)显示：
添加交换器
添加消息队列
交换器与队列绑定
exchange.diret与exchange.fanout绑定的队列：
exchange.topic绑定的队列：
路由键中#表示匹配零个或多个单词，*表示匹配一个单词
-
创建项目
-
引入依赖如下：
<dependencies> <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.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit-test</artifactId> <scope>test</scope> </dependency> </dependencies> -
配置application.properties
spring.rabbitmq.host=主机地址 spring.rabbitmq.username=guest spring.rabbitmq.password=guest #spring.rabbitmq.port=端口号，默认为5672 #spring.rabbitmq.virtual-host=指定vhost，默认为/
-
测试RabbitTemplate发送和接受消息
-
测试单播
@Autowired RabbitTemplate rabbitTemplate; @Test public void testDirect() { // 需要自己构造一个Message，定义消息体内容和消息头 // rabbitTemplate.send(exchange,routeKey,message); // object默认作为消息体，只需要传入要发送的对象，自动序列化发送给rabbitmq // rabbitTemplate.convertAndSend(exchange,routeKey,object); Map<String, Object> map = new HashMap<>(); map.put("msg", "2020加油"); map.put("data", Arrays.asList("helloworld", "666", true)); rabbitTemplate.convertAndSend("exchange.direct", "atguigu", map); }运行代码可以看到队列atguigu里收到了一条消息
但是消息体内容是经过序列化之后的结果，不方便阅读，这是因为rabbitmq默认采用jdk的序列化机制，我们可以改为json的序列化机制，创建MyAMQPConfig配置类
@Configuration public class MyAMQPConfig { @Bean public MessageConverter messageConverter(){ return new Jackson2JsonMessageConverter(); } }再次运行代码
-
测试广播
@Test public void testFanout() { rabbitTemplate.convertAndSend("exchange.fanout", "", new Book("三国演义", "罗贯中")); }运行代码可以看到每个队列都收到了一条消息
-
测试主题订阅
@Test public void testTopic(){ rabbitTemplate.convertAndSend("exchange.topic","atguigu",new Book("红楼梦","曹雪芹")); }运行代码可以看到与exchange.topic绑定的路由键相对应的队列都收到了一条消息
-
测试接受消息
@Test public void receive() { Object o = rabbitTemplate.receiveAndConvert("atguigu"); System.out.println(o.getClass()); System.out.println(o); }运行代码输出
class com.chenpeng.amqp.bean.Book book{bookName='红楼梦', author='曹雪芹'}
测试AmqpAdmin创建Exchange、Queue和Binding
AmqpAdmin：RabbitMQ系统管理功能组件
@Autowired AmqpAdmin amqpAdmin;
创建交换器
amqpAdmin.declareExchange(new DirectExchange("amqpadmin.direct"));
创建队列并指定为持久化
amqpAdmin.declareQueue(new Queue("amqpadmin.queue", true));
创建绑定关系
amqpAdmin.declareBinding(
new Binding(
"amqpadmin.queue",
Binding.DestinationType.QUEUE,
"amqpadmin.direct",
"amqpadmin.2020",
null));
- destination：目的地，可指定队列
- destinationType：目的地类型，QUEUE, EXCHANGE
- exchange：指定交换器
- routingKey：路由键
- arguments：参数头
运行代码可以看到
@EnableRabbit + @RabbitListener监听消息队列的内容
@EnableRabbit开启监听，加载springboot启动类上，@RabbitListener监听消息队列
@Service
public class BookService {
@RabbitListener(queues = "atguigu")
public void receive(Book book){
System.out.println("收到消息:"+book);
}
@RabbitListener(queues = "atguigu.emps")
public void receiveMessage(Message message){
System.out.println(message.getBody());
System.out.println(message.getMessageProperties());
}
}