RabbitMQ-工作模式

RabbitMQ提供了6中模式：简单模式，工作队列模式，Plublish/Subscribe发布/订阅模式，Routing路由模式，Topics主题模式，RPC远程调用模式（远程调用，不太算MQ；暂不做介绍）。

Exchange 交换机

Exchange：一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如传递交给某个特别队列、递交给所有队列、或者将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型：

• Fanout：广播，将消息交给所有绑定到交换机的队列；
• Direct：定向，把消息交给符合指定Routing key 的队列；
• Topic：通配符，把消息交给符合Routing pattern（路由模式）的队列。

简单模式（”Hello World!”）

一个生产者，一个消费者。

具体案例已经在《RabbitMQ学习》中使用。

工作队列模式（Work queues）

简介

Work queues与简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。

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

代码示例

Work queues与简单模式的代码几乎一样，可以完全复制一个或多个消费者进行多个消费者同时消费消息的测试。

1）工具类：

public class ConnectionUtil {
    public static Connection getConnection() throws IOException, TimeoutException {
        //创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //主机地址：默认为localhost
        connectionFactory.setHost("192.168.0.8");
        //连接端口
        connectionFactory.setPort(5672);
        //虚拟主机名称：默认为/
        connectionFactory.setVirtualHost("/ligangit");
        //连接用户名：默认为guest
        connectionFactory.setUsername("ligangit");
        //连接密码：默认为guest
        connectionFactory.setPassword("ligangit");
        //创建连接
        return connectionFactory.newConnection();
    }
}

2）生产者：

public class Producer {
    static final String QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接
        Connection connection = ConnectionUtil.getConnection();
        //创建频道
        Channel channel = connection.createChannel();
        //声明（创建）队列
        /**
         * 参数一：队列名称
         * 参数二：是否定义持久化队列
         * 参数三：是否独占本次连接
         * 参数四：是否在不使用的时候自动删除队列
         * 参数五：队列其他参数
         */
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        for (int i = 1; i <= 30; i++) {
            //发送消息
            String message = "你好！work模式==="+i;
            /**
             * 参数一：交换机名称，如果没有指定则使用默认Deafult Exchange
             * 参数二：路由key，简单模式可以传递队列名称
             * 参数三：消息其他属性
             * 参数四：消息内容
             */
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("已发送消息：" + message);
        }
        //释放资源
        channel.close();
        connection.close();

    }
}

3）消费者1：

public class Consumer1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接
        Connection connection = ConnectionUtil.getConnection();
        //创建频道
        Channel channel = connection.createChannel();
        //创建队列，并设置消息处理
        channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
        //监听消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag：消息标签，在 channel.basicConsume 的时候可以指定
             * envelope：消息包内容，可从中获取消息id，消息routingkey，交换机，消息和重转标记（收到消息失败后是否需要重新发送）
             * properties：消息属性
             * body：消息
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("");
                System.out.println("=========================消费者1开始=====================================");
                System.out.println("");
                //路由key
                System.out.println("路由key为："+envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为："+envelope.getExchange());
                //消息id
                System.out.println("消息id为："+envelope.getDeliveryTag());
                //收到的消息
                System.out.println("消费者1__接收到的消息：" + new String(body, "UTF-8"));
                System.out.println("");
                System.out.println("===============================消费者1结束===============================");
                System.out.println("");
                //为了看到效果，让当前线程休眠
                try {
                    Thread.sleep(1000);//休眠1秒钟
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        /**
         * 监听消息
         * 参数一：队列名称
         * 参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到回复消息后会删除消息；设置为false则需要手动确认
         */
        channel.basicConsume(Producer.QUEUE_NAME,true,consumer);
        //不关闭资源，应该一直监听消息
//        channel.close();
//        connection.close();
    }
}

4）消费者2：

public class Consumer2 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接
        Connection connection = ConnectionUtil.getConnection();
        //创建频道
        Channel channel = connection.createChannel();
        //创建队列，并设置消息处理
        channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
        //监听消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag：消息标签，在 channel.basicConsume 的时候可以指定
             * envelope：消息包内容，可从中获取消息id，消息routingkey，交换机，消息和重转标记（收到消息失败后是否需要重新发送）
             * properties：消息属性
             * body：消息
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("");
                System.out.println("=========================消费者2开始=====================================");
                System.out.println("");
                //路由key
                System.out.println("路由key为："+envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为："+envelope.getExchange());
                //消息id
                System.out.println("消息id为："+envelope.getDeliveryTag());
                //收到的消息
                System.out.println("消费者2___接收到的消息：" + new String(body, "UTF-8"));
                System.out.println("");
                System.out.println("=========================消费者2结束=====================================");
                System.out.println("");
                //为了看到效果，让当前线程休眠
                try {
                    Thread.sleep(1000);//休眠1秒钟
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        /**
         * 监听消息
         * 参数一：队列名称
         * 参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到回复消息后会删除消息；设置为false则需要手动确认
         */
        channel.basicConsume(Producer.QUEUE_NAME,true,consumer);
        //不关闭资源，应该一直监听消息
//        channel.close();
//        connection.close();
    }
}

