RabbitMQ入门

什么是消息队列

MQ全称为Message Queue，即消息队列。“消息队列”是在消息的传输过程中保存消息的容器。它是典型的：生产者、消费者模型。生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，这样就实现了生产者和消费者的解耦

常见MQ产品

ActiveMQ：基于JMS
RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好
RocketMQ：基于JMS，阿里巴巴产品，目前交由Apache基金会
Kafka：分布式消息系统，高吞吐量

MQ是消息通信的模型，并发具体实现

实现MQ的有两种主流方式：AMQP、JMS。

两者间的区别和联系：

JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式
JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
JMS规定了两种消息模型；而AMQP的消息模型更加丰富

JMS(Java Message Service)：

基于JVM消息代理的规范。ActiveMQ、HornetMQ是JMS实现

AMQP(Advanced Message Queuing Protocol)

高级消息队列协议，也是一个消息代理的规范，兼容JMS

RabbitMQ是AMQP的实现

在介绍RabbitMQ之前，我们先来看下面一个电商项目的场景：

商品的原始数据保存在数据库中，增删改查都在数据库中完成。
搜索服务数据来源是索引库（Elasticsearch），如果数据库商品发生变化，索引库数据不能及时更新。
商品详情做了页面静态化处理，静态页面数据也不会随着数据库商品更新而变化。

如果我们在后台修改了商品的价格，搜索页面和商品详情页显示的依然是旧的价格，这样显然不对。该如何解决？

我们可能会想到这么做：

方案1：每当后台对商品做增删改操作，同时修改索引库数据及更新静态页面。
方案2：搜索服务和商品页面静态化服务对外提供操作接口，后台在商品增删改后，调用接口。

这两种方案都有个严重的问题：就是代码耦合，后台服务中需要嵌入搜索和商品页面服务，违背了微服务的独立原则。

这时，我们就会采用另外一种解决办法，那就是消息队列！

商品服务对商品增删改以后，无需去操作索引库和静态页面，只需向MQ发送一条消息（比如包含商品id的消息），也不关心消息被谁接收。搜索服务和静态页面服务监听MQ，接收消息，然后分别去处理索引库和静态页面（根据商品id去更新索引库和商品详情静态页面）。

大多应用中，可通过消息服务中间件来提升系统异步通信、扩展解耦能力

消息服务中两个重要概念：

==消息代理（message broker）== 和==目的地（destination）==

当消息发送者发送消息后，将有消息代理接管，消息代理保证消息传递到指定目的地。

消息队列主要有两种目的地

队列（queue）：点对点消息通信（point-to-point）

主题（topic）：发布（publish）/订阅（subscribe）消息通信

点对点式：

消息发送者发送消息，消息代理将其放入一个队列中，消息接收者从队列中获取消息内容，消息读取后被移除队列
消息只有唯一的发送者和接受者，但并不是说只能有一个接收者

发布订阅式：

发送者（发布站）发送消息到主题，多个接收者(订阅者)监听(订阅)这个主题，那么就会在消息到达同时收到消息

Spring支持

spring-jms提供了对JMS的支持

spring-rabbit提供了对AMQP的支持

需要ConnectionFactory的实现来连接消息代理

提供JmsTemplate、RabbitTemplate来发送消息

@JsmListener（JMS）、@RabbitListener（AMQP）注解在方法上监听消息代理发布的消息

@EnableJms、@EnableRabbit开启支持

Spring Boot自动配置

JmsAutoConfiguration
RabbitAutoConfiguration

RabbitMQ简介
RabbitMQ是一个由erlang开发的AMQP(Advanved Message Queue Protocol)的开源实现

RabbitMQ的工作原理
下图是RabbitMQ的基本结构：

核心概念
Message
消息，消息是不具名的，它由消息头和消息体组成。消息体是透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键），priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等

Publisher
消息的生产者，也是一个向交换器发布消息的客户端应用程序。

Exchange
交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。

Exchange有4种类型：direct默认 fanout，topic，headers，不同类型的Exchange转发消息的策略有所区别

Queue
消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或者多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。

Binding
绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。支持多对多一对多

Connection
网络连接，比如一个TCP连接。

Channel
信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接，AMQP命令都是通过信道发出去的，不管是发布消息，订阅队列，还是接收消息，这些动作都是通过信道完成，因为对于操作系统来说建立和销毁TCP都是非常昂贵的开销，所以引入了信道的概念，已复用一条TCP连接。

Consumer
消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。

Virtual Host
虚拟主机，表示一批交换器，消息队列，和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域名。每个vhost本质上就是一个mini版的RabbitMQ服务器，拥有自己的队列，交换器，绑定和权限机制。vHost是AMQP概念的基础，必须在连接时指定，RabbitMQ默认的vhost是/。

Broker
表示消息队列服务器实体

RabbitMQ运行机制

生产者发送消息流程：

1、生产者和Broker建立TCP连接。

