RabbitMQ笔记-阿里云开发者社区

依赖

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

简单模式

生产者

package com.imooc.mq;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
 * 构建简单模式的生产者，发送消息
 */
public class FooProducer {
    public static void main(String[] args) throws Exception {
        // 1. 创建连接工厂以及相关的参数配置
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.233.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        // 2. 通过工程创建连接 Connection
        Connection connection = factory.newConnection();
        // 3. 创建管道 Channel
        Channel channel = connection.createChannel();
        // 4. 创建队列 Queue（简单模式不需要交换机Exchange）
        /**
         * queue: 队列名
         * durable: 是否持久化，true：重启之后，队列依然存在，false则不存在
         * exclusive: 是否独占，true：只能有一个消费者监听这个队列，一般设置为false
         * autoDelete: 是否自动删除，true：当没有消费者的时候，则自动删除这个队列
         * arguments: map类型的其他参数
         */
        channel.queueDeclare("hello", true, false, false, null);
        // 5. 向队列发送消息
        /**
         * exchange: 交换机的名称，简单模式下没有，所以直接设置为 ""
         * routingKey: 路由key，映射路径，如果交换机没有，则路由key和队列名保持一致
         * props: 配置参数
         * body: 消息数据
         */
        String msg = "Hello 慕课网~~~";
        channel.basicPublish("", "hello", null, msg.getBytes());
        // 6. 释放资源
        channel.close();
        connection.close();
    }
}

消费者

package com.imooc.mq;
import com.rabbitmq.client.*;
import java.io.IOException;
/**
 * 构建简单模式的消费者，监听消费消息
 */
public class FooConsumer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.233.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("hello", true, false, false, null);
        Consumer consumer = new DefaultConsumer(channel){
            /**
             * 重写消息配送方法
             * @param consumerTag 消息的标签（标识）
             * @param envelope  信封（一些信息，比如交换机路由等等信息）
             * @param properties 配置信息
             * @param body 收到的消息数据
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                System.out.println("consumerTag = " + consumerTag);
                System.out.println("envelope = " + envelope.toString());
                System.out.println("BasicProperties = " + properties.toString());
                System.out.println("body = " + new String(body));
            }
        };
        /**
         * queue: 监听的队列名
         * autoAck: 是否自动确认，true：告知mq消费者已经消费的确认通知
         * callback: 回调函数，处理监听到的消息
         */
        channel.basicConsume("hello", true, consumer);
    }
}

工作队列模式

生产者

package com.imooc.mq;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
 * 构建工作队列的生产者，发送消息
 */
public class WorkQueuesProducer {
    public static void main(String[] args) throws Exception {
        // 1. 创建连接工厂以及相关的参数配置
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.233.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        // 2. 通过工程创建连接 Connection
        Connection connection = factory.newConnection();
        // 3. 创建管道 Channel
        Channel channel = connection.createChannel();
        // 4. 创建队列 Queue（简单模式不需要交换机Exchange）
        /**
         * queue: 队列名
         * durable: 是否持久化，true：重启之后，队列依然存在，false则不存在
         * exclusive: 是否独占，true：只能有一个消费者监听这个队列，一般设置为false
         * autoDelete: 是否自动删除，true：当没有消费者的时候，则自动删除这个队列
         * arguments: map类型的其他参数
         */
        channel.queueDeclare("work_queue", true, false, false, null);
        // 5. 向队列发送消息
        /**
         * exchange: 交换机的名称，简单模式下没有，所以直接设置为 ""
         * routingKey: 路由key，映射路径，如果交换机没有，则路由key和队列名保持一致
         * props: 配置参数
         * body: 消息数据
         */
        for (int i = 0 ; i < 10 ; i ++) {
            String task = "开始上班，搬砖喽~ 开启任务[" + i + "]";
            channel.basicPublish("", "work_queue", null, task.getBytes());
        }
        // 6. 释放资源
        channel.close();
        connection.close();
    }
}

消费者A

package com.imooc.mq;
import com.rabbitmq.client.*;
import java.io.IOException;
/**
 * 构建工作队列模式的消费者，监听消费消息
 */
public class WorkQueuesConsumerA {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.233.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work_queue", true, false, false, null);
        Consumer consumer = new DefaultConsumer(channel){
            /**
             * 重写消息配送方法
             * @param consumerTag 消息的标签（标识）
             * @param envelope  信封（一些信息，比如交换机路由等等信息）
             * @param properties 配置信息
             * @param body 收到的消息数据
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                System.out.println("body = " + new String(body));
            }
        };
        /**
         * queue: 监听的队列名
         * autoAck: 是否自动确认，true：告知mq消费者已经消费的确认通知
         * callback: 回调函数，处理监听到的消息
         */
        channel.basicConsume("work_queue", true, consumer);
    }
}

