码迷,mamicode.com
首页 > 其他好文 > 详细

4 交换机-fanout(订阅发布模式)

时间:2019-08-21 21:43:57      阅读:91      评论:0      收藏:0      [点我收藏+]

标签:close   OLE   生产者和消费者   系统   png   默认   规则   vat   问题   

订阅发布模式

1、交换器(Exchange)

Work Queue背后,其实是rabbitMQ把每条任务消息只发给一个消费者。本篇中我们将要研究如何把一条消息推送给多个消费者,这种模式被称为publish/subscribe(发布/订阅)

技术图片

RabbitMQ的消息发送模型核心思想是生产者不直接把消息发送到消息队列中。事实上,生产者不知道自己的消息将会被缓存到哪个队列中。

其实生产者者可以把消息发送到exchange(消息交换机)上。exchange是一个很简单的事情,它一边接收生产者的消息,另一边再把消息推送到消息队列中。Exchange必须知道在它接收到一条消息应该怎么去处理。应该把这条消息推送到指定的消息队列中?还是把消息推送到所有的队列中?或是把消息丢掉?这些规则都可以用exchange类型来定义

1.1、创建交换器

有一些可用的exchange类型:direct, topic, headersfanout。这里我们主要看最后一个:fanout,这里我们创建一个名字为logs、类型为fanoutexchange:

channel.exchangeDeclare("logs", "fanout");

fanout类型的exchange是很简单的。就是它把它能接收到的所有消息广播到它知道的所有队列中。

  • 没有名字的exchange
channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());

如上面的代码我们没有指定exchagne的名字,采用的是“”,空字符串的符号指的是默认的或没有命名的exchange:消息会根据routingKey被路由到指定的消息队列中

// 申明交换器,第一个参数:交换器的名字;第二个参数:交换器的类型
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

1.2 、推送消息到交换器

现在我们来把消息推送到已命名的exchange上,原来的做法是推送到默认的交换器上面的;

  • 原来的做法
// 第一个参数:交换器的名称
// 第二个参数:队列名称
// 第三个参数:消息的属性
// 第四个参数:消息体
 channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
  • 推送到交换器

// 第一个参数:交换器名称;
// 第二个参数:队列名称;
// 第三个参数:消息属性;
// 第四个参数:消息体
channel.basicPublish(EXCHANGE_NAME,"",null,"i am message".getBytes());

2、临时队列

之前的例子中,应该会发现我们都是使用了一个指定名字的消息队列。对应的生产者和消费者之间都要使用相同的消息队列名称

但是在我们的log系统中却不是这样,我们希望能够接收到所有的log消息,不只是其中的一部分。我们只要处理当前的log消息,不用管过去的历史log。为了实现,我们需要做以下两步:

  • 无论什么时候我们和RabbitMQ建立连接时,我们都要刷新、清空Queue。为了达到这一的目的,我们可以用一个随机的名字(随机性可由自己来定义)来创建Queue,也可以让服务器来自动建立一个随见的Queue
  • 当消费者断开连接时,Queue能自动被删除。

使用java客户端时,我们使用无参数的queueDeclare方法,就可以创建一个已经生成名字的排他性会自动删除Queue

String queueName = channel.queueDeclare().getQueue();

这里面我们就可以拿到一个随机名字的queue,如:amq.gen-JzTY20BRgKO-HjmUJj0wLg

3、绑定(bingdings)

技术图片

现在已经创建好了一个fanout类型的exchange和一个队列。那么接下来我们就需要让exchange向我们的queue里发送消息,Exchangequeue之间的关系就是绑定(bindings

    channel.queueBind(queueName,exchangeName,"");

5、代码例子

现在的代码和之前的区别不是很大;

主要的区别就是:

  • 我们把消息推送到一个命名的exchange上,而不是之前未命名的默认exchange
  • 在我们发送消息时需要提供一个routingKey,但对于fanout类型的exchange可以忽略

5.1、生产者代码示例


/**
 * @author zhaodi
 * @description
 * @date 2018/9/28 16:50
 */
public class Producer {
    private static final String EXCHANGE_NAME = "my-exchange-1";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = MqConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        // 申明交换器,
        // 第一个参数:交换器的名字;
        // 第二个参数:交换器的类型
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

        // 第一个参数:交换器名称;
        // 第二个参数:队列名称;
        // 第三个参数:消息属性;
        // 第四个参数:消息体
        channel.basicPublish(EXCHANGE_NAME,"",null,"i am message".getBytes());
        channel.close();
        connection.close();
}

正如你所见,在建立连接后我们声明了exchange。这一步是必须的,因为禁止向一个不存在的exchange推送消息。

如果没有对exchange负责的queue,那么消息将会被丢失,这是没有问题的;如果没有消费者监听的话,我们会安全的丢掉这些消息。

5.2、消费者代码示例


/**
 * @author zhaodi
 * @desc 发布订阅模式
 */
public class Consumer {

    private static final String EXCHANGE_NAME="my-exchange-1";
    public static void main(String[] args) throws IOException {


        Connection connection = MqConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        // 申明消息路由的名称和类型
        channel.exchangeDeclare(EXCHANGE_NAME,"fanout");

        // 申明一个随机的消息队列名称
        String queueName = channel.queueDeclare().getQueue();

        // 绑定消息路由和消息队列
        channel.queueBind(queueName,EXCHANGE_NAME,"");

        // 创建消费者
        com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel){

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("c1--->:"+new String(body));
                // 手动应答
                // 第一个参数:消息标志
                channel.basicAck(envelope.getDeliveryTag(),false);

            }
        };
        // 监听,关闭自动应答
        boolean autoAck = false;
        channel.basicConsume(queueName,autoAck,consumer);
    }
}

4 交换机-fanout(订阅发布模式)

标签:close   OLE   生产者和消费者   系统   png   默认   规则   vat   问题   

原文地址:https://www.cnblogs.com/zhaod/p/11391258.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!