本文RabbitMQ版本为rabbitmq-server-3.7.17,erlang为erlang-22.0.7.请各位去官网查看版本匹配和下载,也可以留言,我发安装包

首先我们看一下官网的图

 

 

 

 

 

 上篇文章中我们介绍了第一个简单模式,本文我们来介绍其他2-5的模式,开始吧

 

 

 

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

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

Work Queues与入门程序的简单模式的代码是几乎一样的；可以完全复制，并复制多一个消费者进行多个消费者同时消费消息的测试。

1.消息生产者

  1. import com.itheima.rabbitmq.util.ConnectionUtil;
  2. import com.rabbitmq.client.Channel;
  3. import com.rabbitmq.client.Connection;
  5. public class Producer {
  7.     static final String QUEUE_NAME = "work_queue";
  9.     public static void main(String[] args) throws Exception{
  10.         //创建连接
  11.         Connection connection = ConnectionUtil.getConnection();
  12.         //创建频道
  13.         Channel channel = connection.createChannel();
  14.         //声明（创建）队列
  15.         /*
  16.         参数一：队列名称
  17.         参数二：是否定义持久化队列
  18.         参数三：是否独占本次连接
  19.         参数四：是否在不使用的时候自动删除队列
  20.         参数五：队列其他参数
  21.          */
  22.         channel.queueDeclare(QUEUE_NAME,true,false,false,null);
  23.         //要发送的消息
  24.         for(int i=1;i<=30;i++){
  25.             String message = "work_queue------->"+i;
  26.             channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
  27.             System.out.println("已发送消息："+message);
  28.         }
  29.         //释放资源
  30.         channel.close();
  31.         connection.close();
  32.     }
  34. }

执行看结果

 

 

2.消息消费者(多个,这里演示两个)

  1. import com.itheima.rabbitmq.util.ConnectionUtil;
  2. import com.rabbitmq.client.*;
  3. import java.io.IOException;
  5. public class Consumer1 {
  7.     public static void main(String[] args) throws Exception {
  8.         //获取连接
  9.         Connection connection = ConnectionUtil.getConnection();
  10.         //创建频道
  11.         Channel channel = connection.createChannel();
  12.         //创建队列：并设置消息处理
  13.         channel.queueDeclare(Producer.QUEUE_NAME,true,false,false,null);
  14.         //监听消息
  15.         DefaultConsumer consumer = new DefaultConsumer(channel) {
  16.             @Override
  17.             /*
  18.             consumerTag ：消息者标签，在channel.basicConsume时候可以指定
  19.             envelope: 消息包内容，可从中获取消息id，消息routingkey，交换机，消息和重转标记（收到消息失败后是否需要重新发送）
  20.             properties: 消息属性
  21.             body： 消息
  22.              */
  23.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  24.                 System.out.println("");
  25.                 System.out.println("=======================消费者1开始=========================================");
  26.                 System.out.println("");
  27.                 //路由key
  28.                 System.out.println("路由key为：" + envelope.getRoutingKey());
  29.                 //交换机
  30.                 System.out.println("交换机为：" + envelope.getExchange());
  31.                 //消息id
  32.                 System.out.println("消息id为：" + envelope.getDeliveryTag());
  33.                 //收到的消息
  34.                 System.out.println("接收到的消息：" + new String(body, "UTF-8"));
  35.                 System.out.println("");
  36.                 System.out.println("======================消费者1结束==========================================");
  37.                 System.out.println("");
  38.                 try {
  39.                     Thread.sleep(1000);//休眠一秒钟
  40.                 } catch (InterruptedException e) {
  41.                     e.printStackTrace();
  42.                 }
  43.             }
  44.         };
  45.         /*
  46.         监听消息
  47.         参数一：队列名称
  48.         参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到回复后会删除消息；设置为false则需要手动确认
  49.          */
  50.         channel.basicConsume(Producer.QUEUE_NAME, true, consumer);
  52.     }
  53. }

