一文带大家快速掌握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确认消息

热门文章

最新文章

相关课程

相关电子书

相关实验场景