消费者B

package com.imooc.mq;
import com.rabbitmq.client.*;
import java.io.IOException;
/**
 * 构建工作队列模式的消费者，监听消费消息
 */
public class WorkQueuesConsumerB {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.233.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work_queue", true, false, false, null);
        Consumer consumer = new DefaultConsumer(channel){
            /**
             * 重写消息配送方法
             * @param consumerTag 消息的标签（标识）
             * @param envelope  信封（一些信息，比如交换机路由等等信息）
             * @param properties 配置信息
             * @param body 收到的消息数据
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                System.out.println("body = " + new String(body));
            }
        };
        /**
         * queue: 监听的队列名
         * autoAck: 是否自动确认，true：告知mq消费者已经消费的确认通知
         * callback: 回调函数，处理监听到的消息
         */
        channel.basicConsume("work_queue", true, consumer);
    }
}

发布订阅模式

生产者

package com.imooc.mq;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
 * 构建发布订阅模式的生产者，发送消息
 */
public class PubSubProducer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.233.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 创建交换机 Exchange
        /**
         * exchange: 交换机的名称
         * type: 交换机的类型
         *  FANOUT("fanout"): 广播模式，发布订阅，把消息发送给所有的绑定的队列
         *  DIRECT("direct"): 定向投递模式，把消息发送给指定的“routing key”的队列
         *  TOPIC("topic"): 通配符模式，把消息发送给符合的“routing pattern”的队列
         *  HEADERS("headers"): 使用率不多，参数匹配
         * durable: 是否持久化
         * autoDelete: 是否自动删除
         * internal: 内部意思，true：表示当前exchange是rabbitmq内部使用的，用户创建的队列不会消费该类型交换机下的消息，所以我们一般使用false即可
         * arguments: map类型的参数
         */
        String fanout_exchange = "fanout_exchange";
        channel.exchangeDeclare(fanout_exchange,
                BuiltinExchangeType.FANOUT,
                true, false, false, null);
        // 定义两个队列
        String fanout_queue_a = "fanout_queue_a";
        String fanout_queue_b = "fanout_queue_b";
        channel.queueDeclare(fanout_queue_a, true, false, false, null);
        channel.queueDeclare(fanout_queue_b, true, false, false, null);
        // 绑定交换机和队列
        channel.queueBind(fanout_queue_a, fanout_exchange, "");
        channel.queueBind(fanout_queue_b, fanout_exchange, "");
        for (int i = 0 ; i < 10 ; i ++) {
            String task = "开始上班，搬砖喽~ 开启任务[" + i + "]";
            channel.basicPublish(fanout_exchange, "", null, task.getBytes());
        }
        channel.close();
        connection.close();
    }
}

消费者A

package com.imooc.mq;
import com.rabbitmq.client.*;
import java.io.IOException;
/**
 * 构建发布订阅模式的消费者，监听消费消息
 */
public class PubSubConsumerA {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.233.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        String fanout_queue_a = "fanout_queue_a";
        channel.queueDeclare(fanout_queue_a, true, false, false, null);
        Consumer consumer = new DefaultConsumer(channel){
            /**
             * 重写消息配送方法
             * @param consumerTag 消息的标签（标识）
             * @param envelope  信封（一些信息，比如交换机路由等等信息）
             * @param properties 配置信息
             * @param body 收到的消息数据
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                System.out.println("body = " + new String(body));
            }
        };
        /**
         * queue: 监听的队列名
         * autoAck: 是否自动确认，true：告知mq消费者已经消费的确认通知
         * callback: 回调函数，处理监听到的消息
         */
        channel.basicConsume(fanout_queue_a, true, consumer);
    }
}

消费者B

package com.imooc.mq;
import com.rabbitmq.client.*;
import java.io.IOException;
/**
 * 构建发布订阅模式的消费者，监听消费消息
 */