2、生产者和Broker建立通道。

3、生产者通过通道消息发送给Broker，由Exchange将消息进行转发。

4、Exchange将消息转发到指定的Queue（队列）

消费者接收消息流程：

1、消费者和Broker建立TCP连接

2、消费者和Broker建立通道

3、消费者监听指定的Queue（队列）

4、当有消息到达Queue时Broker默认将消息推送给消费者。

5、消费者接收到消息。

6、ack回复

AMQP中的消息路由
AMQP中消息的路由过程和Java开发者熟悉的JMS存在一些差别，AMQP中增加了Exchange和Binding的角色。生产者把消息发布到Exc上，消息最终到达队列并被消费者接收，而Binding决定交换器的消息应该发送到那个队列。

Exchange类型

Exchange分发消息时根据类型的不同分发策略有区别，目前共四种类型：direct，fanout，topic，headers。headers匹配AMQP消息的header而不是路由键，headers交换器和direct交换器完全一致，但性能差很多，目前几乎用不到了，所以直接看另外三种类型

RabbitMQ整合

引入spring-boot-starter-amqp
application.yml配置
测试rabbitMQ
1. AmqpAdmin:管理组件
2. RabbitTemplate：消息发送处理组件

maven

//maven引入 spring-boot-starter-amqp
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

启动类

/**
 * 自动配置
 * 1.RabbitAutoConfiguration
 * 2.有自动配置了连接工厂ConnectionFactory
 * 3.RabbitProperties 封装了RabbitMQ的配置
 * 4.RabbitTemplate：给Rabbit发送和接收消息
 * 5.AmqpAdmin RabbitMQ系统管理功能组件
 */
@EnableRabbit       //开启RabbitMQ
@SpringBootApplication
public class Springboot02AmqpApplication {

    public static void main(String[] args) {
        SpringApplication.run(Springboot02AmqpApplication.class, args);
    }

}

yml配置

server:
  port: ***
spring:
  rabbitmq:
    host: ******
    port: ****
    username: ***
    password: ***

实体类

@Data
public class Book {
    private Integer id;
    private String name;
    public Book() {
    }
    public Book(Integer id, String name) {
        this.id = id;
        this.name = name;
    }
}

Config配置

@Configuration
public class MyAMPQConfig {
    @Bean   //将自定义的消息类序列化成json格式
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}

发送以及接收消息

@RunWith(SpringRunner.class)
@SpringBootTest
class Springboot02AmqpApplicationTests {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private AmqpAdmin amqpAdmin;
    @Test
    public void createExchange(){
//        amqpAdmin.declareExchange(new DirectExchange("amqp.exchange11"));
        amqpAdmin.deleteExchange("amqp.exchange11");
    }

    /**
     * 单播
     */
    @Test
    public void direct() {
        //message需要自己定义
        //rabbitTemplate.send(exchange,routekey,message);
        //只需要传入要发送的对象，自动序列化给rabbitmq
        //rabbitTemplate.convertAndSend(exchange,routeKey,object);
//        Map<String,Object> map=new HashMap();
//        map.put("msg","这是一个消息");
//        map.put("date","123");
//        rabbitTemplate.convertAndSend("exchange.direct","atguigu.news",map);
        Book book = new Book();
        book.setId(1);
        book.setName("王毅辉");
        rabbitTemplate.convertAndSend("exchange.direct","atguigu.news",book);
    }

    /**
     * 接收消息
     */
    @Test
    public void receive(){
        Object o = rabbitTemplate.receiveAndConvert("atguigu.news");
        System.out.println(o.getClass());
    }

    /**
     * 发送消息
     */
    @Test
    public void sendMsg(){
        rabbitTemplate.convertAndSend("exchange.fanout","",new Book(1,"曹雪芹"));
    }
}

在Service中监听消息

@RabbitListener(queues = “监听的队列名”)

@Service
public class BookService {
    @RabbitListener(queues = "atguigu.news")
    public void receive(Book book){
        System.out.println("收到消息"+book);
    }
    @RabbitListener(queues = "atguigu")
    public void receive2(Message message){
        System.out.println(message);
        System.out.println(message.getBody());
        System.out.println(message.getMessageProperties());
    }
}

RabbitMQ入门

什么是消息队列

常见MQ产品

MQ是消息通信的模型，并发具体实现

实现MQ的有两种主流方式：AMQP、JMS。

JMS(Java Message Service)：

AMQP(Advanced Message Queuing Protocol)

大多应用中，可通过消息服务中间件来提升系统异步通信、扩展解耦能力

消息服务中两个重要概念：

消息队列主要有两种目的地

点对点式：

发布订阅式：

Spring支持

Spring Boot自动配置

RabbitMQ简介

RabbitMQ的工作原理

核心概念

Message

Publisher

Exchange

Queue

Binding

Connection

Channel

Consumer

Virtual Host

Broker