第二个

  1. import com.itheima.rabbitmq.util.ConnectionUtil;
  2. import com.rabbitmq.client.*;
  3. import java.io.IOException;
  5. public class Consumer2 {
  7.     public static void main(String[] args) throws Exception {
  8.         //获取连接
  9.         Connection connection = ConnectionUtil.getConnection();
  10.         //创建频道
  11.         Channel channel = connection.createChannel();
  12.         //创建队列：并设置消息处理
  13.         channel.queueDeclare(Producer.QUEUE_NAME,true,false,false,null);
  14.         //监听消息
  15.         DefaultConsumer consumer = new DefaultConsumer(channel) {
  16.             @Override
  17.             /*
  18.             consumerTag ：消息者标签，在channel.basicConsume时候可以指定
  19.             envelope: 消息包内容，可从中获取消息id，消息routingkey，交换机，消息和重转标记（收到消息失败后是否需要重新发送）
  20.             properties: 消息属性
  21.             body： 消息
  22.              */
  23.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  24.                 System.out.println("");
  25.                 System.out.println("=====================消费者2开始===========================================");
  26.                 System.out.println("");
  27.                 //路由key
  28.                 System.out.println("路由key为：" + envelope.getRoutingKey());
  29.                 //交换机
  30.                 System.out.println("交换机为：" + envelope.getExchange());
  31.                 //消息id
  32.                 System.out.println("消息id为：" + envelope.getDeliveryTag());
  33.                 //收到的消息
  34.                 System.out.println("接收到的消息：" + new String(body, "UTF-8"));
  35.                 System.out.println("");
  36.                 System.out.println("======================消费者2结束==========================================");
  37.                 System.out.println("");
  38.                 try {
  39.                     Thread.sleep(1000);//休眠一秒钟
  40.                 } catch (InterruptedException e) {
  41.                     e.printStackTrace();
  42.                 }
  43.             }
  44.         };
  45.         /*
  46.         监听消息
  47.         参数一：队列名称
  48.         参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到回复后会删除消息；设置为false则需要手动确认
  49.          */
  50.         channel.basicConsume(Producer.QUEUE_NAME, true, consumer);
  52.     }
  53. }

启动两个消费者，然后再启动生产者发送消息；到IDEA的两个消费者对应的控制台查看是否竞争性的接收到消息.查看结果

 

 

 

 

 

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

 

 

 

 

前面2个案例中，只有3个角色：

  • P：生产者，也就是要发送消息的程序
  • C：消费者：消息的接受者，会一直等待消息到来。
  • queue：消息队列，图中红色部分

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

  • P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）

  • C：消费者，消息的接受者，会一直等待消息到来。

  • Queue：消息队列，接收消息、缓存消息。

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

    1. Fanout：广播，直接将消息路由到所有绑定的队列中，无须对消息的routingkey进行匹配操作。
    2. Direct：定向，把消息交给符合指定routing key 的队列.交换机在和queue进行binding时会设置routingkey，将消息发送到交换机时会设置对应的routingkey，只有这两个routingkey完全相同，交换机才会选择对应的binging进行消息路由。
    3. Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列.

      此类型和direct类型差不多，但direct类型要求routingkey完全相等，这里的routingkey可以有通配符：\’*\’,\’#\’。

      其中\’*\’表示匹配一个单词， \’#\’则表示匹配没有或者多个单词

    4. header：其路由的规则是根据header来判断，其中的header就是binding时的arguments参数：header类型用的比较少

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

总结：一般direct和topic用来具体的路由消息，如果要用广播的消息一般用fanout。header类型用的比较少