测试结果

消费者1结果：

消费者2结果：

小结

一个队列如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系。

发布/订阅模式（Plublish/Subscribe）

简介

简单模式、工作队列模式中，只有3个角色

• P：生产者，也就是要发送消息的程序；
• C：消费者，消息的接收者，会一直等待消息的到来；
• Queue：消息队列

在发布/订阅模式中，多了一个exchange角色，而且过程略有变化：

• P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发送给X（交换机）；
• C：消费者，消息的接收者，会一直等待消息的到来；
• Queue：消息队列，接收消息、缓存消息；
• Exchange：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如传递交给某个特别队列、递交给所有队列、或者将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型：
  • Fanout：广播，将消息交给所有绑定到交换机的队列；
  • Direct：定向，把消息交给符合指定Routing key 的队列；
  • Topic：通配符，把消息交给符合Routing pattern（路由模式）的队列。

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失。

发布订阅模式：

1. 每个消费者监听自己的队列；
2. 生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息

代码示例

1）工具类：

public class ConnectionUtil {
    public static Connection getConnection() throws IOException, TimeoutException {
        //创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //主机地址：默认为localhost
        connectionFactory.setHost("192.168.0.8");
        //连接端口
        connectionFactory.setPort(5672);
        //虚拟主机名称：默认为/
        connectionFactory.setVirtualHost("/ligangit");
        //连接用户名：默认为guest
        connectionFactory.setUsername("ligangit");
        //连接密码：默认为guest
        connectionFactory.setPassword("ligangit");
        //创建连接
        return connectionFactory.newConnection();
    }
}

2）生产者：

/**
 * 发布/订阅模式使用的交换机类型为：fanout
 */
public class Producer {
    //交换机名称
    static final String FANOUT_EXCHANGE = "fanout_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "fanout_queue_1";
    //队列名称
    static final String FANOUT_QUEUE_2 = "fanout_queue_2";

    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接
        Connection connection = ConnectionUtil.getConnection();
        //创建频道
        Channel channel = connection.createChannel();
        /**
         * 声明交换机
         * 参数一：交换机名称
         * 参数二：交换机类型，fanout、topic、direct、headers
         */
        channel.exchangeDeclare(FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT);
        //声明（创建）队列
        /**
         * 参数一：队列名称
         * 参数二：是否定义持久化队列
         * 参数三：是否独占本次连接
         * 参数四：是否在不使用的时候自动删除队列
         * 参数五：队列其他参数
         */
        channel.queueDeclare(FANOUT_QUEUE_1, true, false, false, null);
        channel.queueDeclare(FANOUT_QUEUE_2, true, false, false, null);
        //队列绑定交换机
        channel.queueBind(FANOUT_QUEUE_1, FANOUT_EXCHANGE, "");
        channel.queueBind(FANOUT_QUEUE_2, FANOUT_EXCHANGE, "");

        for (int i = 1; i <= 10; i++) {
            //发送消息
            String message = "你好！发布订阅模式==="+i;
            /**
             * 参数一：交换机名称，如果没有指定则使用默认Deafult Exchange
             * 参数二：路由key，简单模式可以传递队列名称
             * 参数三：消息其他属性
             * 参数四：消息内容
             */
            channel.basicPublish(FANOUT_EXCHANGE, "", null, message.getBytes());
            System.out.println("已发送消息：" + message);
        }
        //释放资源
        channel.close();
        connection.close();

    }
}

3）消费者1：

