5.1 Work queues 工作队列模式

1. 模式说明
image.png

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

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

    2. 代码编写

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

    1) 生产者代码编写 ```java /**

    • 发送消息 */ public class Producer_WorkQueues { public static void main(String[] args) throws IOException, TimeoutException {

      //1.创建连接工厂 ConnectionFactory factory = new ConnectionFactory(); //2. 设置参数 factory.setHost(“172.16.98.133”);//ip 默认值 localhost factory.setPort(5672); //端口 默认值 5672 factory.setVirtualHost(“/itcast”);//虚拟机 默认值/ factory.setUsername(“heima”);//用户名 默认 guest factory.setPassword(“heima”);//密码 默认值 guest //3. 创建连接 Connection Connection connection = factory.newConnection(); //4. 创建Channel Channel channel = connection.createChannel(); //5. 创建队列Queue /* queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments) 参数：

      1.    1. queue：队列名称
      2.    2. durable:是否持久化，当mq重启之后，还在
      3.    3. exclusive：
      4.        * 是否独占。只能有一个消费者监听这队列
      5.        * 当Connection关闭时，是否删除队列
      6.        *
      7.    4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉
      8.    5. arguments：参数。
      10. */

      //如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建 channel.queueDeclare(“work_queues”,true,false,false,null); /* basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) 参数：

      1.    1. exchange：交换机名称。简单模式下交换机会使用默认的 ""
      2.    2. routingKey：路由名称
      3.    3. props：配置信息
      4.    4. body：发送消息数据
      6. */

      for (int i = 1; i <= 10; i++) {

      1.    String body = i+"hello rabbitmq~~~";
      3.    //6. 发送消息
      4.    channel.basicPublish("","work_queues",null,body.getBytes());

      }

  1.     //7.释放资源
  2.   channel.close();
  3.   connection.close();
  5. }

}

  1. **2）消费者代码编写**
  3. Consumer_WorkQueues1.java
  4. ```java
  5. public class Consumer_WorkQueues1 {
  6.     public static void main(String[] args) throws IOException, TimeoutException {
  8.         //1.创建连接工厂
  9.         ConnectionFactory factory = new ConnectionFactory();
  10.         //2. 设置参数
  11.         factory.setHost("172.16.98.133");//ip  默认值 localhost
  12.         factory.setPort(5672); //端口  默认值 5672
  13.         factory.setVirtualHost("/itcast");//虚拟机 默认值/
  14.         factory.setUsername("heima");//用户名 默认 guest
  15.         factory.setPassword("heima");//密码 默认值 guest
  16.         //3. 创建连接 Connection
  17.         Connection connection = factory.newConnection();
  18.         //4. 创建Channel
  19.         Channel channel = connection.createChannel();
  21.         /*
  22.         basicConsume(String queue, boolean autoAck, Consumer callback)
  23.         参数：
  24.             1. queue：队列名称
  25.             2. autoAck：是否自动确认
  26.             3. callback：回调对象
  28.          */
  29.         // 接收消息
  30.         Consumer consumer = new DefaultConsumer(channel){
  31.             /*
  32.                 回调方法，当收到消息后，会自动执行该方法
  34.                 1. consumerTag：标识
  35.                 2. envelope：获取一些信息，交换机，路由key...
  36.                 3. properties:配置信息
  37.                 4. body：数据
  39.              */
  40.             @Override
  41.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  42.               /*  System.out.println("consumerTag："+consumerTag);
  43.                 System.out.println("Exchange："+envelope.getExchange());
  44.                 System.out.println("RoutingKey："+envelope.getRoutingKey());
  45.                 System.out.println("properties："+properties);*/
  46.                 System.out.println("body："+new String(body));
  47.             }
  48.         };
  49.         channel.basicConsume("work_queues",true,consumer);
  52.         //关闭资源？不要
  54.     }
  55. }

