RabbitMq入门教程

RabbitMQ 简介

RabbitMQ是一个开源的，在AMQP基础上完整的，可复用的企业消息系统。

支持主流的操作系统，Linux、Windows、MacOX等

多种开发语言支持，Java、Python、Ruby、.NET、PHP、C/C++、node.js等

AMQP，即 Advanced Message Queuing Protocol（高级消息队列协议），是一个网络协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。2006年，AMQP 规范发布。

2007年，Rabbit 技术公司基于 AMQP 标准开发的 RabbitMQ 1.0 发布。RabbitMQ 采用 Erlang 语言开发。Erlang 语言由 Ericson 设计，专门为开发高并发和分布式系统的一种语言，在电信领域使用广泛。

RabbitMQ基本概念

RabbitMQ 基础架构：

Broker

接收和分发消息的应用，RabbitMQ Server就是 Message Broker

Virtual host

虚拟主机，出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时，可以划分出多个vhost，每个用户在自己的 vhost 创建 exchange／queue 等，每一个虚拟主机都有AMQP的全套基础组件，并且可以针对每个虚拟主机进行权限以及数据分配，并且不同虚拟主机之间是完全隔离的。

Connection

客户端与RabbitMQ进行交互，首先就需要建立一个TPC连接。RabbitMQ为了减少性能开销，也会在一个Connection中建立多个Channel，这样便于客户端进行多线程连接，这些连接会复用同一个Connection的TCP通道，提高性能。

Channel

客户端与RabbitMQ建立了连接，就会分配一个AMQP信道 Channel。每个信道都会被分配一个唯一的ID。

Exchange

消息队列交换机，消息发送到RabbitMQ中后，会首先进入一个交换机，然后由交换机负责将数据转发到不同的队列中。RabbitMQ中有多种不同类型的交换机来支持不同的路由策略。

交换机多用来与生产者打交道。生产者发送的消息通过Exchange交换机分配到各个不同的Queue队列上，而对于消息消费者来说，通常只需要关注自己的队列就可以了。

Queue

消息队列，队列是实际保存数据的最小单位。队列结构天生就具有FIFO的顺序。

Producer

消息生产者，即生产方客户端，生产方客户端将消息发送

Consumer

消息消费者，即消费方客户端，接收MQ转发的消息。

消息发送者的固定步骤

1.创建消息生产者producer，并制定生产者组名

2.指定Nameserver地址

3.启动producer

4.创建消息对象，指定主题Topic、Tag和消息体

5.发送消息

6.关闭生产者producer

消息消费者的固定步骤

1.创建消费者Consumer，制定消费者组名

2.指定Nameserver地址

3.订阅主题Topic和Tag

4.设置回调函数，处理消息

5.启动消费者consumer

编程模型

引入依赖

    com.rabbitmq
    amqp-client
    5.9.0

创建连接获取Channel

ConnectionFactory factory = new ConnectionFactory()
factory.setHost(HOST_NAME)
factory.setPort(HOST_PORT)
factory.setUsername(USER_NAME)
factory.setPassword(PASSWORD)
factory.setVirtualHost(VIRTUAL_HOST)
Connection connection = factory.newConnection()
Channel channel = connection.createChannel()

声明Exchange(可选)

channel.exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete,Map arguments) throws IOException;

Exchange有四种类型: fanout、 topic 、headers 、direct

声明queue

 channel.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments);

durable 表示是否持久化。Durable选项表示会将队列的消息写入硬盘，这样服务重启后这些消息就不会丢失。

声明Exchange与Queue的绑定关系(可选)

channel.queueBind(String queue, String exchange, String routingKey) throws IOException;

声明了Exchange和Queue，那么就还需要声明Exchange与Queue的绑定关系Binding。有了这些Binding，Exchange才可以知道Producer发送过来的消息将要分发到哪些Queue上。

这些Binding涉及到消息的不同分发逻辑，与Exchange和Queue一样，如果Broker上没有建立绑定关系，那么RabbitMQ会按照客户端的声明，创建这些绑定关系。

发送消息

channel.basicPublish(String exchange, String routingKey, BasicProperties props,message.getBytes("UTF-8")) ;

其中Exchange如果不需要，传个空字符串。

props的这些配置项，可以用RabbitMQ中提供的一个Builder对象来构建。

  AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder()
  //对应页面上的Properties部分，传入一些预定的参数值。
  builder.deliveryMode(MessageProperties.PERSISTENT_TEXT_PLAIN.getDeliveryMode())
  builder.priority(MessageProperties.PERSISTENT_TEXT_PLAIN.getPriority())
  //builder.headers(headers)
  builder.build()
  AMQP.BasicProperties prop = builder.build()