1.消息生产者

  1. import com.itheima.rabbitmq.util.ConnectionUtil;
  2. import com.rabbitmq.client.BuiltinExchangeType;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  6. /**
  7.  * 发布与订阅使用的交换机类型为：fanout
  8.  */
  9. public class Producer {
  11.     //交换机名称
  12.     static final String FANOUT_EXCHAGE = "fanout_exchange";
  13.     //队列名称
  14.     static final String FANOUT_QUEUE_1 = "fanout_queue_1";
  15.     //队列名称
  16.     static final String FANOUT_QUEUE_2 = "fanout_queue_2";
  18.     public static void main(String[] args) throws Exception {
  19.         //创建连接
  20.         Connection connection = ConnectionUtil.getConnection();
  21.         // 创建频道
  22.         Channel channel = connection.createChannel();
  23.         /**
  24.          * 声明交换机
  25.          * 参数1：交换机名称
  26.          * 参数2：交换机类型，fanout、topic、direct、headers
  27.          */
  28.         channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
  30.         // 声明（创建）队列
  31.         /**
  32.          * 参数1：队列名称
  33.          * 参数2：是否定义持久化队列
  34.          * 参数3：是否独占本次连接
  35.          * 参数4：是否在不使用的时候自动删除队列
  36.          * 参数5：队列其它参数
  37.          */
  38.         channel.queueDeclare(FANOUT_QUEUE_1, true, false, false, null);
  39.         channel.queueDeclare(FANOUT_QUEUE_2, true, false, false, null);
  41.         //队列绑定交换机
  42.         channel.queueBind(FANOUT_QUEUE_1, FANOUT_EXCHAGE, "");
  43.         channel.queueBind(FANOUT_QUEUE_2, FANOUT_EXCHAGE, "");
  45.         for (int i = 1; i <= 10; i++) {
  46.             // 发送信息
  47.             String message = "你好；rabbitmq ！发布订阅模式--" + i;
  48.             /**
  49.              * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
  50.              * 参数2：路由key,简单模式可以传递队列名称
  51.              * 参数3：消息其它属性
  52.              * 参数4：消息内容
  53.              */
  54.             channel.basicPublish(FANOUT_EXCHAGE, "", null, message.getBytes());
  55.             System.out.println("已发送消息：" + message);
  56.         }
  58.         // 关闭资源
  59.         channel.close();
  60.         connection.close();
  61.     }
  62. }

执行,看结果

 

 

 

2.消息消费者(两个,分别在两个队列)

  1. import com.itheima.rabbitmq.util.ConnectionUtil;
  2. import com.rabbitmq.client.*;
  3. import java.io.IOException;
  5. public class Consumer1 {
  7.     public static void main(String[] args) throws Exception {
  8.         Connection connection = ConnectionUtil.getConnection();
  10.         // 创建频道
  11.         Channel channel = connection.createChannel();
  13.         //声明交换机
  14.         channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
  16.         // 声明（创建）队列
  17.         /**
  18.          * 参数1：队列名称
  19.          * 参数2：是否定义持久化队列
  20.          * 参数3：是否独占本次连接
  21.          * 参数4：是否在不使用的时候自动删除队列
  22.          * 参数5：队列其它参数
  23.          */
  24.         channel.queueDeclare(Producer.FANOUT_QUEUE_1, true, false, false, null);
  26.         //队列绑定交换机
  27.         channel.queueBind(Producer.FANOUT_QUEUE_1, Producer.FANOUT_EXCHAGE, "");
  29.         //创建消费者；并设置消息处理
  30.         DefaultConsumer consumer = new DefaultConsumer(channel){
  31.             @Override
  32.             /**
  33.              * consumerTag 消息者标签，在channel.basicConsume时候可以指定
  34.              * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
  35.              * properties 属性信息
  36.              * body 消息
  37.              */
  38.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  39.                 //路由key
  40.                 System.out.println("路由key为：" + envelope.getRoutingKey());
  41.                 //交换机
  42.                 System.out.println("交换机为：" + envelope.getExchange());
  43.                 //消息id
  44.                 System.out.println("消息id为：" + envelope.getDeliveryTag());
  45.                 //收到的消息
  46.                 System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));
  47.             }
  48.         };
  49.         //监听消息
  50.         /**
  51.          * 参数1：队列名称
  52.          * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
  53.          * 参数3：消息接收到后回调
  54.          */
  55.         channel.basicConsume(Producer.FANOUT_QUEUE_1, true, consumer);
  56.     }
  57. }

