RabbitMQ使用详解

zhoucheng0

2020-03-08

关注关注

RabbitMQ是基于AMQP的一款消息管理系统。AMQP(Advanced Message Queuing Protocol)，是一个提供消息服务的应用层标准高级消息队列协议，其中RabbitMQ就是基于这种协议的一种实现。

常见mq：

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

1 消息模型

RabbitMq有5种常用的消息模型

1.1 基本消息模型

这是最简单的消息模型，如下图：
RabbitMQ使用详解

生产者将消息发送到队列，消费者从队列中获取消息，队列是存储消息的缓冲区。

再演示代码之前，我们先创建一个工程rabbitmq-demo，并编写一个工具类，用于提供与mq服务创建连接

public class ConnectionUtil {
    /**
     * 建立与RabbitMQ的连接
     * @return
     * @throws Exception
     */
    public static Connection getConnection() throws Exception {
        //定义连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置服务地址
        factory.setHost("192.168.18.130");
        //端口
        factory.setPort(5672);
        //设置账号信息，用户名、密码、vhost
        factory.setUsername("admin");
        factory.setPassword("admin");
        // 通过工程获取连接
        Connection connection = factory.newConnection();
        return connection;
    }
}

生产者发送消息

接下来是生产者发送消息，其过程包括：1.与mq服务建立连接，2.建立通道，3.声明队列（有相同队列则不创建，没有则创建），4.发送消息，代码如下：

public class Send {
    private static final String QUEUE_NAME = "basic_queue";
    public static void main(String[] args) throws Exception {
        //消息发送端与mq服务创建连接
        Connection connection = ConnectionUtil.getConnection();
        //建立通道
        Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        String message = "hello world";
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println("生产者已发送：" + message);
        channel.close();
        connection.close();
    }
}

消费者接受消息

消费者在接收消息的过程需要经历如下几个步骤： 1.与mqfuwu建立连接，2.建立通道，3.声明队列，4，接收消息，代码如下：

public class Consumer1 {
    private static final String QUEUE_NAME = "basic_queue";
    public static void main(String[] args) throws Exception {
        //消息消费者与mq服务建立连接
        Connection connection = ConnectionUtil.getConnection();
        //建立通道
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println("消费者1接收到消息：" + msg);
            }
        };
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

消息的接收与消费使用都需要在一个匿名内部类DefaultConsumer中完成

注意：队列需要提前声明，如果未声明就使用队列，则会报错。如果不清楚生产者和消费者谁先声明，为了保证不报错，生产者和消费者都声明队列，队列的创建会保证幂等性，也就是说生产者和消费者都声明同一个队列，则只会创建一个队列

1.2 Work Queues工作队列模型

在基本消息模型中，一个生产者对应一个消费者，而实际生产过程中，往往消息生产会发送很多条消息，如果消费者只有一个的话效率就会很低，因此rabbitmq有另外一种消息模型，这种模型下，一个生产发送消息到队列，允许有多个消费者接收消息，但是一条消息只会被一个消费者获取。

RabbitMQ使用详解

生产者发送消息

与基本消息模型基本一致，这里测试循环发布20条消息：

public class Send {
    private static final String QUEUE_NAME = "work_queue";
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 循环发布任务
        for (int i = 1; i <= 20; i++) {
            // 消息内容
            String message = "task .. " + i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("生产者发送消息：" + message);
            Thread.sleep(500);
        }
        channel.close();
        connection.close();
    }
}

消费者1