public class PubSubConsumerB {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.233.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        String fanout_queue_b = "fanout_queue_b";
        channel.queueDeclare(fanout_queue_b, true, false, false, null);
        Consumer consumer = new DefaultConsumer(channel){
            /**
             * 重写消息配送方法
             * @param consumerTag 消息的标签（标识）
             * @param envelope  信封（一些信息，比如交换机路由等等信息）
             * @param properties 配置信息
             * @param body 收到的消息数据
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                System.out.println("body = " + new String(body));
            }
        };
        /**
         * queue: 监听的队列名
         * autoAck: 是否自动确认，true：告知mq消费者已经消费的确认通知
         * callback: 回调函数，处理监听到的消息
         */
        channel.basicConsume(fanout_queue_b, true, consumer);
    }
}

路由（定向）模式

生产者

package com.imooc.mq;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
 * 构建路由（定向）模式的生产者，发送消息
 */
public class RoutingProducer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.233.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 创建交换机 Exchange
        /**
         * exchange: 交换机的名称
         * type: 交换机的类型
         *  FANOUT("fanout"): 广播模式，发布订阅，把消息发送给所有的绑定的队列
         *  DIRECT("direct"): 定向投递模式，把消息发送给指定的“routing key”的队列
         *  TOPIC("topic"): 通配符模式，把消息发送给符合的“routing pattern”的队列
         *  HEADERS("headers"): 使用率不多，参数匹配
         * durable: 是否持久化
         * autoDelete: 是否自动删除
         * internal: 内部意思，true：表示当前exchange是rabbitmq内部使用的，用户创建的队列不会消费该类型交换机下的消息，所以我们一般使用false即可
         * arguments: map类型的参数
         */
        String routing_exchange = "routing_exchange";
        channel.exchangeDeclare(routing_exchange,
                BuiltinExchangeType.DIRECT,
                true, false, false, null);
        // 定义两个队列
        String routing_queue_order = "routing_queue_order";
        String routing_queue_pay = "routing_queue_pay";
        channel.queueDeclare(routing_queue_order, true, false, false, null);
        channel.queueDeclare(routing_queue_pay, true, false, false, null);
        // 绑定交换机和队列
        channel.queueBind(routing_queue_order, routing_exchange, "order_create");
        channel.queueBind(routing_queue_order, routing_exchange, "order_delete");
        channel.queueBind(routing_queue_order, routing_exchange, "order_update");
        channel.queueBind(routing_queue_pay, routing_exchange, "order_pay");
        String msg1 = "创建订单A";
        String msg2 = "创建订单B";
        String msg3 = "删除订单C";
        String msg4 = "修改订单D";
        String msg5 = "支付订单E";
        String msg6 = "支付订单F";
        channel.basicPublish(routing_exchange, "order_create", null, msg1.getBytes());
        channel.basicPublish(routing_exchange, "order_create", null, msg2.getBytes());
        channel.basicPublish(routing_exchange, "order_delete", null, msg3.getBytes());
        channel.basicPublish(routing_exchange, "order_update", null, msg4.getBytes());
        channel.basicPublish(routing_exchange, "order_pay", null, msg5.getBytes());
        channel.basicPublish(routing_exchange, "order_pay", null, msg6.getBytes());
        channel.close();
        connection.close();
    }
}

消费者A

package com.imooc.mq;
import com.rabbitmq.client.*;
import java.io.IOException;
/**
 * 构建路由模式的消费者，监听消费消息
 */
public class RoutingOrderConsumerA {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.233.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        String routing_queue_order = "routing_queue_order";
        channel.queueDeclare(routing_queue_order, true, false, false, null);
        Consumer consumer = new DefaultConsumer(channel){
            /**
             * 重写消息配送方法
             * @param consumerTag 消息的标签（标识）
             * @param envelope  信封（一些信息，比如交换机路由等等信息）
             * @param properties 配置信息
             * @param body 收到的消息数据
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                System.out.println("body = " + new String(body));
            }
        };
        /**
         * queue: 监听的队列名
         * autoAck: 是否自动确认，true：告知mq消费者已经消费的确认通知
         * callback: 回调函数，处理监听到的消息
         */
        channel.basicConsume(routing_queue_order, true, consumer);
    }
}

消费者B

package com.imooc.mq;
import com.rabbitmq.client.*;
import java.io.IOException;
/**
 * 构建路由模式的消费者，监听消费消息
 */