第二个

  1. import com.itheima.rabbitmq.util.ConnectionUtil;
  2. import com.rabbitmq.client.*;
  3. import java.io.IOException;
  5. public class Consumer2 {
  7.     public static void main(String[] args) throws Exception {
  8.         Connection connection = ConnectionUtil.getConnection();
  10.         // 创建频道
  11.         Channel channel = connection.createChannel();
  13.         //声明交换机
  14.         channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
  16.         // 声明（创建）队列
  17.         /**
  18.          * 参数1：队列名称
  19.          * 参数2：是否定义持久化队列
  20.          * 参数3：是否独占本次连接
  21.          * 参数4：是否在不使用的时候自动删除队列
  22.          * 参数5：队列其它参数
  23.          */
  24.         channel.queueDeclare(Producer.FANOUT_QUEUE_2, true, false, false, null);
  26.         //队列绑定交换机
  27.         channel.queueBind(Producer.FANOUT_QUEUE_2, Producer.FANOUT_EXCHAGE, "");
  29.         //创建消费者；并设置消息处理
  30.         DefaultConsumer consumer = new DefaultConsumer(channel){
  31.             @Override
  32.             /**
  33.              * consumerTag 消息者标签，在channel.basicConsume时候可以指定
  34.              * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
  35.              * properties 属性信息
  36.              * body 消息
  37.              */
  38.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  39.                 //路由key
  40.                 System.out.println("路由key为：" + envelope.getRoutingKey());
  41.                 //交换机
  42.                 System.out.println("交换机为：" + envelope.getExchange());
  43.                 //消息id
  44.                 System.out.println("消息id为：" + envelope.getDeliveryTag());
  45.                 //收到的消息
  46.                 System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8"));
  47.             }
  48.         };
  49.         //监听消息
  50.         /**
  51.          * 参数1：队列名称
  52.          * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
  53.          * 参数3：消息接收到后回调
  54.          */
  55.         channel.basicConsume(Producer.FANOUT_QUEUE_2, true, consumer);
  56.     }
  57. }

3.测试

启动所有消费者，然后使用生产者发送消息；在每个消费者对应的控制台可以查看到生产者发送的所有消息；到达广播的效果。

 

 

 在执行完测试代码后，其实到RabbitMQ的管理后台找到Exchanges选项卡，点击 fanout_exchange 的交换机，可以查看到如下的绑定

 

 

 

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

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

1、工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机。

2、发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息(底层使用默认交换机)。

3、发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑定到默认的交换机 。

 

 

 

 

图解：

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

路由模式特点：

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

在编码上与 Publish/Subscribe发布与订阅模式 的区别是交换机的类型为：Direct，还有队列绑定交换机的时候需要指定routing key

  1. import com.itheima.rabbitmq.util.ConnectionUtil;
  2. import com.rabbitmq.client.BuiltinExchangeType;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  6. /**
  7.  * 路由模式的交换机类型为：direct
  8.  */
  9. public class Producer {
  11.     //交换机名称
  12.     static final String DIRECT_EXCHAGE = "direct_exchange";
  13.     //队列名称
  14.     static final String DIRECT_QUEUE_INSERT = "direct_queue_insert";
  15.     //队列名称
  16.     static final String DIRECT_QUEUE_UPDATE = "direct_queue_update";
  18.     public static void main(String[] args) throws Exception {
  19.         //创建连接
  20.         Connection connection = ConnectionUtil.getConnection();
  21.         // 创建频道
  22.         Channel channel = connection.createChannel();
  23.         /**
  24.          * 声明交换机
  25.          * 参数1：交换机名称
  26.          * 参数2：交换机类型，fanout、topic、direct、headers
  27.          */
  28.         channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
  30.         // 声明（创建）队列
  31.         /**
  32.          * 参数1：队列名称
  33.          * 参数2：是否定义持久化队列
  34.          * 参数3：是否独占本次连接
  35.          * 参数4：是否在不使用的时候自动删除队列
  36.          * 参数5：队列其它参数
  37.          */
  38.         channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null);
  39.         channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null);
  41.         //队列绑定交换机
  42.         channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHAGE, "insert");
  43.         channel.queueBind(DIRECT_QUEUE_UPDATE, DIRECT_EXCHAGE, "update");
  45.         // 发送信息
  46.         String message = "新增了商品。路由模式；routing key 为 insert " ;
  47.         /**
  48.          * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
  49.          * 参数2：路由key,简单模式可以传递队列名称
  50.          * 参数3：消息其它属性
  51.          * 参数4：消息内容
  52.          */
  53.         channel.basicPublish(DIRECT_EXCHAGE, "insert", null, message.getBytes());
  54.         System.out.println("已发送消息：" + message);
  56.         // 发送信息
  57.         message = "修改了商品。路由模式；routing key 为 update" ;
  58.         /**
  59.          * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
  60.          * 参数2：路由key,简单模式可以传递队列名称
  61.          * 参数3：消息其它属性
  62.          * 参数4：消息内容
  63.          */
  64.         channel.basicPublish(DIRECT_EXCHAGE, "update", null, message.getBytes());
  65.         System.out.println("已发送消息：" + message);
  67.         // 关闭资源
  68.         channel.close();
  69.         connection.close();
  70.     }
  71. }