public class Consumer1 {
    private static final String QUEUE_NAME = "work_queue";
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body);
                System.out.println("消费者1接收到消息：" + msg);
                try {
                    Thread.sleep(50);//模拟消费耗时
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

消费者2

public class Consumer2 {
    private static final String QUEUE_NAME = "work_queue";
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {
                String msg = new String(body);
                System.out.println("消费者2接收到消息：" + msg);
                try {
                    Thread.sleep(50);//模拟消费耗时
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

此时有两个消费者监听同一个队列，当两个消费者都工作时，生成者发送消息，就会按照负载均衡算法分配给不同消费者，如下图：
RabbitMQ使用详解

1.3 订阅模型

在之前的模型中，一条消息只能被一个消费者获取，而在订阅模式中，可以实现一条消息被多个消费者获取。在这种模型下，消息传递过程中比之前多了一个exchange交换机，生产者不是直接发送消息到队列，而是先发送给交换机，经由交换机分配到不同的队列，而每个消费者都有自己的队列：
RabbitMQ使用详解

解读：

1、1个生产者，多个消费者

2、每一个消费者都有自己的一个队列

3、生产者没有将消息直接发送到队列，而是发送到了交换机

4、每个队列都要绑定到交换机

5、生产者发送的消息，经过交换机到达队列，实现一个消息被多个消费者获取的目的

X（exchange）交换机的类型有以下几种：

Fanout：广播，交换机将消息发送到所有与之绑定的队列中去

Direct：定向，交换机按照指定的Routing Key发送到匹配的队列中去

Topics：通配符，与Direct大致相同，不同在于Routing Key可以根据通配符进行匹配

注意：在发布订阅模型中，生产者只负责发消息到交换机，至于消息该怎么发，以及发送到哪个队列，生产者都不负责。一般由消费者创建队列，并且绑定到交换机

订阅模型之Fanout

在广播模式下，消息发送的流程如下：

可以有多个消费者，每个消费者都有自己的队列
每个队列都要与exchange绑定
生产者发送消息到exchange
exchange将消息把消息发送到所有绑定的队列中去
消费者从各自的队列中获取消息

生产者发送消息

public class Send {
    private static final String EXCHANGE_NAME = "fanout_exchange";
    
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明exchange，指定类型为fanout
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
        String message = "hello world";
        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
        System.out.println("生产者发送消息：" + message);
        channel.close();
        connection.close();
    }
}

消费者

public class Consumer1 {
    private static final String QUEUE_NAME = "fanout_queue_1";
    private static final String EXCHANGE_NAME = "fanout_exchange";

    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        //消费者声明自己的队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 声明exchange，指定类型为direct
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
        //消费者将队列与交换机进行绑定
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body)
                    throws IOException
            {
                String msg = new String(body);
                System.out.println("消费者1获取到消息：" + msg);
            }
        });
    }
}

其他消费者只需修改QUEUE_NAME即可

注意：exchange与队列一样都需要提前声明，如果未声明就使用交换机，则会报错。如果不清楚生产者和消费者谁先声明，为了保证不报错，生产者和消费者都声明交换机，同样的，交换机的创建也会保证幂等性。

订阅模型之Direct

在fanout模型中，生产者发布消息，所有消费者都可以获取所有消息。在路由模式（Direct）中，可以实现不同的消息被不同的队列消费，在Direct模式下，交换机不再将消息发送给所有绑定的队列，而是根据Routing Key将消息发送到指定的队列，队列在与交换机绑定时会设定一个Routing Key，而生产者发送的消息时也需要携带一个Routing Key。
RabbitMQ使用详解

如图所示，消费者C1的队列与交换机绑定时设置的Routing Key是“error”，而C2的队列与交换机绑定时设置的Routing Key包括三个：“info”，“error”，“warning”，假如生产者发送一条消息到交换机，并设置消息的Routing Key为“info”，那么交换机只会将消息发送给C2的队列。

生产者发送消息

public class Send {
    private static final String EXCHANGE_NAME = "direct_exchange";

    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明exchange，指定类型为direct
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        String message = "新增一个订单";
        //生产者发送消息时，设置消息的Routing Key:"insert"
        channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes());
        System.out.println("生产者发送消息：" + message);
        channel.close();
        connection.close();
    }
}

消费者1

public class Consumer1 {
    private static final String QUEUE_NAME = "direct_queue_1";
    private static final String EXCHANGE_NAME = "direct_exchange";

    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        //消费者声明自己的队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        //消费者将队列与交换机进行绑定，并且设置Routing Key:"insert"
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");
        channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body)
                    throws IOException
            {
                String msg = new String(body);
                System.out.println("消费者1获取到消息：" + msg);
            }
        });
    }
}

其他消费者需要修改队列名QUEUE_NAME和Routing Key，上述生成者发送的消息，消费者1是可以获取到的

发布订阅之Topics

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

通配符规则：

#：匹配一个或多个词

     *：匹配不多不少恰好1个词

举例：

audit.#：能够匹配audit.irs.corporate 或者 audit.irs

     audit.*：只能匹配audit.irs

Topics生产者代码与Direct大致相同，只不过子声明交换机时，将类型设为BuiltinExchangeType.TOPIC(topic)，

消费者代码也与Direct大致相同，也是在声明交换机时设置类型为topic，代码不再演示

Spring AMQP

Spring AMQP是对AMQP的一种封装，目的是能够让我们更简便的使用消息队列，下面介绍一下Spring AMQP在Spring boot中的使用方法

依赖和配置

添加AMQP的启动器：

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

在application.yml中添加RabbitMQ的地址：

spring:
  rabbitmq:
    host: 192.168.18.130
    username: admin
    password: admin

消费者