Consumer_WorkQueues2.java

  1. public class Consumer_WorkQueues2 {
  2.     public static void main(String[] args) throws IOException, TimeoutException {
  4.         //1.创建连接工厂
  5.         ConnectionFactory factory = new ConnectionFactory();
  6.         //2. 设置参数
  7.         factory.setHost("172.16.98.133");//ip  默认值 localhost
  8.         factory.setPort(5672); //端口  默认值 5672
  9.         factory.setVirtualHost("/itcast");//虚拟机 默认值/
  10.         factory.setUsername("heima");//用户名 默认 guest
  11.         factory.setPassword("heima");//密码 默认值 guest
  12.         //3. 创建连接 Connection
  13.         Connection connection = factory.newConnection();
  14.         //4. 创建Channel
  15.         Channel channel = connection.createChannel();
  17.         /*
  18.         basicConsume(String queue, boolean autoAck, Consumer callback)
  19.         参数：
  20.             1. queue：队列名称
  21.             2. autoAck：是否自动确认
  22.             3. callback：回调对象
  24.          */
  25.         // 接收消息
  26.         Consumer consumer = new DefaultConsumer(channel){
  27.             /*
  28.                 回调方法，当收到消息后，会自动执行该方法
  30.                 1. consumerTag：标识
  31.                 2. envelope：获取一些信息，交换机，路由key...
  32.                 3. properties:配置信息
  33.                 4. body：数据
  35.              */
  36.             @Override
  37.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  38.               /*  System.out.println("consumerTag："+consumerTag);
  39.                 System.out.println("Exchange："+envelope.getExchange());
  40.                 System.out.println("RoutingKey："+envelope.getRoutingKey());
  41.                 System.out.println("properties："+properties);*/
  42.                 System.out.println("body："+new String(body));
  43.             }
  44.         };
  45.         channel.basicConsume("work_queues",true,consumer);
  48.         //关闭资源？不要
  50.     }
  51. }

3. 小结

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

  2. Work Queues 对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。例如：短信服务部署多个， 只需要有一个节点成功发送即可。

5.2 Pub/Sub 订阅模式

1. 模式说明
image.png
在订阅模型中，多了一个 Exchange 角色，而且过程略有变化：

  • P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
  • C：消费者，消息的接收者，会一直等待消息到来
  • Queue：消息队列，接收消息、缓存消息
  • Exchange：交换机（X）。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、 递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。

Exchange有常见以下3种类型：

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

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

1）生产者代码编写

  1. /**
  2.  * 发送消息
  3.  */
  4. public class Producer_PubSub {
  5.     public static void main(String[] args) throws IOException, TimeoutException {
  7.         //1.创建连接工厂
  8.         ConnectionFactory factory = new ConnectionFactory();
  9.         //2. 设置参数
  10.         factory.setHost("172.16.98.133");//ip  默认值 localhost
  11.         factory.setPort(5672); //端口  默认值 5672
  12.         factory.setVirtualHost("/itcast");//虚拟机 默认值/
  13.         factory.setUsername("heima");//用户名 默认 guest
  14.         factory.setPassword("heima");//密码 默认值 guest
  15.         //3. 创建连接 Connection
  16.         Connection connection = factory.newConnection();
  17.         //4. 创建Channel
  18.         Channel channel = connection.createChannel();
  19.        /*
  21.        exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
  22.        参数：
  23.         1. exchange:交换机名称
  24.         2. type:交换机类型
  25.             DIRECT("direct"),：定向
  26.             FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
  27.             TOPIC("topic"),通配符的方式
  28.             HEADERS("headers");参数匹配
  30.         3. durable:是否持久化
  31.         4. autoDelete:自动删除
  32.         5. internal：内部使用。 一般false
  33.         6. arguments：参数
  34.         */
  36.        String exchangeName = "test_fanout";
  37.         //5. 创建交换机
  38.         channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null);
  39.         //6. 创建队列
  40.         String queue1Name = "test_fanout_queue1";
  41.         String queue2Name = "test_fanout_queue2";
  42.         channel.queueDeclare(queue1Name,true,false,false,null);
  43.         channel.queueDeclare(queue2Name,true,false,false,null);
  44.         //7. 绑定队列和交换机
  45.         /*
  46.         queueBind(String queue, String exchange, String routingKey)
  47.         参数：
  48.             1. queue：队列名称
  49.             2. exchange：交换机名称
  50.             3. routingKey：路由键，绑定规则
  51.                 如果交换机的类型为fanout ，routingKey设置为""
  52.          */
  53.         channel.queueBind(queue1Name,exchangeName,"");
  54.         channel.queueBind(queue2Name,exchangeName,"");
  56.         String body = "日志信息：张三调用了findAll方法...日志级别：info...";
  57.         //8. 发送消息
  58.         channel.basicPublish(exchangeName,"",null,body.getBytes());
  60.         //9. 释放资源
  61.         channel.close();
  62.         connection.close();
  64.     }
  65. }