public class Consumer1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接
        Connection connection = ConnectionUtil.getConnection();
        //创建频道
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare(Producer.FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT);
        //创建队列，并设置消息处理
        channel.queueDeclare(Producer.FANOUT_QUEUE_1, true, false, false, null);
        //队列绑定交换机
        channel.queueBind(Producer.FANOUT_QUEUE_1, Producer.FANOUT_EXCHANGE, "");
        //监听消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag：消息标签，在 channel.basicConsume 的时候可以指定
             * envelope：消息包内容，可从中获取消息id，消息routingkey，交换机，消息和重转标记（收到消息失败后是否需要重新发送）
             * properties：消息属性
             * body：消息
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("");
                System.out.println("=========================消费者1开始=====================================");
                System.out.println("");
                //路由key
                System.out.println("路由key为："+envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为："+envelope.getExchange());
                //消息id
                System.out.println("消息id为："+envelope.getDeliveryTag());
                //收到的消息
                System.out.println("消费者1__接收到的消息：" + new String(body, "UTF-8"));
                System.out.println("");
                System.out.println("===============================消费者1结束===============================");
                System.out.println("");
                //为了看到效果，让当前线程休眠
                try {
                    Thread.sleep(1000);//休眠1秒钟
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        /**
         * 监听消息
         * 参数一：队列名称
         * 参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到回复消息后会删除消息；设置为false则需要手动确认
         */
        channel.basicConsume(Producer.FANOUT_QUEUE_1,true,consumer);
        //不关闭资源，应该一直监听消息
//        channel.close();
//        connection.close();
    }
}

4）消费者2：

public class Consumer2 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接
        Connection connection = ConnectionUtil.getConnection();
        //创建频道
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare(Producer.FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT);
        //创建队列，并设置消息处理
        channel.queueDeclare(Producer.FANOUT_QUEUE_2, true, false, false, null);
        //队列绑定交换机
        channel.queueBind(Producer.FANOUT_QUEUE_2, Producer.FANOUT_EXCHANGE, "");
        //监听消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag：消息标签，在 channel.basicConsume 的时候可以指定
             * envelope：消息包内容，可从中获取消息id，消息routingkey，交换机，消息和重转标记（收到消息失败后是否需要重新发送）
             * properties：消息属性
             * body：消息
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("");
                System.out.println("=========================消费者2开始=====================================");
                System.out.println("");
                //路由key
                System.out.println("路由key为："+envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为："+envelope.getExchange());
                //消息id
                System.out.println("消息id为："+envelope.getDeliveryTag());
                //收到的消息
                System.out.println("消费者2__接收到的消息：" + new String(body, "UTF-8"));
                System.out.println("");
                System.out.println("===============================消费者2结束===============================");
                System.out.println("");
                //为了看到效果，让当前线程休眠
                try {
                    Thread.sleep(1000);//休眠1秒钟
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        /**
         * 监听消息
         * 参数一：队列名称
         * 参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到回复消息后会删除消息；设置为false则需要手动确认
         */
        channel.basicConsume(Producer.FANOUT_QUEUE_2,true,consumer);
        //不关闭资源，应该一直监听消息
//        channel.close();
//        connection.close();
    }
}

测试结果

消费者1：

消费者2：

小结

交换机需要与队列进行绑定，绑定之后，一个消息可以被多个消费者都收到。

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

1. 工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机；
2. 发布/订阅模式的生产者是向交换机发送消息，工作队列模式的生产者是面向队列发送消息（底层使用默认交换机）；
3. 发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑定到默认的交换机上。

路由模式（Routing）

简介

路由模式的特点：

• 队列和交换机绑定，不能是任意绑定了，而是指定一个Routing Key（路由key）；
• 生产者在向Exchange发送消息时，也必须指定消息的Routing Key；
• Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routing Key与消息的Routing Key完全一致，才会接收到消息。

• P：生产者，向Exchange发送消息，发送消息时，会指定一个Routing Key；
• X：Exchange（交换机），接收生产者的消息，然后把消息递交给与Routing Key完全匹配的队列；
• C1：消费者，其所在队列指定了需要Routing Key为error的消息；
• C2：消费者，其所在队列指定了需要Routing Key为info、error、warning的消息；

路由模式与发布订阅模式的区别：

1. 路由模式中交换机的类型为：Direct，发布订阅模式中交换机类型为Fanout；
2. 路由模式队列绑定交换机的时候，需要指定Routing Key，发布订阅模式不需要。

代码示例

1）生产者：

/**
 * 路由模式使用的交换机类型为：direct
 */
