原文:RabbitMQ入门教程(五):扇形交换机发布/订阅(Publish/Subscribe)

版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。

简介

本节主要演示交换机的广播类型fanout,广播类型不需要routingKey,交换机会将所有的消息都发送到每个绑定的队列中去。

在发布消息时可以只先指定交换机的名称,交换机的声明的代码可以放到消费者端进行声明,队列的声明也放在消费者端来声明。

发布订阅类似观察者模式设计模式,一般适用于当接收到某条消息时同时做多种类似的任务的处理,如一个发短信,另一个一个发邮件;一个插入数据库,另一个保存在文件等类似操作,扇形交换机将消息传送给不同的队列,不同的队列对同一种消息采取不同的行为。

扇形交换机是最基本的交换机类型,它所能做的事情非常简单———广播消息。扇形交换机会把能接收到的消息全部发送给绑定在自己身上的队列。因为广播不需要“思考”,所以扇形交换机处理消息的速度也是所有的交换机类型里面最快的。

生产者

public class Producer {
    @Test
    public void testBasicPublish() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(AMQP.PROTOCOL.PORT);
        factory.setUsername("mengday");
        factory.setPassword("mengday");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        String EXCHANGE_NAME = "exchange.fanout";
        // 生产者不需要队列声明,也不需要声明交换机,只需要指定交换机的名称即可,队列和交换机的声明可以在消费者中声明
        // 循环发布多条消息, 注意广播模式不需要routingKey, 可以写成"", 也可以随意写个名字,在消费者也随便写一个,生产者和消费者的routingKey的不一样看看可以不
        for (int i = 0; i < 10; i++){
            String message = "Hello RabbitMQ " + i;
            // 广播类型不需要routingKey,但是不能写成null,可以写成空字符串""
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
        }

        // 关闭资源
        channel.close();
        connection.close();
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26

消费者1

public class Consumer1 {
  @Test
  public void testBasicConsumer1() throws Exception{
      ConnectionFactory factory = new ConnectionFactory();
      factory.setHost("127.0.0.1");
      factory.setPort(AMQP.PROTOCOL.PORT);
      factory.setUsername("mengday");
      factory.setPassword("mengday");

      Connection connection = factory.newConnection();
      final Channel channel = connection.createChannel();
      channel.basicQos(1);

      String QUEUE_NAME = "queue.fanout.q1";
      String EXCHANGE_NAME = "exchange.fanout";
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);
      // 声明交换机:指定交换机的名称和类型(广播:fanout)
      channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
      // 在消费者端队列绑定
      channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
      System.out.println("Consumer Wating Receive Message");

      Consumer consumer = new DefaultConsumer(channel){
          @Override
          public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              String message = new String(body, "UTF-8");
              System.out.println(" [C] Received '" + message + "', 处理业务中...");
          }
      };

      channel.basicConsume(QUEUE_NAME, true, consumer);

      Thread.sleep(1000000);
  }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35

消费者2

消费者2和消费者1仅仅是队列名称不一样而已

public class Consumer2 {
    @Test
    public void testBasicConsumer2() throws Exception{
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(AMQP.PROTOCOL.PORT);
        factory.setUsername("mengday");
        factory.setPassword("mengday");

        Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();
        channel.basicQos(1);

        String QUEUE_NAME = "queue.fanout.q2";
        String EXCHANGE_NAME = "exchange.fanout";
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 在消费者端队列绑定
        // 声明交换机:指定交换机的名称和类型(广播:fanout)
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        System.out.println("Consumer Wating Receive Message");

        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println(" [C] Received '" + message + "', 处理业务中...");
            }
        };

        channel.basicConsume(QUEUE_NAME, true, consumer);

        Thread.sleep(1000000);
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35

运行效果

注意:先启动消费者,再启动生产者。可以看到两个消费者都收到同样的消息


我的微信公众号:

12-31 18:38