消费者需要定义一个类，类中定义监听队列的方法

@Component
public class Listener {

    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = "spring.test.queue", durable = "false"),
                    exchange = @Exchange(value = "spring.test.exchange", type = ExchangeTypes.DIRECT),
                    key = "insert"
            )
    )
    public void listen(String msg){
        System.out.println("消费者接受到消息：" + msg);
    }
}

注解：

@Component：保证监听类被spring扫描到

@RabbitListener：
RabbitMQ使用详解

@RabbitListener包含很多内容，在发布订阅模式中，我们可以使用其中的“QueueBinding[] bindings”，其中QueueBinding底层如下：

RabbitMQ使用详解

其中Queue表示队列，Exchange表示交换机，key表示Routing Key

@RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = "spring.test.queue", durable = "false"),
                    exchange = @Exchange(value = "spring.test.exchange", type = ExchangeTypes.DIRECT),
                    key = "insert"
            )
    )

@Queue会创建队列

@Exchange会创建交换机

@QueueBinding会绑定队列和交换机

生产者发送消息

可以通过注解引入AmqpTemplate：

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAMQPTest {
    @Resource
    private AmqpTemplate template;

    @Test
    public void testSendMsg() throws InterruptedException {
        String message = "hello spring";
        template.convertAndSend("spring.test.exchange", "insert", message);
        System.out.println("生产者发送消息：" + message);
        Thread.sleep(10000);//等待10s，让测试方法延迟结束，防止消费者未来得及获取消息
    }
}

RabbitMQ如何防止消息丢失

1. 消息确认机制（ACK）

RabbitMQ有一个ACK机制，消费者在接收到消息后会向mq服务发送回执ACK，告知消息已被接收。这种ACK分为两种情况：

自动ACK：消息一旦被接收，消费者会自动发送ACK
手动ACK：消息接收后，不会自动发送ACK，而是需要手动发送ACK

如果消费者没有发送ACK，则消息会一直保留在队列中，等待下次接收。但这里存在一个问题，就是一旦消费者发送了ACK，如果消费者后面宕机，则消息会丢失。因此自动ACK不能保证消费者在接收到消息之后能够正常完成业务功能，因此需要在消息被充分利用之后，手动ACK确认

自动ACK，basicConsume方法中将autoAck参数设为true即可：
RabbitMQ使用详解

手动ack，在匿名内部类中，手动发送ACK：
RabbitMQ使用详解

当然，如果设置了手动ack，但又不手动发送ACK确认，消息会一直停留在队列中，可能造成消息的重复获取

2. 持久化

消息确认机制（ACK）能够保证消费者不丢失消息，但假如消费者在获取消息之前mq服务宕机，则消息也会丢失，因此要保证消息在服务端不丢失，则需要将消息进行持久化。队列、交换机、消息都要持久化。

队列持久化

RabbitMQ使用详解

exchange持久化

RabbitMQ使用详解

消息持久化

RabbitMQ使用详解

3. 生产者确认

生成者在发送消息过程中也可能出现错误或者网络延迟灯故障，导致消息未成功发送到交换机或者队列，或重复发送消息，为了解决这个问题，rabbitmq中有多个解决办法：

事务：

用事务将消息发送代码包围起来：
RabbitMQ使用详解

Confirm模式：

如下所示，在发送代码前执行channel.confirmSelect()，如果消息未正常发送，就会进入if代码块，可以进行重发也可以对失败消息进行记录
RabbitMQ使用详解

异步confirm方法：

顾名思义，就是生产者发送消息后不用等待服务端回馈发送状态，可以继续执行后面的代码，对于失败消息重发进行异步处理：
RabbitMQ使用详解

Spring AMQP中添加配置：

生产者确认机制，确保消息正确发送，如果发送失败会有错误回执，从而触发重试

spring:
  rabbitmq:
    publisher-confirms: true

rabbitmq 消息队列 mq 科技新闻

RabbitMQ使用详解

1 消息模型

1.1 基本消息模型

生产者发送消息

消费者接受消息

1.2 Work Queues工作队列模型

生产者发送消息

消费者1

消费者2

1.3 订阅模型

订阅模型之Fanout

生产者发送消息

消费者

订阅模型之Direct

生产者发送消息

消费者1

发布订阅之Topics

Spring AMQP

依赖和配置

消费者

生产者发送消息

RabbitMQ如何防止消息丢失

1. 消息确认机制（ACK）

2. 持久化

队列持久化

exchange持久化

消息持久化

3. 生产者确认

事务：

Confirm模式：

异步confirm方法：

Spring AMQP中添加配置：

相关推荐