public class Producer {
    //交换机名称
    static final String DIRECT_EXCHANGE = "direct_exchange";
    //队列名称
    static final String DIRECT_QUEUE_INSERT = "direct_queue_insert";
    //队列名称
    static final String DIRECT_QUEUE_UPDATE = "direct_queue_update";

    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接
        Connection connection = ConnectionUtil.getConnection();
        //创建频道
        Channel channel = connection.createChannel();
        /**
         * 声明交换机
         * 参数一：交换机名称
         * 参数二：交换机类型，fanout、topic、direct、headers
         */
        channel.exchangeDeclare(DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT);
        //声明（创建）队列
        /**
         * 参数一：队列名称
         * 参数二：是否定义持久化队列
         * 参数三：是否独占本次连接
         * 参数四：是否在不使用的时候自动删除队列
         * 参数五：队列其他参数
         */
        channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null);
        channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null);
        //队列绑定交换机
        /**
         * 参数一：队列名称
         * 参数二：队列需要绑定的交换机名称
         * 参数三：路由key
         */
        channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHANGE, "insert");
        channel.queueBind(DIRECT_QUEUE_UPDATE, DIRECT_EXCHANGE, "update");
        //发送消息
        String message = "你好！路由模式===新增了商品。routing key 为 insert";
        /**
         * 参数一：交换机名称，如果没有指定则使用默认Deafult Exchange
         * 参数二：路由key，简单模式可以传递队列名称
         * 参数三：消息其他属性
         * 参数四：消息内容
         */
        channel.basicPublish(DIRECT_EXCHANGE, "insert", null, message.getBytes());
        System.out.println("已发送消息：" + message);
        //发送消息
        message = "你好！路由模式===修改了商品。routing key 为 update";
        /**
         * 参数一：交换机名称，如果没有指定则使用默认Deafult Exchange
         * 参数二：路由key，简单模式可以传递队列名称
         * 参数三：消息其他属性
         * 参数四：消息内容
         */
        channel.basicPublish(DIRECT_EXCHANGE, "update", null, message.getBytes());
        System.out.println("已发送消息：" + message);
        //释放资源
        channel.close();
        connection.close();

    }
}

2）消费者1：

public class Consumer1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接
        Connection connection = ConnectionUtil.getConnection();
        //创建频道
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare(Producer.DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT);
        //创建队列，并设置消息处理
        channel.queueDeclare(Producer.DIRECT_QUEUE_INSERT, true, false, false, null);
        //队列绑定交换机
        channel.queueBind(Producer.DIRECT_QUEUE_INSERT, Producer.DIRECT_EXCHANGE, "insert");
        //监听消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag：消息标签，在 channel.basicConsume 的时候可以指定
             * envelope：消息包内容，可从中获取消息id，消息routingkey，交换机，消息和重转标记（收到消息失败后是否需要重新发送）
             * properties：消息属性
             * body：消息
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("");
                System.out.println("=========================消费者1开始=====================================");
                System.out.println("");
                //路由key
                System.out.println("路由key为："+envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为："+envelope.getExchange());
                //消息id
                System.out.println("消息id为："+envelope.getDeliveryTag());
                //收到的消息
                System.out.println("消费者1__接收到的消息：" + new String(body, "UTF-8"));
                System.out.println("");
                System.out.println("===============================消费者1结束===============================");
                System.out.println("");
                //为了看到效果，让当前线程休眠
                try {
                    Thread.sleep(1000);//休眠1秒钟
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        /**
         * 监听消息
         * 参数一：队列名称
         * 参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到回复消息后会删除消息；设置为false则需要手动确认
         */
        channel.basicConsume(Producer.DIRECT_QUEUE_INSERT,true,consumer);
        //不关闭资源，应该一直监听消息
//        channel.close();
//        connection.close();
    }
}

3）消费者2：

public class Consumer2 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接
        Connection connection = ConnectionUtil.getConnection();
        //创建频道
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare(Producer.DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT);
        //创建队列，并设置消息处理
        channel.queueDeclare(Producer.DIRECT_QUEUE_UPDATE, true, false, false, null);
        //队列绑定交换机
        channel.queueBind(Producer.DIRECT_QUEUE_UPDATE, Producer.DIRECT_EXCHANGE, "update");
        //监听消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag：消息标签，在 channel.basicConsume 的时候可以指定
             * envelope：消息包内容，可从中获取消息id，消息routingkey，交换机，消息和重转标记（收到消息失败后是否需要重新发送）
             * properties：消息属性
             * body：消息
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("");
                System.out.println("=========================消费者2开始=====================================");
                System.out.println("");
                //路由key
                System.out.println("路由key为："+envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为："+envelope.getExchange());
                //消息id
                System.out.println("消息id为："+envelope.getDeliveryTag());
                //收到的消息
                System.out.println("消费者2__接收到的消息：" + new String(body, "UTF-8"));
                System.out.println("");
                System.out.println("===============================消费者2结束===============================");
                System.out.println("");
                //为了看到效果，让当前线程休眠
                try {
                    Thread.sleep(1000);//休眠1秒钟
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        /**
         * 监听消息
         * 参数一：队列名称
         * 参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到回复消息后会删除消息；设置为false则需要手动确认
         */
        channel.basicConsume(Producer.DIRECT_QUEUE_UPDATE,true,consumer);
        //不关闭资源，应该一直监听消息
//        channel.close();
//        connection.close();
    }
}