public class RoutingPayConsumerB {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.233.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        String routing_queue_pay = "routing_queue_pay";
        channel.queueDeclare(routing_queue_pay, true, false, false, null);
        Consumer consumer = new DefaultConsumer(channel){
            /**
             * 重写消息配送方法
             * @param consumerTag 消息的标签（标识）
             * @param envelope  信封（一些信息，比如交换机路由等等信息）
             * @param properties 配置信息
             * @param body 收到的消息数据
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                System.out.println("body = " + new String(body));
            }
        };
        /**
         * queue: 监听的队列名
         * autoAck: 是否自动确认，true：告知mq消费者已经消费的确认通知
         * callback: 回调函数，处理监听到的消息
         */
        channel.basicConsume(routing_queue_pay, true, consumer);
    }
}

通配符匹配模式 topic

生产者

package com.imooc.mq;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
 * 构建通配符模式的生产者，发送消息
 */
public class TopicsProducer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.233.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 创建交换机 Exchange
        /**
         * exchange: 交换机的名称
         * type: 交换机的类型
         *  FANOUT("fanout"): 广播模式，发布订阅，把消息发送给所有的绑定的队列
         *  DIRECT("direct"): 定向投递模式，把消息发送给指定的“routing key”的队列
         *  TOPIC("topic"): 通配符模式，把消息发送给符合的“routing pattern”的队列
         *  HEADERS("headers"): 使用率不多，参数匹配
         * durable: 是否持久化
         * autoDelete: 是否自动删除
         * internal: 内部意思，true：表示当前exchange是rabbitmq内部使用的，用户创建的队列不会消费该类型交换机下的消息，所以我们一般使用false即可
         * arguments: map类型的参数
         */
        String topics_exchange = "topics_exchange";
        channel.exchangeDeclare(topics_exchange,
                BuiltinExchangeType.TOPIC,
                true, false, false, null);
        // 定义两个队列
        String topics_queue_order = "topics_queue_order";
        String topics_queue_pay = "topics_queue_pay";
        channel.queueDeclare(topics_queue_order, true, false, false, null);
        channel.queueDeclare(topics_queue_pay, true, false, false, null);
        // 绑定交换机和队列 *表示通配一个词，#表示通配一个或多个词
        channel.queueBind(topics_queue_order, topics_exchange, "order.*");
        channel.queueBind(topics_queue_pay, topics_exchange, "*.pay.#");
        String msg1 = "创建订单A";
        String msg2 = "创建订单B";
        String msg3 = "删除订单C";
        String msg4 = "修改订单D";
        String msg5 = "支付订单E";
        String msg6 = "超市订单F";
        String msg7 = "慕课订单G";
        channel.basicPublish(topics_exchange, "order.create", null, msg1.getBytes());
        channel.basicPublish(topics_exchange, "order.create", null, msg2.getBytes());
        channel.basicPublish(topics_exchange, "order.delete", null, msg3.getBytes());
        channel.basicPublish(topics_exchange, "order.update", null, msg4.getBytes());
        // order.pay 能匹配到两个队列，注意看两个队列是否都能消费到
        channel.basicPublish(topics_exchange, "order.pay", null, msg5.getBytes());
        channel.basicPublish(topics_exchange, "imooc.pay.super.market", null, msg6.getBytes());
        channel.basicPublish(topics_exchange, "imooc.pay.course", null, msg7.getBytes());
        channel.close();
        connection.close();
    }
}

消费者A

package com.imooc.mq;
import com.rabbitmq.client.*;
import java.io.IOException;
/**
 * 构建通配符模式的消费者，监听消费消息
 */
public class TopicsOrderConsumerA {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.233.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        String topics_queue_order = "topics_queue_order";
        channel.queueDeclare(topics_queue_order, true, false, false, null);
        Consumer consumer = new DefaultConsumer(channel){
            /**
             * 重写消息配送方法
             * @param consumerTag 消息的标签（标识）
             * @param envelope  信封（一些信息，比如交换机路由等等信息）
             * @param properties 配置信息
             * @param body 收到的消息数据
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                System.out.println("body = " + new String(body));
            }
        };
        /**
         * queue: 监听的队列名
         * autoAck: 是否自动确认，true：告知mq消费者已经消费的确认通知
         * callback: 回调函数，处理监听到的消息
         */
        channel.basicConsume(topics_queue_order, true, consumer);
    }
}

消费者B

package com.imooc.mq;
import com.rabbitmq.client.*;
import java.io.IOException;
/**
 * 构建通配符模式的消费者，监听消费消息
 */