MessageProperties.PERSISTENT_TEXT_PLAIN是RabbitMQ提供的持久化消息的默认配置。

消费消息

被动消费模式

Consumer等待rabbitMQ 服务器将message推送过来再消费。

channel.basicConsume(String queue, boolean autoAck, Consumer callback);

主动消费模式

Comsumer主动到rabbitMQ服务器上去拉取messge进行消费。

GetResponse response = channel.basicGet(QUEUE_NAME, boolean autoAck)

消费消息确认

自动ACK：autoAck为true，消息一旦被接收，消费者自动发送ACK，如果消费失败了，后续也无法再消费了

手动ACK：autoAck为false，消息接收后，不会发送ACK，需要手动调用 channel.basicAck 来通知服务器已经消费了该message.这样即使Consumer在执行message过程中出问题了，也不会造成消息丢失。

释放资源

channel.close()
conection.clouse()

消息模型

简单模式

最直接的方式，P端发送一个消息到一个指定的queue，中间不需要任何exchange规则。C端按queue方式进行消费。

在上图的模型中，有以下概念：

P：生产者，也就是要发送消息的程序

C：消费者：消息的接受者。

queue：消息队列，图中红色部分。可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）。

producer:

channel.queueDeclare(QUEUE_NAME,false,false,false,null);
channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));

consumer:

channel.queueDeclare(QUEUE_NAME, false, false, false, null);

Work queues 工作队列模式

Work Queues：与简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。一个消息只会被一个消费者消费。

一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）。

应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null); 
channel.basicPublish("", TASK_QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,
        message.getBytes("UTF-8"));

Consumer: 每次拉取一条消息。

channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
channel.basicQos(1);
channel.basicConsume(TASK_QUEUE_NAME, false, consumer);

Publish/Subscribe 发布订阅

exchange type是 fanout。

在订阅模型中，多了一个 Exchange 角色，而且过程略有变化：

P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）

C：消费者，消息的接收者

Queue：消息队列，接收消息、缓存消息Exchange：交换机（X）。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。

Exchange有常见以下3种类型：

Fanout：广播，将消息交给所有绑定到交换机的队列，交换机需要与队列进行绑定，绑定之后；一个消息可以被多个消费者都收到。

Direct：定向，把消息交给符合指定routing key 的队列

Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与 Exchange 绑定，或者没有符合路由规则的队列，那么消息会丢失！producer只负责发送消息，至于消息进入哪个queue，由exchange来分配。

使用场景：

所有消费者获得相同的消息，例如天气预报。

生产者：

channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));

消费者：

channel.exchangeDeclare(EXCHANGE_NAME, "fanout")
String queueName = channel.queueDeclare().getQueue()
channel.queueBind(queueName, EXCHANGE_NAME, "")

发布订阅模式与工作队列模式的区别：

工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机

发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息(底层使用默认交换机)

发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑 定到默认的交换机

Rout 路由模式

exchange typ 是 direct。

P：生产者，向 Exchange 发送消息，发送消息时，会指定一个routing key

X：Exchange（交换机），接收生产者的消息，然后把消息递交给与 routing key 完全匹配的队列

C1：消费者，其所在队列指定了需要 routing key 为 error 的消息

C2：消费者，其所在队列指定了需要 routing key 为 info、error、warning 的消息

路由模式要求队列在绑定交换机时要指定 routing key，消息会转发到符合 routing key 的队列。

生产者：

channel.exchangeDeclare(EXCHANGE_NAME, "direct");
channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));

消费者:

channel.exchangeDeclare(EXCHANGE_NAME, "direct")
channel.queueBind(queueName, EXCHANGE_NAME, routingKey1)
channel.queueBind(queueName, EXCHANGE_NAME, routingKey2)
channel.basicConsume(queueName, true, consumer)

Topics 通配符模式

exchange type 是 topic

红色 Queue：绑定的是 usa.# ，因此凡是以 usa. 开头的 routing key 都会被匹配到

黄色 Queue：绑定的是 #.news ，因此凡是以 .news 结尾的 routing key 都会被匹配

对routingKey进行了模糊匹配单词之间用,隔开，* 代表一个具体的单词。# 代表0个或多个单词

Topic 主题模式可以实现 Pub/Sub 发布与订阅模式和 Routing 路由模式的功能，只是 Topic 在配置routing key 的时候可以使用通配符，显得更加灵活。

Producer:

