RabbitMQ的Publish/Subscribe 发布与订阅模式

RabbitMQ的Publish/Subscribe 发布与订阅模式

• Publish/Subscribe 模式图片
  
• 其实简单模式和work queues模式都有交换机,只是我们没定义用的是默认交换机
• 模式说明:
  1. 产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
  2. 消息的接收者，会一直等待消息到来
  3. 消息队列，接收消息、缓存消息
  4. 交换机（X）。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型
  5. Exchange有常见以下3种类型：
     1. Fanout：广播，将消息交给所有绑定到交换机的队列
     2. Direct：定向，把消息交给符合指定routing key 的队列
     3. Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列
  6. Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与 Exchange 绑定，或者没有符合路由规则的队列，那么消息会丢失！

• 代码编写生产者(Fanout模式)步骤
  1. 创建工厂
  2. 设置参数
  3. 创建连接
  4. 创建channel(管道)
  5. 创建交换机
  6. 创建队列(需要广播几个创建几个)
  7. 绑定队列和交换机
  8. 发送消息
  9. 关闭连接

package com.yang;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Producer_PubSub {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1. 创建工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2. 设置参数
        factory.setHost("192.168.20.146");
        factory.setPort(5672);
        factory.setUsername("test");
        factory.setPassword("test");
        factory.setVirtualHost("/test");
        //3. 创建连接
        Connection connection = factory.newConnection();
        //4. 创建管道
        Channel channel = connection.createChannel();
        //5. 创建交换机
        /*
            Exchange.DeclareOk exchangeDeclare(String exchange,
                                              String type,
                                              boolean durable,
                                              boolean autoDelete,
                                              boolean internal,
                                              Map<String, Object> arguments) throws IOException;
            参数:
                1. exchange: 交换机的名称
                2. type: 交换机类型
                    DIRECT("direct"),：定向
                    FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
                    TOPIC("topic"),通配符的方式
                    HEADERS("headers");参数匹配
                3. durable: 是否持久化
                4. autoDelete:　是否自动删除
                5. internal: 内部使用, 一般为false
                6. arguments:　参数
         */
        String exchangeName = "test_fanout";
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,
                false,false,null);
        // 6. 创建队列
        String queue1Name = "test_fanout_queue1";
        String queue2Name = "test_fanout_queue2";
        channel.queueDeclare(queue1Name,true,false,false,null);
        channel.queueDeclare(queue2Name,true,false,false,null);

        //7. 绑定队列和交换机
        channel.queueBind(queue1Name,exchangeName,"");
        channel.queueBind(queue2Name,exchangeName,"");
        //8. 发送消息
        String body = "日志信息：张三调用了findAll方法...日志级别：info...";
        channel.basicPublish(exchangeName,"",null,body.getBytes());
        //9. 关闭连接
        channel.close();
        connection.close();
    }
}

• 尝试执行
  

• 代码编写消费者(Fanout模式)5步骤
  1. 创建工厂
  2. 设置参数
  3. 创建连接
  4. 创建channel(管道)
  5. 创建交换机
  6. 创建队列
  7. 绑定队列和交换机
  8. 发送消息
  9. 关闭连接

package com.yang;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer_PubSub1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1. 创建工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2. 设置参数
        factory.setHost("192.168.20.146");
        factory.setPort(5672);
        factory.setUsername("test");
        factory.setPassword("test");
        factory.setVirtualHost("/test");
        //3. 创建连接
        Connection connection = factory.newConnection();
        //4. 创建管道
        Channel channel = connection.createChannel();
        //5. 创建队列?
        //6. 接受消息
        String queue1Name = "test_fanout_queue1";
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                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);
        //7. 关闭连接?
    }
}

注: 第二个Consumer_PubSub2队列名称不一样而已,其他一模一样

• 分别运行Consumer_PubSub1和Consumer_PubSub2
  
  

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

以上就是Publish/Subscribe 发布与订阅模式的全部内容

最后

以上就是安详蜡烛为你收集整理的RabbitMQ的Publish/Subscribe 发布与订阅模式的全部内容，希望文章能够帮你解决RabbitMQ的Publish/Subscribe 发布与订阅模式所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。