执行看结果

 

 

 

 

 

 

 

 

 

 

  1. public class Consumer1 {
  3.     public static void main(String[] args) throws Exception {
  4.         //获取连接
  5.         Connection connection = ConnectionUtil.getConnection();
  6.         // 创建频道
  7.         Channel channel = connection.createChannel();
  8.         //声明交换机
  9.         channel.exchangeDeclare(Producer.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
  11.         // 声明（创建）队列
  12.         /**
  13.          * 参数1：队列名称
  14.          * 参数2：是否定义持久化队列
  15.          * 参数3：是否独占本次连接
  16.          * 参数4：是否在不使用的时候自动删除队列
  17.          * 参数5：队列其它参数
  18.          */
  19.         channel.queueDeclare(Producer.DIRECT_QUEUE_INSERT, true, false, false, null);
  21.         //队列绑定交换机
  22.         channel.queueBind(Producer.DIRECT_QUEUE_INSERT, Producer.DIRECT_EXCHAGE, "insert");
  24.         //创建消费者；并设置消息处理
  25.         DefaultConsumer consumer = new DefaultConsumer(channel){
  26.             @Override
  27.             /**
  28.              * consumerTag 消息者标签，在channel.basicConsume时候可以指定
  29.              * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
  30.              * properties 属性信息
  31.              * body 消息
  32.              */
  33.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  34.                 //路由key
  35.                 System.out.println("路由key为：" + envelope.getRoutingKey());
  36.                 //交换机
  37.                 System.out.println("交换机为：" + envelope.getExchange());
  38.                 //消息id
  39.                 System.out.println("消息id为：" + envelope.getDeliveryTag());
  40.                 //收到的消息
  41.                 System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));
  42.             }
  43.         };
  44.         //监听消息
  45.         /**
  46.          * 参数1：队列名称
  47.          * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
  48.          * 参数3：消息接收到后回调
  49.          */
  50.         channel.basicConsume(Producer.DIRECT_QUEUE_INSERT, true, consumer);
  51.     }
  52. }