channel.exchangeDeclare(EXCHANGE_NAME, "topic");
channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));

Receiver:

channel.exchangeDeclare(EXCHANGE_NAME, "topic")
channel.queueBind(queueName, EXCHANGE_NAME, routingKey1)
channel.queueBind(queueName, EXCHANGE_NAME, routingKey2)
channel.basicConsume(queueName, true, consumer)

发送消息确认

发送的消息如果没有被消费者及时消费有可能会导致消息丢失。

发送者确认模式默认是不开启的，所以如果需要开启发送者确认模式，需要手动在channel中进行声明。

channel.confirmSelect()

使用异步确认消息保证消息在生产端不丢失。

Producer在channel中注册监听器来对消息进行确认。核心代码：

channel.addConfirmListener(ConfirmCallback var1, ConfirmCallback var2)

ConfirmCallback，监听器接口，里面只有一个方法：

void handle(long sequenceNumber, boolean multiple) throws IOException; 

这方法中的两个参数，

sequenceNumer：这个是一个唯一的序列号，代表一个唯一的消息。在RabbitMQ中，他的消息体只是一个二进制数组，默认消息是没有序列号的。那么在回调的时候，Producer怎么知道是哪一条消息成功或者失败呢？RabbitMQ提供了一个方法int sequenceNumber = channel.getNextPublishSeqNo();来生成一个全局递增的序列号，这个序列号将会分配给新发送的那一条消息。然后应用程序需要自己来将这个序列号与消息对应起来。没错！是的！需要客户端自己去做对应！

multiple：这个是一个Boolean型的参数。如果是false，就表示这一次只确认了当前一条消息。如果是true，就表示RabbitMQ这一次确认了一批消息，在sequenceNumber之前的所有消息都已经确认完成了。

SpringBoot集成RabbitMQ

添加依赖

       
            org.springframework.boot
            spring-boot-starter-amqp
        

配置文件

server:
  port: 8081
spring:
  application:
    name: test-rabbitmq-producer
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    virtualHost: /

rabbitMQ配置类

配置Exchange、Queue、及绑定交换机，下面配置Topic交换机。

package com.example.config;
 
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 


 * RabbitmqConfig
 */
@Configuration
public class RabbitmqConfig {
 
    public static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
    public static final String QUEUE_INFORM_SMS = "queue_inform_sms";
    public static final String EXCHANGE_TOPICS_INFORM="exchange_topics_inform";
    public static final String ROUTINGKEY_EMAIL="inform.#.email.#";
    public static final String ROUTINGKEY_SMS="inform.#.sms.#";
 
    
    @Bean(EXCHANGE_TOPICS_INFORM)
    public Exchange EXCHANGE_TOPICS_INFORM(){
        
        return ExchangeBuilder.topicExchange(EXCHANGE_TOPICS_INFORM).durable(true).build();
    }
 
    
    @Bean(QUEUE_INFORM_EMAIL)
    public Queue QUEUE_INFORM_EMAIL(){
        return new Queue(QUEUE_INFORM_EMAIL);
    }
    
    @Bean(QUEUE_INFORM_SMS)
    public Queue QUEUE_INFORM_SMS(){
        return new Queue(QUEUE_INFORM_SMS);
    }
 
    
    @Bean
    public Binding BINDING_QUEUE_INFORM_EMAIL(@Qualifier(QUEUE_INFORM_EMAIL) Queue queue,
                                              @Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_EMAIL).noargs();
    }
    
    @Bean
    public Binding BINDING_ROUTINGKEY_SMS(@Qualifier(QUEUE_INFORM_SMS) Queue queue,
                                          @Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_SMS).noargs();
    }
 
}

发送消息

String message = "hello world"
rabbitTemplate.convertAndSend(RabbitmqTopicConfig.EXCHANGE_TOPICS_INFORM, "inform.email", message)

消费消息

消费者都是通过@RabbitListener注解来声明。在@RabbitMQListener注解中包含了非常多对Queue进行定制的属性，大部分的属性都是有默认值的。

    @RabbitListener(queues = {RabbitmqTopicConfig.QUEUE_INFORM_EMAIL})
    public void receive_email(Object msg, Message message, Channel channel){
        System.out.println("QUEUE_INFORM_EMAIL msg"+msg);
    }
    
    @RabbitListener(queues = {RabbitmqTopicConfig.QUEUE_INFORM_SMS})
    public void receive_sms(Object msg, Message message, Channel channel){
        System.out.println("QUEUE_INFORM_SMS msg"+msg);
    }

审核编辑：汤梓红