一文带大家快速掌握RabbitMQ！（一）https://developer.aliyun.com/article/1624433

入门使用

引入RabbitMQ依赖：

<dependency>
 <groupId>com.rabbitmq</groupId>
 <artifactId>amqp-client</artifactId>
 <version>3.6.5</version>
</dependency>

创建一个生产者

public class Producer {
    private static final String QUEUE_NAME = "test01";
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 创建连接工厂并配置
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.58.129");
        connectionFactory.setPort(5672);
        // 设置虚拟机
        connectionFactory.setVirtualHost("/test");
        // 2. 通过连接工厂建立连接
        Connection connection = connectionFactory.newConnection();
        // 3. 通过connection创建Channel
        Channel channel = connection.createChannel();
        // 4. 通过Channel发送数据 (exchange, routingKey, props, body)
        // 不指定Exchange时, 交换机默认是AMQP default, 此时就看RoutingKey
        // RoutingKey要等于队列名才能被路由, 否则消息会被删除
        for (int i = 0; i < 5; i++) {
            String msg = "Learn For RabbitMQ-" + i;
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
            System.out.println("Send message : " + msg);
        }
        // 5.关闭连接
        channel.close();
        connection.close();
    }
}

创建一个消费者

public class Consumer {
    private static final String QUEUE_NAME = "test01";
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 创建连接工厂并配置
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.58.129");
        connectionFactory.setPort(5672);
        // 设置虚拟机
        connectionFactory.setVirtualHost("/test");
        // 2. 通过连接工厂建立连接
        Connection connection = connectionFactory.newConnection();
        // 3. 通过connection创建Channel
        Channel channel = connection.createChannel();
        // 4. 声明队列 (queue, durable, exclusive, autoDelete, args)
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        // 5. 创建消费者
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            /**
             * 获取消息 （监听到有消息时调用）
             * @param consumerTag 消费者标签, 在监听队列时可以设置autoAck为false,即手动确认（避免消息的丢失）, 消息唯一性处理
             * @param envelope 信封
             * @param properties 消息的属性
             * @param body 消息的内容
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("Received message : " + msg);
            }
        };
        // 6. 设置Channel, 监听队列(String queue, boolean autoAck,Consumer callback)
        channel.basicConsume(QUEUE_NAME, true, defaultConsumer);
    }
}

参数：

queue：队列名称
durable：持久化，true 即使服务重启也不会被删除
exclusive：独占，true 队列只能使用一个连接，连接断开队列删除
autoDelete：自动删除，true 脱离了Exchange（连接断开），即队列没有Exchange关联时，自动删除
arguments：扩展参数
autoAck：是否自动签收（回执）

不指定Exchange时，交换机默认是AMQP default，此时就看RoutingKey，RoutingKey要等于队列名才能被路由，否则消息会被删除

交换机属性

Name：交换机名称

Type：交换机类型—— direct、topic、fanout、header

Durability：是否需要持久化，true为持久化

Auto Delete：当最后一个绑定到Exchange上的队列删除后，即Exchange上没有队列绑定，自动删除该Exhcange

Internal：当前Exchange是否用于RabbitMQ内部使用，大多数使用默认False

Arguments：扩展参数，用于扩展AMQP协议定制化使用

Direct Exchange：

// Consumer
 // 声明交换机: 
 // (String exchange, String type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object) arguments)
 channel.exchangeDeclare("exchangeName", BuiltinExchangeType.DIRECT, true, false, false, null);
 // 声明队列 (String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object) args)
 channel.queueDeclare("queueName", true, false, false, null);
 
 // 建立绑定关系:
 channel.queueBind("queueName", "exchangeName", "routingKey");
 // ===================================================================
 // Producer
 // 发送消息 (String exchange, String routingKey, BasicProperties props, Bytes[] body)
 channel.basicPublish("exchangeName", "routingKey", null, "msg".getBytes());

Topic Exchange：

// Consumer
 // 声明交换机: 
 // (String exchange, String type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object) arguments)
 channel.exchangeDeclare("exchangeName", BuiltinExchangeType.TOPIC, true, false, false, null);
 // 声明队列 (String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object) args)
 channel.queueDeclare("queueName", true, false, false, null);
 
 // 建立绑定关系:
 channel.queueBind("queueName", "exchangeName", "routingKey.#");
 // ===================================================================
 // Producer
 // 发送消息 (String exchange, String routingKey, BasicProperties props, Bytes[] body)
 channel.basicPublish("exchangeName", "routingKey.hi", null, "msg".getBytes());
 channel.basicPublish("exchangeName", "routingKey.save", null, "msg".getBytes());
 channel.basicPublish("exchangeName", "routingKey.save.hi", null, "msg".getBytes());

因为使用了模糊匹配的"#"，可以匹配到发送的三条消息。因此可以收到三条消息

Fanout Exchange：

// Consumer
 // 声明交换机: 
 // (String exchange, String type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object) arguments)
 channel.exchangeDeclare("exchangeName", BuiltinExchangeType.FANOUT, true, false, false, null);
 // 声明队列 (String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object) args)
 channel.queueDeclare("queueName", true, false, false, null);
 
 // 建立绑定关系: 
 //（不设置routingKey, 这里为空）
 channel.queueBind("queueName", "exchangeName", "");
 // ===================================================================
 // Producer
 // 发送消息 (String exchange, String routingKey, BasicProperties props, Bytes[] body)
 // 同样routingKey为空 （也可以是任意字符串, 因为fanout并不依据routingKey）
 channel.basicPublish("exchangeName", "", null, "msg".getBytes());

高级特性

可靠性投递

什么是生产端的可靠性投递

保障消息的成功发出
保障MQ节点成功接收
发送端收到MQ节点（Broker）确认应答（已收到）
完善消息进行补偿机制

可靠性投递的方案一

消息落库（持久化至数据库），对消息状态进行打标，如若消息未响应，进行轮询操作

1.把业务消息落库，再生成一条消息落库到消息DB用来记录（譬如消息刚创建，正在发送中 status: 0）

缺点：对数据库进行两次持久化

2.生产端发送消息。

3.Broker端收到后，应答至生产端。Confirm Listener异步监听Broker的应答。

4.应答表明消息投递成功后，去消息DB中抓取到指定的消息记录，更新状态，如status: 1

5.如在3中出现网络不稳定等情况，导致Listener未收到消息成功确认的应答。

那么消息数据库中的status就还是0，而Broker可能是接收到消息的状态。

因此设定一个规则（定时任务），例如消息在落库5分钟后（超时）还是0的状态，就把该条记录抽取出来。

6.重新投递

7.限制一个重试的次数，譬如3次，如果大于3次，即为投递失败，更新status的值

用人工补偿机制去查询消息失败的原因

高并发场景消息的延迟投递，做二次确认，回调检查

Upstream service：生产端

Downstream service：消费端

1：业务消息落库后，发送消息至Broker。

2：紧接着发送第二条延迟（设置延迟时间）检查的消息。

3：消费端监听指定的队列接收到消息进行处理

4：处理完后，生成一条响应消息发送到Broker。

5：由Callback服务去监听该响应消息，收到该响应消息后持久化至消息DB（记录成功状态）。

6：到了延迟时间，延迟发送的消息也被Callback服务的监听器监听到后，去检查消息DB。如果未查询到成功的状态，Callback服务需要做补偿，发起RPC通讯，让生产端重新发送。生产端通过接收到的命令中所带的id去数据库查询该业务消息，再重新发送，即跳转到1。

该方案减少了对数据库的存储，保证了性能

消费端幂等性

避免消息的重复消费

消费端实现幂等性，接收到多条相同的消息，但不会重复消费，即收到多条一样的消息。

方案：

1.唯一ID + 指纹码机制

唯一ID + 指纹码（业务规则、时间戳等拼接）机制，利用数据库主键去重
SELECT COUNT(1) FROM T_ORDER WHERE ID = 唯一ID + 指纹码 未查询到就insert，如有说明已处理过该消息，返回失败
优点：实现简单
缺点：高并发下有数据库写入的性能瓶颈
解决方案：根据ID进行分库分表、算法路由

2.利用Redis的原子性

需要考虑的问题：

是否要落库数据库，如落库，数据库和缓存如何做到数据的一致性
不落库，数据存储在缓存中，如何设置定时同步的策略（可靠性保障）

Confirm确认消息

生产者投递消息后，如果Broker收到消息，则会给生产者一个应答。

生产者进行接收应答，用来确认这条消息是否正常发送到Broker是消息可靠性投递的核心保障。

确认机制的流程图

发送消息与监听应答的消息是异步操作。

确认消息的实现

在channel开启确认模式：channel.confirmSelect();
在channel添加监听：channel.addConfirmListener(ConfirmListener listener); 返回监听成功和失败的结果，对具体结果进行相应的处理（重新发送、记录日志等待后续处理等）

具体代码：

public class ConfirmProducer {
    private static final String EXCHANGE_NAME = "confirm_exchange";
    private static final String ROUTING_KEY = "confirm.key";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.58.129");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/test");
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        // 指定消息的投递模式: 确认模式
        channel.confirmSelect();
        // 发送消息
        String msg = "Send message of confirm demo";
        channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, null, msg.getBytes());
        // 添加确认监听
        channel.addConfirmListener(new ConfirmListener() {
            // 成功
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("========= Ack ========");
            }
            // 失败
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("========= Nack ========");
            }
        });
    }
}

一文带大家快速掌握RabbitMQ！（三）https://developer.aliyun.com/article/1624435

一文带大家快速掌握RabbitMQ！（二）

入门使用

高级特性

可靠性投递

消费端幂等性

Confirm确认消息

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

一文带大家快速掌握RabbitMQ！（二）

入门使用

高级特性

可靠性投递

消费端幂等性

Confirm确认消息

热门文章

最新文章

相关课程

相关电子书

相关实验场景