执行,看结果

 

 可以看到insert的那条消息被消费了

  1. public class Consumer2 {
  3.     public static void main(String[] args) throws Exception {
  4.         //创建连接
  5.         Connection connection = ConnectionUtil.getConnection();
  7.         // 创建频道
  8.         Channel channel = connection.createChannel();
  10.         //声明交换机
  11.         channel.exchangeDeclare(Producer.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
  13.         // 声明（创建）队列
  14.         /**
  15.          * 参数1：队列名称
  16.          * 参数2：是否定义持久化队列
  17.          * 参数3：是否独占本次连接
  18.          * 参数4：是否在不使用的时候自动删除队列
  19.          * 参数5：队列其它参数
  20.          */
  21.         channel.queueDeclare(Producer.DIRECT_QUEUE_UPDATE, true, false, false, null);
  23.         //队列绑定交换机
  24.         channel.queueBind(Producer.DIRECT_QUEUE_UPDATE, Producer.DIRECT_EXCHAGE, "update");
  26.         //创建消费者；并设置消息处理
  27.         DefaultConsumer consumer = new DefaultConsumer(channel){
  28.             @Override
  29.             /**
  30.              * consumerTag 消息者标签，在channel.basicConsume时候可以指定
  31.              * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
  32.              * properties 属性信息
  33.              * body 消息
  34.              */
  35.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  36.                 //路由key
  37.                 System.out.println("路由key为：" + envelope.getRoutingKey());
  38.                 //交换机
  39.                 System.out.println("交换机为：" + envelope.getExchange());
  40.                 //消息id
  41.                 System.out.println("消息id为：" + envelope.getDeliveryTag());
  42.                 //收到的消息
  43.                 System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8"));
  44.             }
  45.         };
  46.         //监听消息
  47.         /**
  48.          * 参数1：队列名称
  49.          * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
  50.          * 参数3：消息接收到后回调
  51.          */
  52.         channel.basicConsume(Producer.DIRECT_QUEUE_UPDATE, true, consumer);
  53.     }
  54. }

看结果

 

 在消费者对应的控制台可以查看到生产者发送对应routing key对应队列的消息；到达按照需要接收的效果。

Routing模式要求队列在绑定交换机时要指定routing key，消息会转发到符合routing key的队列。

 

 

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

 

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

通配符规则：

#：匹配一个或多个词

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

 

举例：

item.#：能够匹配item.insert.abc 或者 item.insert

item.*：只能匹配item.insert

  1. public class Producer {
  3.     //交换机名称
  4.     static final String TOPIC_EXCHAGE = "topic_exchange";
  5.     //队列名称
  6.     static final String TOPIC_QUEUE_1 = "topic_queue_1";
  7.     //队列名称
  8.     static final String TOPIC_QUEUE_2 = "topic_queue_2";
  10.     public static void main(String[] args) throws Exception {
  12.         //创建连接
  13.         Connection connection = ConnectionUtil.getConnection();
  15.         // 创建频道
  16.         Channel channel = connection.createChannel();
  18.         /**
  19.          * 声明交换机
  20.          * 参数1：交换机名称
  21.          * 参数2：交换机类型，fanout、topic、topic、headers
  22.          */
  23.         channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
  25.         // 发送信息
  26.         String message = "新增了商品。Topic模式；routing key 为 item.insert " ;
  27.         channel.basicPublish(TOPIC_EXCHAGE, "item.insert", null, message.getBytes());
  28.         System.out.println("已发送消息：" + message);
  30.         // 发送信息
  31.         message = "修改了商品。Topic模式；routing key 为 item.update" ;
  32.         channel.basicPublish(TOPIC_EXCHAGE, "item.update", null, message.getBytes());
  33.         System.out.println("已发送消息：" + message);
  35.         // 发送信息
  36.         message = "删除了商品。Topic模式；routing key 为 product.delete" ;
  37.         channel.basicPublish(TOPIC_EXCHAGE, "product.delete", null, message.getBytes());
  38.         System.out.println("已发送消息：" + message);
  40.         // 发送信息
  41.         message = "其他的信息发送过来了。。。。" ;
  42.         channel.basicPublish(TOPIC_EXCHAGE, "item.delete.abc", null, message.getBytes());
  43.         System.out.println("已发送消息：" + message);
  46.         // 关闭资源
  47.         channel.close();
  48.         connection.close();
  49.     }
  50. }