测试结果

消费者1：

消费者2：

小结

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

主题模式（Topics）

简介

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

Routing Key一般都是由一个或多个单词组成，多个单词之间以“.”分隔，例如：item.insert。

通配符规则：

• #：匹配一个或多个词；
• *：只匹配一个词。

举例：

• item.#：能够匹配 item.insert.abc 或者 item.insert；
• item.*：只能匹配item.insert。

图解：

• 红色Queue：绑定的是usa.#，因此凡是以usa.开头的routing key都会被匹配到；
• 黄色Queue：绑定的是#.news，因此凡是以.news结尾的routing key都会被匹配。

代码示例

1）生产者：

/**
 * 主题模式(通配符topics)使用的交换机类型为：topic
 */
public class Producer {
    //交换机名称
    static final String TOPIC_EXCHANGE = "topic_exchange";
    //队列名称
    static final String TOPIC_QUEUE_1 = "topic_queue_1";
    //队列名称
    static final String TOPIC_QUEUE_2 = "topic_queue_2";

    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接
        Connection connection = ConnectionUtil.getConnection();
        //创建频道
        Channel channel = connection.createChannel();
        /**
         * 声明交换机
         * 参数一：交换机名称
         * 参数二：交换机类型，fanout、topic、direct、headers
         */
        channel.exchangeDeclare(TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC);

        //声明（创建）队列
        /**
         * 参数一：队列名称
         * 参数二：是否定义持久化队列
         * 参数三：是否独占本次连接
         * 参数四：是否在不使用的时候自动删除队列
         * 参数五：队列其他参数
         */
        channel.queueDeclare(TOPIC_QUEUE_1, true, false, false, null);
        channel.queueDeclare(TOPIC_QUEUE_2, true, false, false, null);
        //队列绑定交换机
        /**
         * 参数一：队列名称
         * 参数二：队列需要绑定的交换机名称
         * 参数三：路由key
         */
        channel.queueBind(TOPIC_QUEUE_1, TOPIC_EXCHANGE, "item.#");
        channel.queueBind(TOPIC_QUEUE_2, TOPIC_EXCHANGE, "*.delete");

        //发送消息
        String message = "你好！Topics模式===新增了商品。routing key 为 item.insert";
        /**
         * 参数一：交换机名称，如果没有指定则使用默认Deafult Exchange
         * 参数二：路由key，简单模式可以传递队列名称
         * 参数三：消息其他属性
         * 参数四：消息内容
         */
        channel.basicPublish(TOPIC_EXCHANGE, "item.insert", null, message.getBytes());
        System.out.println("已发送消息：" + message);
        //发送消息
        message = "你好！Topics模式===修改了商品。routing key 为 item.update";
        channel.basicPublish(TOPIC_EXCHANGE, "item.update", null, message.getBytes());
        System.out.println("已发送消息：" + message);
        //发送消息
        message = "你好！Topics模式===删除了商品。routing key 为 product.delete";
        channel.basicPublish(TOPIC_EXCHANGE, "product.delete", null, message.getBytes());
        System.out.println("已发送消息：" + message);
        //发送消息
        message = "其他信息发送过来了。。。。。。。routing key 为 item.delete.abc";
        channel.basicPublish(TOPIC_EXCHANGE, "item.delete.abc", null, message.getBytes());
        System.out.println("已发送消息：" + message);
        //释放资源
        channel.close();
        connection.close();

    }
}

2）消费者1：