public class TopicsPayConsumerB {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.233.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        String topics_queue_pay = "topics_queue_pay";
        channel.queueDeclare(topics_queue_pay, true, false, false, null);
        Consumer consumer = new DefaultConsumer(channel){
            /**
             * 重写消息配送方法
             * @param consumerTag 消息的标签（标识）
             * @param envelope  信封（一些信息，比如交换机路由等等信息）
             * @param properties 配置信息
             * @param body 收到的消息数据
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                System.out.println("body = " + new String(body));
            }
        };
        /**
         * queue: 监听的队列名
         * autoAck: 是否自动确认，true：告知mq消费者已经消费的确认通知
         * callback: 回调函数，处理监听到的消息
         */
        channel.basicConsume(topics_queue_pay, true, consumer);
    }
}

SpringBoot集成RabbitMQ

依赖（生产和消费）

<!-- SpringBoot 整合RabbitMQ 依赖 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

配置类（生产者）

package com.imooc.api.mq;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * RabbitMQ 的配置类
 */
@Configuration
public class RabbitMQSMSConfig {
    // 定义交换机的名称
    public static final String SMS_EXCHANGE = "sms_exchange";
    // 定义队列的名称
    public static final String SMS_QUEUE = "sms_queue";
    // 统一定义路由key
    public static final String ROUTING_KEY_SMS_SEND_LOGIN = "imooc.sms.send.login";
    // 创建交换机
    @Bean(SMS_EXCHANGE)
    public Exchange exchange() {
        return ExchangeBuilder
                    .topicExchange(SMS_EXCHANGE)
                    .durable(true)
                    .build();
    }
    // 创建队列
    @Bean(SMS_QUEUE)
    public Queue queue() {
//        return new Queue(SMS_QUEUE);
        return QueueBuilder
                .durable(SMS_QUEUE)
                .build();
    }
    // 创建绑定关系
    @Bean
    public Binding smsBinding(@Qualifier(SMS_EXCHANGE) Exchange exchange,
                              @Qualifier(SMS_QUEUE) Queue queue) {
        return BindingBuilder
                .bind(queue)
                .to(exchange)
                .with("imooc.sms.#")
                .noargs();
    }
}

yml配置（生产和消费）

spring:
  rabbitmq:
    host: 192.168.233.128
    port: 5672
    virtual-host: /
    username: guest
    password: guest

生产者