执行,看结果

 

 

 

 

  1. public class Consumer1 {
  3.     public static void main(String[] args) throws Exception {
  4.         //创建连接
  5.         Connection connection = ConnectionUtil.getConnection();
  7.         // 创建频道
  8.         Channel channel = connection.createChannel();
  10.         //声明交换机
  11.         channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
  13.         // 声明（创建）队列
  14.         /**
  15.          * 参数1：队列名称
  16.          * 参数2：是否定义持久化队列
  17.          * 参数3：是否独占本次连接
  18.          * 参数4：是否在不使用的时候自动删除队列
  19.          * 参数5：队列其它参数
  20.          */
  21.         channel.queueDeclare(Producer.TOPIC_QUEUE_1, true, false, false, null);
  23.         //队列绑定交换机
  24.         channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.#");
  25.         //channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.delete");
  27.         //创建消费者；并设置消息处理
  28.         DefaultConsumer consumer = new DefaultConsumer(channel){
  29.             @Override
  30.             /**
  31.              * consumerTag 消息者标签，在channel.basicConsume时候可以指定
  32.              * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
  33.              * properties 属性信息
  34.              * body 消息
  35.              */
  36.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  37.                 //路由key
  38.                 System.out.println("路由key为：" + envelope.getRoutingKey());
  39.                 //交换机
  40.                 System.out.println("交换机为：" + envelope.getExchange());
  41.                 //消息id
  42.                 System.out.println("消息id为：" + envelope.getDeliveryTag());
  43.                 //收到的消息
  44.                 System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));
  45.             }
  46.         };
  47.         //监听消息
  48.         /**
  49.          * 参数1：队列名称
  50.          * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
  51.          * 参数3：消息接收到后回调
  52.          */
  53.         channel.basicConsume(Producer.TOPIC_QUEUE_1, true, consumer);
  54.     }
  55. }

 

 可以看到,队列topic_queue_1里的消息都被消费了

  1. public class Consumer2 {
  3.     public static void main(String[] args) throws Exception {
  4.         //获取连接
  5.         Connection connection = ConnectionUtil.getConnection();
  7.         // 创建频道
  8.         Channel channel = connection.createChannel();
  10.         //声明交换机
  11.         channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
  13.         // 声明（创建）队列
  14.         /**
  15.          * 参数1：队列名称
  16.          * 参数2：是否定义持久化队列
  17.          * 参数3：是否独占本次连接
  18.          * 参数4：是否在不使用的时候自动删除队列
  19.          * 参数5：队列其它参数
  20.          */
  21.         channel.queueDeclare(Producer.TOPIC_QUEUE_2, true, false, false, null);
  23.         //队列绑定交换机
  24.         channel.queueBind(Producer.TOPIC_QUEUE_2, Producer.TOPIC_EXCHAGE, "*.delete");
  26.         //创建消费者；并设置消息处理
  27.         DefaultConsumer consumer = new DefaultConsumer(channel){
  28.             @Override
  29.             /**
  30.              * consumerTag 消息者标签，在channel.basicConsume时候可以指定
  31.              * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
  32.              * properties 属性信息
  33.              * body 消息
  34.              */
  35.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  36.                 //路由key
  37.                 System.out.println("路由key为：" + envelope.getRoutingKey());
  38.                 //交换机
  39.                 System.out.println("交换机为：" + envelope.getExchange());
  40.                 //消息id
  41.                 System.out.println("消息id为：" + envelope.getDeliveryTag());
  42.                 //收到的消息
  43.                 System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8"));
  44.             }
  45.         };
  46.         //监听消息
  47.         /**
  48.          * 参数1：队列名称
  49.          * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
  50.          * 参数3：消息接收到后回调
  51.          */
  52.         channel.basicConsume(Producer.TOPIC_QUEUE_2, true, consumer);
  53.     }
  54. }

执行,看结果

 

  可以看到,队列topic_queue_2里的消息也被消费了

在执行完测试代码后，其实到RabbitMQ的管理后台找到Exchanges选项卡，点击 topic_exchange 的交换机，可以查看到如下的绑定：

 

 Topic主题模式可以实现 Publish/Subscribe发布与订阅模式 和 Routing路由模式 的功能；只是Topic在配置routing key 的时候可以使用通配符，显得更加灵活。

 

RabbitMQ工作模式：

1、简单模式 HelloWorld 一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）

2、工作队列模式 Work Queue 一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）

3、发布订阅模式 Publish/subscribe 需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列

4、路由模式 Routing 需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

5、通配符模式 Topic 需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

 

版权声明：本文为lusaisai原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://www.cnblogs.com/lusaisai/p/13019368.html