2）消费者代码编写

Consumer_PubSub1.java

  1. public class Consumer_PubSub1 {
  2.     public static void main(String[] args) throws IOException, TimeoutException {
  4.         //1.创建连接工厂
  5.         ConnectionFactory factory = new ConnectionFactory();
  6.         //2. 设置参数
  7.         factory.setHost("172.16.98.133");//ip  默认值 localhost
  8.         factory.setPort(5672); //端口  默认值 5672
  9.         factory.setVirtualHost("/itcast");//虚拟机 默认值/
  10.         factory.setUsername("heima");//用户名 默认 guest
  11.         factory.setPassword("heima");//密码 默认值 guest
  12.         //3. 创建连接 Connection
  13.         Connection connection = factory.newConnection();
  14.         //4. 创建Channel
  15.         Channel channel = connection.createChannel();
  18.         String queue1Name = "test_fanout_queue1";
  19.         String queue2Name = "test_fanout_queue2";
  22.         /*
  23.         basicConsume(String queue, boolean autoAck, Consumer callback)
  24.         参数：
  25.             1. queue：队列名称
  26.             2. autoAck：是否自动确认
  27.             3. callback：回调对象
  29.          */
  30.         // 接收消息
  31.         Consumer consumer = new DefaultConsumer(channel){
  32.             /*
  33.                 回调方法，当收到消息后，会自动执行该方法
  35.                 1. consumerTag：标识
  36.                 2. envelope：获取一些信息，交换机，路由key...
  37.                 3. properties:配置信息
  38.                 4. body：数据
  40.              */
  41.             @Override
  42.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  43.               /*  System.out.println("consumerTag："+consumerTag);
  44.                 System.out.println("Exchange："+envelope.getExchange());
  45.                 System.out.println("RoutingKey："+envelope.getRoutingKey());
  46.                 System.out.println("properties："+properties);*/
  47.                 System.out.println("body："+new String(body));
  48.                 System.out.println("将日志信息打印到控制台.....");
  49.             }
  50.         };
  51.         channel.basicConsume(queue1Name,true,consumer);
  54.         //关闭资源？不要
  56.     }
  57. }

Consumer_PubSub2.java

  1. public class Consumer_PubSub2 {
  2.     public static void main(String[] args) throws IOException, TimeoutException {
  4.         //1.创建连接工厂
  5.         ConnectionFactory factory = new ConnectionFactory();
  6.         //2. 设置参数
  7.         factory.setHost("172.16.98.133");//ip  默认值 localhost
  8.         factory.setPort(5672); //端口  默认值 5672
  9.         factory.setVirtualHost("/itcast");//虚拟机 默认值/
  10.         factory.setUsername("heima");//用户名 默认 guest
  11.         factory.setPassword("heima");//密码 默认值 guest
  12.         //3. 创建连接 Connection
  13.         Connection connection = factory.newConnection();
  14.         //4. 创建Channel
  15.         Channel channel = connection.createChannel();
  18.         String queue1Name = "test_fanout_queue1";
  19.         String queue2Name = "test_fanout_queue2";
  22.         /*
  23.         basicConsume(String queue, boolean autoAck, Consumer callback)
  24.         参数：
  25.             1. queue：队列名称
  26.             2. autoAck：是否自动确认
  27.             3. callback：回调对象
  29.          */
  30.         // 接收消息
  31.         Consumer consumer = new DefaultConsumer(channel){
  32.             /*
  33.                 回调方法，当收到消息后，会自动执行该方法
  35.                 1. consumerTag：标识
  36.                 2. envelope：获取一些信息，交换机，路由key...
  37.                 3. properties:配置信息
  38.                 4. body：数据
  40.              */
  41.             @Override
  42.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  43.               /*  System.out.println("consumerTag："+consumerTag);
  44.                 System.out.println("Exchange："+envelope.getExchange());
  45.                 System.out.println("RoutingKey："+envelope.getRoutingKey());
  46.                 System.out.println("properties："+properties);*/
  47.                 System.out.println("body："+new String(body));
  48.                 System.out.println("将日志信息保存数据库.....");
  49.             }
  50.         };
  51.         channel.basicConsume(queue2Name,true,consumer);
  54.         //关闭资源？不要
  56.     }
  57. }

2. 小结

  1. 交换机需要与队列进行绑定，绑定之后；一个消息可以被多个消费者都收到。
  1. 发布订阅模式与工作队列模式的区别：
  • 工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机 ;
  • 发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息(底层使用 默认交换机) ;
  • 发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑 定到默认的交换机 ;