@Autowired
    private RabbitTemplate rabbitTemplate;
    @PostMapping("getSMSCode")
        public GraceJSONResult getSMSCode(String mobile,
                HttpServletRequest request) throws Exception {
            if (StringUtils.isBlank(mobile)) {
                return GraceJSONResult.error();
            }
            // 获得用户ip
            String userIp = IPUtil.getRequestIp(request);
            // 限制用户只能在60s以内获得一次验证码
            redis.setnx60s(MOBILE_SMSCODE + ":" + userIp, mobile);
            String code = (int)((Math.random() * 9 + 1) * 100000) + "";
            // 使用消息队列异步解耦发送短信
            SMSContentQO contentQO = new SMSContentQO();
            contentQO.setMobile(mobile);
            contentQO.setContent(code);
            rabbitTemplate.convertAndSend(
                    "sms_exchange",
                    "imooc.sms.send.login",
                    GsonUtils.object2String(contentQO)
            );

GsonUtils

package com.imooc.utils;
import com.google.gson.*;
import com.google.gson.reflect.TypeToken;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
public class GsonUtils {
    /**
     * 不用创建对象,直接使用Gson.就可以调用方法
     */
    private static Gson gson = null;
    private static JsonParser jsonParser = null;
    /**
     * 判断gson对象是否存在了,不存在则创建对象
     */
    static {
        if (gson == null) {
            //gson = new Gson();
            // 当使用GsonBuilder方式时属性为空的时候输出来的json字符串是有键值key的,显示形式是"key":null，而直接new出来的就没有"key":null的
            gson = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create();
        }
        if(jsonParser == null ){
            jsonParser = new JsonParser();
        }
    }
    private GsonUtils() {}
    /**
     * json 转对象
     * @param strJson
     * @return
     */
    public static JsonObject string2Object(String strJson) {
        return jsonParser.parse(strJson).getAsJsonObject();
    }
    /**
     * 将对象转成json格式
     * @param object
     * @return String
     */
    public static String object2String(Object object) {
        String gsonString = null;
        if (gson != null) {
            gsonString = gson.toJson(object);
        }
        return gsonString;
    }
    /**
     * 将json转成特定的cls的对象
     * @param gsonString
     * @param cls
     * @return
     */
    public static <T> T stringToBean(String gsonString, Class<T> cls) {
        T t = null;
        if (gson != null) {
            //传入json对象和对象类型,将json转成对象
            t = gson.fromJson(gsonString, cls);
        }
        return t;
    }
//    public static <T> T stringToBean2(String gsonString, Class<T> cls) {
//
//        JsonParser jsonParser = new JsonParser();
//        JsonObject jsonObject = jsonParser.parse(gsonString).getAsJsonObject();
//
//        T t = null;
//        if (gson != null) {
//            //传入json对象和对象类型,将json转成对象
//            t = gson.fromJson(jsonObject, cls);
//        }
//        return t;
//    }
    /**
     * json字符串转成list
     * @param gsonString
     * @param cls
     * @return
     */
    public static <T> List<T> stringToList(String gsonString, Class<T> cls) {
        List<T> list = null;
        if (gson != null) {
            //根据泛型返回解析指定的类型,TypeToken<List<T>>{}.getType()获取返回类型
            list = gson.fromJson(gsonString, new TypeToken<List<T>>() {
            }.getType());
        }
        return list;
    }
    public static <T> List<T> stringToListAnother(String gsonString, Class<T> cls) {
        List<T> list = new ArrayList<>();
        JsonArray jsonArray = new JsonParser().parse(gsonString).getAsJsonArray();
        Gson gson = new Gson();
        for (JsonElement jsonElement : jsonArray) {
            list.add(gson.fromJson(jsonElement,cls));
        }
        return list;
    }
    /**
     * json字符串转成list中有map的
     * @param gsonString
     * @return
     */
    public static <T> List<Map<String, T>> stringToListMaps(String gsonString) {
        List<Map<String, T>> list = null;
        if (gson != null) {
            list = gson.fromJson(gsonString,
                    new TypeToken<List<Map<String, T>>>() {
                    }.getType());
        }
        return list;
    }
    /**
     * json字符串转成map的
     * @param gsonString
     * @return
     */
    public static <T> Map<String, T> stringToMaps(String gsonString) {
        Map<String, T> map = null;
        if (gson != null) {
            map = gson.fromJson(gsonString, new TypeToken<Map<String, T>>() {
            }.getType());
        }
        return map;
    }
    public static String jsonElementAsString(JsonElement jsonElement) {
        return jsonElement == null ? null : jsonElement.getAsString();
    }
    public static Integer jsonElementAsInt(JsonElement jsonElement) {
        return jsonElement == null ? null : jsonElement.getAsInt();
    }
}

消费者

/**
     * 监听队列，并且处理消息
     * @param payload
     * @param message
     */
    @RabbitListener(queues = {RabbitMQSMSConfig.SMS_QUEUE})
    public void watchQueue(String payload, Message message) throws Exception {
        log.info("payload = " + payload);
        String routingKey = message.getMessageProperties().getReceivedRoutingKey();
        log.info("routingKey = " + routingKey);
        String msg = payload;
        log.info("msg = " + msg);
        if (routingKey.equalsIgnoreCase(RabbitMQSMSConfig.ROUTING_KEY_SMS_SEND_LOGIN)) {
            // 此处为短信发送的消息消费处理
            SMSContentQO contentQO = GsonUtils.stringToBean(msg, SMSContentQO.class);
//            smsUtils.sendSMS(contentQO.getMobile(), contentQO.getContent());
        }
    }

生产端消息可靠性机制

setConfirmCallback交换机确认接收到消息机制

配置文件：

rabbitmq:
    host: 192.168.233.128
    port: 5672
    virtual-host: /
    username: guest
    password: guest
    publisher-confirm-type: correlated

生产者实现

rabbitTemplate.convertAndSend(
                "sms_exchange",
                "imooc.sms.send.login",
                GsonUtils.object2String(contentQO),
                new CorrelationData(UUID.randomUUID().toString())
        );
        // 定义confirm回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             * 回调函数
             * @param correlationData 相关性数据
             * @param ack 交换机是否成功接收到消息，true：成功
             * @param cause 失败的原因
             */
            @Override
            public void confirm(CorrelationData correlationData,
                                boolean ack,
                                String cause) {
                log.info("进入confirm");
                log.info("correlationData：{}", correlationData.getId());
                if (ack) {
                    log.info("交换机成功接收到消息~~ {}", cause);
                } else {
                    log.info("交换机接收消息失败~~失败原因： {}", cause);
                }
            }
        });

setReturnsCallback队列未接受到消息回退机制

配置文件

