[大数据]RabbitMQ学习笔记七：Routing订阅模型

Routing订阅模型-Direct(直连)：

队列与交换机的绑定，不能是任意绑定了，而是需要指定一个Routingkey(路由key)

消息的发送方在想Exchage发送消息时，也必须指定消息的Routingkey

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

上图模型中：

P：生产者，向Exchage发送消息，发送消息时，会指定一个Routingkey

X：Exchage交换机，接受生产者的消息，然后把消息递交给与Routingkey完全匹配的队列

C1：消费者，其所在队列指定了需要Routingkey为error的消息

C2：消费者，其所在队列指定了需要Routingkey为info,error,warning的消息

?

代码实现：

1.开发生产者

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import utils.RabbitMQUtils;

/**
 * 生产者
 */
public class Provider {
    public static void main(String[] args) throws Exception {
        //获取连接对象
        Connection connection = RabbitMQUtils.getConnection();
        //获取通道对象
        Channel channel = connection.createChannel();
        //通过通道声明交换机
        channel.exchangeDeclare("logs_direct","direct");
        //发送消息
        String routingkey = "info";
        channel.basicPublish("logs_direct",routingkey,null,("这是direct模型基于route key :["+routingkey+"]发送的消息 ").getBytes());
        //关闭资源
        RabbitMQUtils.closeConnectionAndChanel(channel,connection);
    }
}

2.开发消费者1

import com.rabbitmq.client.*;
import utils.RabbitMQUtils;

import java.io.IOException;

/**
 * 消费者1
 */
public class Customer1 {
    public static void main(String[] args) throws Exception {
        //获取连接对象
        Connection connection = RabbitMQUtils.getConnection();
        //获取通道对象
        Channel channel = connection.createChannel();
        //通道声明交换机以及交换机的类型
        channel.exchangeDeclare("logs_direct","direct");
        //创建一个临时队列
        String queue = channel.queueDeclare().getQueue();
        //基于route key 绑定队列和交换机
        channel.queueBind(queue,"logs_direct","error");
        //获取消费的消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1："+new String(body));
            }
        });
    }
}

3.开发消费者2

import com.rabbitmq.client.*;
import utils.RabbitMQUtils;

import java.io.IOException;

/**
 * 消费者2
 */
public class Customer2 {
    public static void main(String[] args) throws Exception {
        //获取连接对象
        Connection connection = RabbitMQUtils.getConnection();
        //获取通道对象
        Channel channel = connection.createChannel();
        //通道声明交换机以及交换机的类型
        channel.exchangeDeclare("logs_direct","direct");
        //创建一个临时队列
        String queue = channel.queueDeclare().getQueue();
        //基于route key 绑定队列和交换机
        channel.queueBind(queue,"logs_direct","info");
        channel.queueBind(queue,"logs_direct","error");
        channel.queueBind(queue,"logs_direct","warning");
        //获取消费的消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2："+new String(body));
            }
        });
    }
}

4.运行消费者和生产者

?

发现消费者2中消费了消息，因为消费者2中绑定了三种Routingkey，而消费者1无对应的Routingkey，若将生产者中的Routingkey改为非error,info,warning，则消费者1和消费者2都接收不到消息。

?

更多：

RabbitMQ学习笔记一：了解及在Linux下安装RabbitMQ

RabbitMQ学习笔记二：管理界面初识和管理命令行

RabbitMQ学习笔记三：HelloWorld模型（直连模型）

RabbitMQ学习笔记四：连接工具类的封装

RabbitMQ学习笔记五：Work模型

RabbitMQ学习笔记六：Fanout模型

?