5.3 Routing 路由模式

1.模式说明

  • 队列与交换机的绑定，不能是任意绑定了，而是要指定一个 RoutingKey（路由key）

  • 消息的发送方在向 Exchange 发送消息时，也必须指定消息的 RoutingKey

  • Exchange 不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key 进行判断，只有队列的 Routingkey 与消息的 Routing key 完全一致，才会接收到消息


    image.png
    图解：

  • P：生产者，向 Exchange 发送消息，发送消息时，会指定一个routing key

  • X：Exchange（交换机），接收生产者的消息，然后把消息递交给与 routing key 完全匹配的队列
  • C1：消费者，其所在队列指定了需要 routing key 为 error 的消息

  • C2：消费者，其所在队列指定了需要 routing key 为 info、error、warning 的消息

    1）生产者代码编写

    1. /**
    2. * 发送消息
    3. */
    4. public class Producer_Routing {
    5.   public static void main(String[] args) throws IOException, TimeoutException {
    7.       //1.创建连接工厂
    8.       ConnectionFactory factory = new ConnectionFactory();
    9.       //2. 设置参数
    10.       factory.setHost("172.16.98.133");//ip  默认值 localhost
    11.       factory.setPort(5672); //端口  默认值 5672
    12.       factory.setVirtualHost("/itcast");//虚拟机 默认值/
    13.       factory.setUsername("heima");//用户名 默认 guest
    14.       factory.setPassword("heima");//密码 默认值 guest
    15.       //3. 创建连接 Connection
    16.       Connection connection = factory.newConnection();
    17.       //4. 创建Channel
    18.       Channel channel = connection.createChannel();
    19.      /*
    21.      exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
    22.      参数：
    23.       1. exchange:交换机名称
    24.       2. type:交换机类型
    25.           DIRECT("direct"),：定向
    26.           FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
    27.           TOPIC("topic"),通配符的方式
    28.           HEADERS("headers");参数匹配
    30.       3. durable:是否持久化
    31.       4. autoDelete:自动删除
    32.       5. internal：内部使用。 一般false
    33.       6. arguments：参数
    34.       */
    36.      String exchangeName = "test_direct";
    37.       //5. 创建交换机
    38.       channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,false,null);
    39.       //6. 创建队列
    40.       String queue1Name = "test_direct_queue1";
    41.       String queue2Name = "test_direct_queue2";
    43.       channel.queueDeclare(queue1Name,true,false,false,null);
    44.       channel.queueDeclare(queue2Name,true,false,false,null);
    45.       //7. 绑定队列和交换机
    46.       /*
    47.       queueBind(String queue, String exchange, String routingKey)
    48.       参数：
    49.           1. queue：队列名称
    50.           2. exchange：交换机名称
    51.           3. routingKey：路由键，绑定规则
    52.               如果交换机的类型为fanout ，routingKey设置为""
    53.        */
    54.       //队列1绑定 error
    55.       channel.queueBind(queue1Name,exchangeName,"error");
    56.       //队列2绑定 info  error  warning
    57.       channel.queueBind(queue2Name,exchangeName,"info");
    58.       channel.queueBind(queue2Name,exchangeName,"error");
    59.       channel.queueBind(queue2Name,exchangeName,"warning");
    61.       String body = "日志信息：张三调用了delete方法...出错误了。。。日志级别：error...";
    62.       //8. 发送消息
    63.       channel.basicPublish(exchangeName,"warning",null,body.getBytes());
    65.       //9. 释放资源
    66.       channel.close();
    67.       connection.close();
    69.   }
    70. }

    2）消费者代码编写

Consumer_Routing1.java