rabbitmq:
    host: 192.168.233.128
    port: 5672
    virtual-host: /
    username: guest
    password: guest
    publisher-confirm-type: correlated
    publisher-returns: true

生产者实现

rabbitTemplate.convertAndSend(
                "sms_exchange",
                "123imooc.sms.send.login",
                GsonUtils.object2String(contentQO),
                new CorrelationData(UUID.randomUUID().toString())
        );
        // 定义路由匹配不到时候 return回调
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                log.info("进入return");
                log.info(returned.toString());
            }
        });

这里把路由故意写为"123imooc.sms.send.login" 这匹配不到RabbitMQSMSConfig类中定义的路由，所以从交换机发布到对应的队列中

所以就会进入到setReturnsCallback方法中

消费端消息可靠性

ACK确认机制

none 默认自动确认
  manual 手动确认（业务处理完成后手动确认，最常用）
  auto 异常类型确认

配置文件

rabbitmq:
    host: 192.168.233.128
    port: 5672
    virtual-host: /
    username: guest
    password: guest
    listener:
      simple:
        acknowledge-mode: manual  # 手动ack确认

消费者实现

@RabbitListener(queues = {RabbitMQSMSConfig.SMS_QUEUE})
    public void watchQueue(Message message, Channel channel) throws Exception {
        try {
            String routingKey = message.getMessageProperties().getReceivedRoutingKey();
            log.info("routingKey = " + routingKey);
            int a = 1/0; // 模拟异常
            String msg = new String(message.getBody());
            log.info("msg = " + msg);
            /**
             * long deliveryTag: 消息投递的标签
             * boolean multiple: 批量确认所有消费者获得的消息
             */
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),
                            true);
        } catch (Exception e) {
            e.printStackTrace();
            /**
             * long deliveryTag
             * boolean multiple
             * boolean requeue: true：重回队列 ；false：丢弃消息
             */
            channel.basicNack(message.getMessageProperties().getDeliveryTag(),
                    true,
                    false);
        }
    }

消费端消息拉取限流

配置文件

rabbitmq:
    host: 192.168.233.128
    port: 5672
    virtual-host: /
    username: guest
    password: guest
    listener:
      simple:
        acknowledge-mode: manual  # 手动ack确认
        prefetch: 2  # 每次每个消费者从mq中拉去的消息的数量，直到手动ack确认之后，才会拉取下一个消息

模拟场景：

生产者连续发10条消息

for (int i = 0 ; i < 10 ; i ++) {
            rabbitTemplate.convertAndSend(RabbitMQSMSConfig.SMS_EXCHANGE,
                    RabbitMQSMSConfig.ROUTING_KEY_SMS_SEND_LOGIN,
                    GsonUtils.object2String(contentQO),
                    new CorrelationData(UUID.randomUUID().toString()));
        }

消费者在手动ack的地方断点

查看RabbitMQ管理平台，每次确实只从队列中拉取2条消息

设置消息过期失效（移除消息）

在RabbitMQSMSConfig配置类中

// 创建队列
    @Bean(SMS_QUEUE)
    public Queue queue() {
        return QueueBuilder
                .durable(SMS_QUEUE)
                .withArgument("x-message-ttl", 30*1000)
                .build();
    }

.withArgument(“x-message-ttl”, 30*1000) 表示30秒后，消息自动失效

注意：只能对新的队列生效，对老的队列需要先在RabbitMQ管理平台中删除老队列

测试方法

可以先将消费者注释掉，或者不启动消费者

生产者生产的消息，过一段时间在RabbitMQ管理平台看看是否自动消失

死信队列

以下3中消息都会进入死信队列

1、 ttl超时被丢弃的消息

2、超过队列长度被丢弃的消息

3、手动nack或者reject，并且requeue为false的消息

死信队列配置类

package com.imooc.api.mq;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * RabbitMQ 的配置类(用于死信队列的配置)
 */