public class Consumer1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接
        Connection connection = ConnectionUtil.getConnection();
        //创建频道
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare(Producer.TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC);
        //创建队列，并设置消息处理
        channel.queueDeclare(Producer.TOPIC_QUEUE_1, true, false, false, null);
        //队列绑定交换机
        channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHANGE, "item.#");
        //监听消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag：消息标签，在 channel.basicConsume 的时候可以指定
             * envelope：消息包内容，可从中获取消息id，消息routingkey，交换机，消息和重转标记（收到消息失败后是否需要重新发送）
             * properties：消息属性
             * body：消息
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("");
                System.out.println("=========================消费者1开始=====================================");
                System.out.println("");
                //路由key
                System.out.println("路由key为："+envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为："+envelope.getExchange());
                //消息id
                System.out.println("消息id为："+envelope.getDeliveryTag());
                //收到的消息
                System.out.println("消费者1__接收到的消息：" + new String(body, "UTF-8"));
                System.out.println("");
                System.out.println("===============================消费者1结束===============================");
                System.out.println("");
                //为了看到效果，让当前线程休眠
                try {
                    Thread.sleep(1000);//休眠1秒钟
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        /**
         * 监听消息
         * 参数一：队列名称
         * 参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到回复消息后会删除消息；设置为false则需要手动确认
         */
        channel.basicConsume(Producer.TOPIC_QUEUE_1,true,consumer);
        //不关闭资源，应该一直监听消息
//        channel.close();
//        connection.close();
    }
}

3）消费者2：

public class Consumer2 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接
        Connection connection = ConnectionUtil.getConnection();
        //创建频道
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare(Producer.TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC);
        //创建队列，并设置消息处理
        channel.queueDeclare(Producer.TOPIC_QUEUE_2, true, false, false, null);
        //队列绑定交换机
        channel.queueBind(Producer.TOPIC_QUEUE_2, Producer.TOPIC_EXCHANGE, "*.delete");
        //监听消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag：消息标签，在 channel.basicConsume 的时候可以指定
             * envelope：消息包内容，可从中获取消息id，消息routingkey，交换机，消息和重转标记（收到消息失败后是否需要重新发送）
             * properties：消息属性
             * body：消息
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("");
                System.out.println("=========================消费者2开始=====================================");
                System.out.println("");
                //路由key
                System.out.println("路由key为："+envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为："+envelope.getExchange());
                //消息id
                System.out.println("消息id为："+envelope.getDeliveryTag());
                //收到的消息
                System.out.println("消费者2__接收到的消息：" + new String(body, "UTF-8"));
                System.out.println("");
                System.out.println("===============================消费者2结束===============================");
                System.out.println("");
                //为了看到效果，让当前线程休眠
                try {
                    Thread.sleep(1000);//休眠1秒钟
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        /**
         * 监听消息
         * 参数一：队列名称
         * 参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到回复消息后会删除消息；设置为false则需要手动确认
         */
        channel.basicConsume(Producer.TOPIC_QUEUE_2,true,consumer);
        //不关闭资源，应该一直监听消息
//        channel.close();
//        connection.close();
    }
}

测试结果

消费者1：

消费者2：

小结

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

远程调用模式（RPC）

1. 当客户端启动的时候，创建一个匿名的独享的回调队列。
2. 在RPC请求中，客户端发送带有两个属性的消息：一个是设置回调队列的 reply_to 属性，另一个是设置唯一值的 correlation_id 属性。
3. 该请求被发送到rpc_queue队列。
4. RPC工作者（又名：服务器）等待请求发送到这个队列中来。当请求出现的时候，它执行他的工作并且将带有执行结果的消息发送给reply_to字段指定的队列。
5. 客户端等待回调队列里的数据。当有消息出现的时候，它会检查correlation_id属性。如果此属性的值与请求匹配，将它返回给应用。

模式总结

RabbitMQ工作模式：

1. 简单模式 HelloWorld
   • 一个生产者，一个消费者，不需要设置交换机（使用默认交换机）；
2. 工作队列模式 Work Queue
   • 一个生产者，多个消费者（竞争关系），不需要设置交换机（使用默认交换机）；
3. 发布订阅模式 Plublish/Subscribe
   • 需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当消息到交换机后，交换机会将消息发送到绑定的队列；
4. 路由模式 Routing
   • 需要设置类型为direct的交换机，交换机与队列进行绑定，并且指定Routing Key，当发送消息到交换机后，交换机会根据Routing Key将消息发送到对应的队列；
5. 主题模式 Topics
   • 需要设置类型为topic的交换机，交换机与队列进行绑定，并且指定通配符方式的Routing Key，当发送消息到交换机后，交换机会根据Routing Key将消息发送到对应的队列；

最后更新： 2020年08月22日 10:40

原始链接： http://ligangit.com/2020/08/21/RabbitMQ-工作模式/

Li Gang

赏

× 请我吃糖~