public class Consumer_Routing1 {
    public static void main(String[] args) throws IOException, TimeoutException {

        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2. 设置参数
        factory.setHost("172.16.98.133");//ip  默认值 localhost
        factory.setPort(5672); //端口  默认值 5672
        factory.setVirtualHost("/itcast");//虚拟机 默认值/
        factory.setUsername("heima");//用户名 默认 guest
        factory.setPassword("heima");//密码 默认值 guest
        //3. 创建连接 Connection
        Connection connection = factory.newConnection();
        //4. 创建Channel
        Channel channel = connection.createChannel();


        String queue1Name = "test_direct_queue1";
        String queue2Name = "test_direct_queue2";


        /*
        basicConsume(String queue, boolean autoAck, Consumer callback)
        参数：
            1. queue：队列名称
            2. autoAck：是否自动确认
            3. callback：回调对象

         */
        // 接收消息
        Consumer consumer = new DefaultConsumer(channel){
            /*
                回调方法，当收到消息后，会自动执行该方法

                1. consumerTag：标识
                2. envelope：获取一些信息，交换机，路由key...
                3. properties:配置信息
                4. body：数据

             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              /*  System.out.println("consumerTag："+consumerTag);
                System.out.println("Exchange："+envelope.getExchange());
                System.out.println("RoutingKey："+envelope.getRoutingKey());
                System.out.println("properties："+properties);*/
                System.out.println("body："+new String(body));
                System.out.println("将日志信息打印到控制台.....");
            }
        };
        channel.basicConsume(queue2Name,true,consumer);


        //关闭资源？不要

    }
}

Consumer_Routing2.java

public class Consumer_Routing2 {
    public static void main(String[] args) throws IOException, TimeoutException {

        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2. 设置参数
        factory.setHost("172.16.98.133");//ip  默认值 localhost
        factory.setPort(5672); //端口  默认值 5672
        factory.setVirtualHost("/itcast");//虚拟机 默认值/
        factory.setUsername("heima");//用户名 默认 guest
        factory.setPassword("heima");//密码 默认值 guest
        //3. 创建连接 Connection
        Connection connection = factory.newConnection();
        //4. 创建Channel
        Channel channel = connection.createChannel();


        String queue1Name = "test_direct_queue1";
        String queue2Name = "test_direct_queue2";


        /*
        basicConsume(String queue, boolean autoAck, Consumer callback)
        参数：
            1. queue：队列名称
            2. autoAck：是否自动确认
            3. callback：回调对象

         */
        // 接收消息
        Consumer consumer = new DefaultConsumer(channel){
            /*
                回调方法，当收到消息后，会自动执行该方法

                1. consumerTag：标识
                2. envelope：获取一些信息，交换机，路由key...
                3. properties:配置信息
                4. body：数据

             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              /*  System.out.println("consumerTag："+consumerTag);
                System.out.println("Exchange："+envelope.getExchange());
                System.out.println("RoutingKey："+envelope.getRoutingKey());
                System.out.println("properties："+properties);*/
                System.out.println("body："+new String(body));
                System.out.println("将日志信息存储到数据库.....");
            }
        };
        channel.basicConsume(queue1Name,true,consumer);


        //关闭资源？不要

    }
}

2. 小结

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

5.4 Topics 通配符模式

1. 模式说明

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

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

  • 通配符规则：# 匹配一个或多个词，* 匹配不多不少恰好1个词，例如：item.# 能够匹配 item.insert.abc 或者 item.insert，item.* 只能匹配 item.insert

    image.png

    图解：

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

  • 黄色 Queue：绑定的是 #.news ，因此凡是以 .news 结尾的 routing key 都会被匹配

1）生产者代码编写

/**
 * 发送消息
 */
public class Producer_Topics {
    public static void main(String[] args) throws IOException, TimeoutException {

        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2. 设置参数
        factory.setHost("172.16.98.133");//ip  默认值 localhost
        factory.setPort(5672); //端口  默认值 5672
        factory.setVirtualHost("/itcast");//虚拟机 默认值/
        factory.setUsername("heima");//用户名 默认 guest
        factory.setPassword("heima");//密码 默认值 guest
        //3. 创建连接 Connection
        Connection connection = factory.newConnection();
        //4. 创建Channel
        Channel channel = connection.createChannel();
       /*

       exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
       参数：
        1. exchange:交换机名称
        2. type:交换机类型
            DIRECT("direct"),：定向
            FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
            TOPIC("topic"),通配符的方式
            HEADERS("headers");参数匹配

        3. durable:是否持久化
        4. autoDelete:自动删除
        5. internal：内部使用。 一般false
        6. arguments：参数
        */

       String exchangeName = "test_topic";
        //5. 创建交换机
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,true,false,false,null);
        //6. 创建队列
        String queue1Name = "test_topic_queue1";
        String queue2Name = "test_topic_queue2";
        channel.queueDeclare(queue1Name,true,false,false,null);
        channel.queueDeclare(queue2Name,true,false,false,null);
        //7. 绑定队列和交换机
        /*
        queueBind(String queue, String exchange, String routingKey)
        参数：
            1. queue：队列名称
            2. exchange：交换机名称
            3. routingKey：路由键，绑定规则
                如果交换机的类型为fanout ，routingKey设置为""
         */