@Configuration
public class RabbitMQSMSConfig_Dead {
    // 定义交换机的名称
    public static final String SMS_EXCHANGE_DEAD = "sms_exchange_dead";
    // 定义队列的名称
    public static final String SMS_QUEUE_DEAD = "sms_queue_dead";
    // 统一定义路由key
    public static final String ROUTING_KEY_SMS_DEAD = "dead.sms.display";
    // 创建交换机
    @Bean(SMS_EXCHANGE_DEAD)
    public Exchange exchange() {
        return ExchangeBuilder
                    .topicExchange(SMS_EXCHANGE_DEAD)
                    .durable(true)
                    .build();
    }
    // 创建队列
    @Bean(SMS_QUEUE_DEAD)
    public Queue queue() {
        return QueueBuilder
                .durable(SMS_QUEUE_DEAD)
                .build();
    }
    // 创建绑定关系
    @Bean
    public Binding smsDeadBinding(@Qualifier(SMS_EXCHANGE_DEAD) Exchange exchange,
                              @Qualifier(SMS_QUEUE_DEAD) Queue queue) {
        return BindingBuilder
                .bind(queue)
                .to(exchange)
                .with("dead.sms.*")
                .noargs();
    }
}

正常队列配置类中添加死信队列配置

package com.imooc.api.mq;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * RabbitMQ 的配置类
 */
@Configuration
public class RabbitMQSMSConfig {
    // 定义交换机的名称
    public static final String SMS_EXCHANGE = "sms_exchange";
    // 定义队列的名称
    public static final String SMS_QUEUE = "sms_queue";
    // 统一定义路由key
    public static final String ROUTING_KEY_SMS_SEND_LOGIN = "imooc.sms.send.login";
    // 创建交换机
    @Bean(SMS_EXCHANGE)
    public Exchange exchange() {
        return ExchangeBuilder
                    .topicExchange(SMS_EXCHANGE)
                    .durable(true)
                    .build();
    }
    // 创建队列
    @Bean(SMS_QUEUE)
    public Queue queue() {
        return QueueBuilder
                .durable(SMS_QUEUE)
                .withArgument("x-message-ttl", 30*1000)  // 队列中消息过期时间，过期进入死信队列
                .withArgument("x-dead-letter-exchange",
                        RabbitMQSMSConfig_Dead.SMS_EXCHANGE_DEAD)    // 死信队列交换机
                .withArgument("x-dead-letter-routing-key",
                        RabbitMQSMSConfig_Dead.ROUTING_KEY_SMS_DEAD)  // 死信队列路由
                .withArgument("x-max-length", 6)    // 队列消息个数最大值，超出的进入死信队列
                .build();
    }
    // 创建绑定关系
    @Bean
    public Binding smsBinding(@Qualifier(SMS_EXCHANGE) Exchange exchange,
                              @Qualifier(SMS_QUEUE) Queue queue) {
        return BindingBuilder
                .bind(queue)
                .to(exchange)
                .with("imooc.sms.#")
                .noargs();
    }
}

死信队列消费者

package com.imooc.mq;
import com.imooc.api.mq.RabbitMQSMSConfig_Dead;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
 * 死信队列监听消费者
 */
@Slf4j
@Component
public class RabbitMQSMSConsumer_Dead {
    /**
     *
     * @param message
     * @param channel
     * @throws Exception
     */
    @RabbitListener(queues = {RabbitMQSMSConfig_Dead.SMS_QUEUE_DEAD})
    public void watchQueue(Message message, Channel channel) throws Exception {
        log.info("++++++++++++++++++++++++++++++");
        String routingKey = message.getMessageProperties().getReceivedRoutingKey();
        log.info("routingKey = " + routingKey);
        String msg = new String(message.getBody());
        log.info("msg = " + msg);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),
                        true);
        log.info("++++++++++++++++++++++++++++++");
    }
}

测试方法：正常的消费者关闭，设置队列长度为6，然后发10条消息，会发现，首先有4条超过队列长度的消息，进入死信队列消费者，然后，到了消息过期时间后，又有6条消息进入死信队列中被死信消费者消费

RabbitMQ笔记

依赖

简单模式

工作队列模式

发布订阅模式

路由（定向）模式

通配符匹配模式 topic

SpringBoot集成RabbitMQ

生产端消息可靠性机制

setConfirmCallback交换机确认接收到消息机制

setReturnsCallback队列未接受到消息回退机制

消费端消息可靠性

ACK确认机制

消费端消息拉取限流

设置消息过期失效（移除消息）

死信队列

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

RabbitMQ笔记

依赖

简单模式

工作队列模式

发布订阅模式

路由（定向）模式

通配符匹配模式 topic

SpringBoot集成RabbitMQ

生产端消息可靠性机制

setConfirmCallback交换机确认接收到消息机制

setReturnsCallback队列未接受到消息回退机制

消费端消息可靠性

ACK确认机制

消费端消息拉取限流

设置消息过期失效（移除消息）

死信队列

热门文章

最新文章

相关课程

相关电子书

相关实验场景