        // routing key  系统的名称.日志的级别。
        //=需求： 所有error级别的日志存入数据库，所有order系统的日志存入数据库
        channel.queueBind(queue1Name,exchangeName,"#.error");
        channel.queueBind(queue1Name,exchangeName,"order.*");
        channel.queueBind(queue2Name,exchangeName,"*.*");

        String body = "日志信息：张三调用了findAll方法...日志级别：info...";
        //8. 发送消息
        channel.basicPublish(exchangeName,"goods.error",null,body.getBytes());

        //9. 释放资源
        channel.close();
        connection.close();

    }
}

2）消费者代码编写

Consumer_Topic1.java

public class Consumer_Topic1 {
    public static void main(String[] args) throws IOException, TimeoutException {

        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2. 设置参数
        factory.setHost("172.16.98.133");//ip  默认值 localhost
        factory.setPort(5672); //端口  默认值 5672
        factory.setVirtualHost("/itcast");//虚拟机 默认值/
        factory.setUsername("heima");//用户名 默认 guest
        factory.setPassword("heima");//密码 默认值 guest
        //3. 创建连接 Connection
        Connection connection = factory.newConnection();
        //4. 创建Channel
        Channel channel = connection.createChannel();


        String queue1Name = "test_topic_queue1";
        String queue2Name = "test_topic_queue2";


        /*
        basicConsume(String queue, boolean autoAck, Consumer callback)
        参数：
            1. queue：队列名称
            2. autoAck：是否自动确认
            3. callback：回调对象

         */
        // 接收消息
        Consumer consumer = new DefaultConsumer(channel){
            /*
                回调方法，当收到消息后，会自动执行该方法

                1. consumerTag：标识
                2. envelope：获取一些信息，交换机，路由key...
                3. properties:配置信息
                4. body：数据

             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              /*  System.out.println("consumerTag："+consumerTag);
                System.out.println("Exchange："+envelope.getExchange());
                System.out.println("RoutingKey："+envelope.getRoutingKey());
                System.out.println("properties："+properties);*/
                System.out.println("body："+new String(body));
                System.out.println("将日志信息存入数据库.......");
            }
        };
        channel.basicConsume(queue1Name,true,consumer);


        //关闭资源？不要

    }
}

Consumer_Topic2.java

public class Consumer_Topic2 {
    public static void main(String[] args) throws IOException, TimeoutException {

        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2. 设置参数
        factory.setHost("172.16.98.133");//ip  默认值 localhost
        factory.setPort(5672); //端口  默认值 5672
        factory.setVirtualHost("/itcast");//虚拟机 默认值/
        factory.setUsername("heima");//用户名 默认 guest
        factory.setPassword("heima");//密码 默认值 guest
        //3. 创建连接 Connection
        Connection connection = factory.newConnection();
        //4. 创建Channel
        Channel channel = connection.createChannel();


        String queue1Name = "test_topic_queue1";
        String queue2Name = "test_topic_queue2";


        /*
        basicConsume(String queue, boolean autoAck, Consumer callback)
        参数：
            1. queue：队列名称
            2. autoAck：是否自动确认
            3. callback：回调对象

         */
        // 接收消息
        Consumer consumer = new DefaultConsumer(channel){
            /*
                回调方法，当收到消息后，会自动执行该方法

                1. consumerTag：标识
                2. envelope：获取一些信息，交换机，路由key...
                3. properties:配置信息
                4. body：数据

             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              /*  System.out.println("consumerTag："+consumerTag);
                System.out.println("Exchange："+envelope.getExchange());
                System.out.println("RoutingKey："+envelope.getRoutingKey());
                System.out.println("properties："+properties);*/
                System.out.println("body："+new String(body));
                System.out.println("将日志信息打印控制台.......");
            }
        };
        channel.basicConsume(queue2Name,true,consumer);


        //关闭资源？不要

    }
}

2. 小结

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

5.5 工作模式总结

1. 简单模式 HelloWorld

一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）。

2. 工作队列模式

Work Queue 一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）。

3. 发布订阅模式 Publish/subscribe

需要设置类型为 fanout 的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消 息发送到绑定的队列。

4. 路由模式 Routing

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

5. 通配符模式 Topic

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