RabbitMQ系列（一）、RabbitMQ入门

RabbitMQ入门

学习目标
1. 消息队列概述
- 1.1. 消息队列MQ
- 1.2、AMQP 和 JMS
- 1.3. 消息队列产品
- 1.4. RabbitMQ
2、安装及配置RabbitMQ
- 2.1. 安装说明
- 2.2. 用户以及Virtual Hosts配置
- - 2.2.1、用户角色
  - 2.2.2、Virtual Hosts配置
3、RabbitMQ案例入门
- 3.1、搭建rabbit模块工程及引入依赖
- 3.2、编写生产者
- 3.3、编写消费者
- 3.4、测试
- 3.5、小结
4、RabbitMQ工作模式
- 4.1、Work queues工作队列模式
- - 4.1.1、测试work queue工作模式
  - 4.1.2、测试
  - 4.1.3、小结
- 4.2、订阅模式类型
- 4.3、Publish/Subscribe发布与订阅模式
- - 4.3.1、模式说明
  - 4.3.2、代码
  - 4.3.3、测试
  - 4.3.4、小结
  - 4.3.5、发布订阅模式与工作队列模式的区别
- 4.4、Routing路由模式
- - 4.4.1、模式说明
  - 4.4.2、代码
  - 4.4.3、测试
  - 4.4.4、小结
- 4.5、Topics通配符模式
- - 4.5.1、模式说明
  - 4.5.2、代码
  - 4.5.3、测试
  - 4.5.4、小结
- 4.6、模式总结

学习目标

能够说出什么是消息队列
能够安装RabbitMQ
能够编写RabbitMQ的入门程序
能够说出RabbitMQ的5种模式特征
能够使用SpringBoot整合RabbitMQ

1. 消息队列概述 1.1. 消息队列MQ

MQ全称是Message Queue，消息队列是应用程序和应用程序之间的通信方法。

为什么使用MQ？
- 在项目中，可将一些无需即时返回且耗时的操作提取出来，进行异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量。
开发中消息队列通常有如下应用场景：
- 任务异步处理：将不需要同步处理的并且耗时长的操作由消息队列通知消息接收方进行异步处理。提高了应用程序的响应时间。
- 应用程序解耦合：MQ相当于一个中介，生产方通过MQ与消费方交互，它将应用程序进行解耦合。

1.2、AMQP 和 JMS

MQ是消息通信的模型；实现MQ的大致有两种主流方式：AMQP、JMS。

AMQP：

AMQP是一种协议，更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。

JMS：

MS即Java消息服务（JavaMessage Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。

AMQP 与 JMS 区别：

JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式
JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
JMS规定了两种消息模式（点对点、订阅）；而AMQP的消息模式更加丰富（最典型是路由）。

1.3. 消息队列产品

市场上常见的消息队列有如下：

ActiveMQ：基于JMS
ZeroMQ：基于C语言开发
RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好
RocketMQ：基于JMS，阿里巴巴产品
Kafka：类似MQ的产品；分布式消息系统，高吞吐量

1.4. RabbitMQ

RabbitMQ是由erlang语言开发，基于AMQP（Advanced Message Queue 高级消息队列协议）协议实现的消息队列，它是一种应用程序之间的通信方法，消息队列在分布式系统开发中应用非常广泛。

RabbitMQ官方地址：http://www.rabbitmq.com/

RabbitMQ提供了6种模式：简单模式，work模式，Publish/Subscribe发布与订阅模式，Routing路由模式，Topics主题模式，RPC远程调用模式（远程调用，不太算MQ；不作介绍）；

官网对应模式介绍：https://www.rabbitmq.com/getstarted.html

2、安装及配置RabbitMQ 2.1. 安装说明

windows下安装Erlang和RabbitMQ详细教程

2.2. 用户以及Virtual Hosts配置 2.2.1、用户角色

RabbitMQ在安装好后，可以访问http://localhost:15672 ；其自带了guest/guest的用户名和密码；如果需要创建自定义用户，那么也可以登录管理界面后，如下操作：

角色说明：

1、超级管理员(administrator)

可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。

2、监控者(monitoring)

可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)

3、策略制定者(policymaker)

可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。

4、普通管理者(management)

仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。

5、其他

无法登陆管理控制台，通常就是普通的生产者和消费者。

2.2.2、Virtual Hosts配置

像mysql拥有数据库的概念并且可以指定用户对库和表等操作的权限。RabbitMQ也有类似的权限管理；在RabbitMQ中可以虚拟消息服务器Virtual Host，每个Virtual Hosts相当于一个相对独立的RabbitMQ服务器，每个VirtualHost之间是相互隔离的。exchange、queue、message不能互通。相当于mysql的db。

注意：Virtual Name一般以/开头。

1. 创建Virtual Hosts

2. 设置Virtual Hosts权限

3、RabbitMQ案例入门 3.1、搭建rabbit模块工程及引入依赖


     org.springframework.cloud
     spring-cloud-stream-binder-rabbit

3.2、编写生产者

抽取创建connection的工具类com.ebuy.rabbitmq.util.ConnectionUtil工具类；

public class ConnectionUtil {

    public static Connection getConnection() throws Exception
    {
        //创建连接工厂
        ConnectionFactory connectionFactory=new ConnectionFactory();
        //主机地址：默认为loalhost
        connectionFactory.setHost("127.0.0.1");
        //连接端口：默认为5672
        connectionFactory.setPort(5672);
        //虚拟主机名称：默认为 /
        connectionFactory.setVirtualHost("/crm");
        //连接用户名：新建的用户（默认为guest）
        connectionFactory.setUsername("wbs");
        //连接民密码：用户密码（默认为guest）
        connectionFactory.setPassword("123456");
        //新建连接
        Connection connection=connectionFactory.newConnection();
        return  connection;
    }
}

编写生产者：

public class Producer {

    //队列名称
    static final String QUEUE_NAME = "simple_queue";

    public static void main(String[] args) {
        try {
            //创建一个连接
            Connection connection = ConnectionUtil.getConnection();
            //创建一个频道
            Channel channel = connection.createChannel();

            
            channel.queueDeclare(QUEUE_NAME,true,false,false,null);
            //生产一个消息
            String message="你好，我是001";

            
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());

            //消息发送后，提示语句
            System.out.println("消息已发送："+message);


            //关闭连接
            channel.close();
            connection.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

在执行上述的消息发送之后；可以登录rabbitMQ的管理控制台，可以发现队列和其消息：

3.3、编写消费者

编写消息的消费者com.ithouse.rabbitmq.simple.Consumer消费者类：

public class Consumer {

    static  final String QUEUE_NAME="simple_queue";

    public static void main(String[] args) {

        try {
            //新建连接
            Connection connection= ConnectionUtil.getConnection();

            //创建频道
            Channel channel=connection.createChannel();

            
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);

            //创建消费者；并设置消息处理
            DefaultConsumer consumer = new DefaultConsumer(channel){

                
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //交换机
                    System.out.println("交换机为：" + envelope.getExchange());
                    //路由key
                    System.out.println("路由key为：" + envelope.getRoutingKey());
                    //消息id
                    System.out.println("消息id为：" + envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("接收到的消息为：" + new String(body, "utf-8"));
                }
            };

            
            channel.basicConsume(QUEUE_NAME, true, consumer);

        }catch (Exception e){
            e.printStackTrace();
        }

    }
}

3.4、测试

先启动生产者：

再启动消费者：

3.5、小结

上述的入门案例中中其实使用的是如下的简单模式：

在上图的模型中，有以下概念：

P：生产者，也就是要发送消息的程序
C：消费者：消息的接受者，会一直等待消息到来。
queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

在rabbitMQ中消息者是一定要到某个消息队列中去获取消息的。

4、RabbitMQ工作模式 4.1、Work queues工作队列模式

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

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

4.1.1、测试work queue工作模式

Work Queues 与入门程序的简单模式的代码是几乎一样的；可以完全复制，并复制多一个消费者进行多个消费者同时消费消息的测试，只需在生产者中添加一个for循环模拟创建多个消息加入到消息队列中：

（1）、生产者

public class Producer {

    static  final String QUEUE_NAME="work_queue";
    public static void main(String[] args) {
        try {
            //创建一个连接
            Connection connection= ConnectionUtil.getConnection();
            //创建一个频道
            Channel channel=connection.createChannel();
            
            channel.queueDeclare(QUEUE_NAME,true,false,false,null);


            for(int i=1;i<=30;i++) {
                String message = "你好，我是work："+i;
                
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());

                System.out.println("消息已发送：" + message);
            }

            channel.close();
            connection.close();


        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

（2）、消费者1

public class Consumer1 {

    static  final String QUEUE_NAME="work_queue";

    public static void main(String[] args) {

        try {
            //新建连接
            Connection connection= ConnectionUtil.getConnection();

            //创建频道
            Channel channel=connection.createChannel();

            
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);

            //创建消费者；并设置消息处理
            DefaultConsumer consumer = new DefaultConsumer(channel){

                
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //交换机
                    System.out.println("交换机为：" + envelope.getExchange());
                    //路由key
                    System.out.println("路由key为：" + envelope.getRoutingKey());
                    //消息id
                    System.out.println("消息id为：" + envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("Consumer1接收到的消息为：" + new String(body, "utf-8"));

                    
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };

            
            channel.basicConsume(QUEUE_NAME, true, consumer);

        }catch (Exception e){
            e.printStackTrace();
        }

    }
}

（3）、消费者2

public class Consumer2 {

    static  final String QUEUE_NAME="work_queue";

    public static void main(String[] args) {

        try {
            //新建连接
            Connection connection= ConnectionUtil.getConnection();

            //创建频道
            Channel channel=connection.createChannel();

            
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);

            //创建消费者；并设置消息处理
            DefaultConsumer consumer = new DefaultConsumer(channel){

                
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //交换机
                    System.out.println("交换机为：" + envelope.getExchange());
                    //路由key
                    System.out.println("路由key为：" + envelope.getRoutingKey());
                    //消息id
                    System.out.println("消息id为：" + envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("Consumer2接收到的消息为：" + new String(body, "utf-8"));

                    
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };

            
            channel.basicConsume(QUEUE_NAME, true, consumer);

        }catch (Exception e){
            e.printStackTrace();
        }

    }
}

4.1.2、测试

先启动两个消费者，然后再启动生产者发送消息；到两个消费者对应的控制台查看是否属于竞争性的接收到消息。

Consumer1：

Consumer2：

4.1.3、小结

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

4.2、订阅模式类型

订阅模式示例图：

前面2个案例中，只有3个角色：

P：生产者，也就是要发送消息的程序
C：消费者：消息的接受者，会一直等待消息到来。
queue：消息队列，图中红色部分

而在订阅模型中，多了一个Exchange角色，而且过程略有变化：

P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
C：消费者，消息的接受者，会一直等待消息到来。
Queue：消息队列，接收消息、缓存消息。
Exchange：交换机，就是图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型：
- Fanout：广播，将消息交给所有绑定到交换机的队列；
- Direct：定向，把消息交给符合指定routing key的队列；
- Topic：通配符，把消息交给符合routing pattern（路由模式）的队列。

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

4.3、Publish/Subscribe发布与订阅模式 4.3.1、模式说明

发布订阅模式：

每个消费者监听自己的队列。
生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息。

4.3.2、代码

（1）Consumer1

public class Consumer1 {

    //交换机名称
    static final String FANOUT_EXCHAGE = "fanout_exchange";
    //队列名称1
    static final String FANOUT_QUEUE_1 = "fanout_queue_1";
    //队列名称1
    static final String FANOUT_QUEUE_2 = "fanout_queue_2";

    public static void main(String[] args) {

        try {
            //新建连接
            Connection connection= ConnectionUtil.getConnection();

            //创建频道
            Channel channel=connection.createChannel();

            
            channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);

            
            channel.queueDeclare(FANOUT_QUEUE_1,true,false,false,null);

            
            channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"");


            //创建消费者；并设置消息处理
            DefaultConsumer consumer = new DefaultConsumer(channel){

                
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //交换机
                    System.out.println("交换机为：" + envelope.getExchange());
                    //路由key
                    System.out.println("路由key为：" + envelope.getRoutingKey());
                    //消息id
                    System.out.println("消息id为：" + envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("Consumer1接收到的消息为：" + new String(body, "utf-8"));

                    
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };

            
            channel.basicConsume(FANOUT_QUEUE_1, true, consumer);

        }catch (Exception e){
            e.printStackTrace();
        }

    }
}

（2）Consumer2

public class Consumer2 {

    //交换机名称
    static final String FANOUT_EXCHAGE = "fanout_exchange";
    //队列名称1
    static final String FANOUT_QUEUE_1 = "fanout_queue_1";
    //队列名称1
    static final String FANOUT_QUEUE_2 = "fanout_queue_2";

    public static void main(String[] args) {

        try {
            //新建连接
            Connection connection= ConnectionUtil.getConnection();

            //创建频道
            Channel channel=connection.createChannel();

            
            channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);

            
            channel.queueDeclare(FANOUT_QUEUE_2,true,false,false,null);

            
            channel.queueBind(FANOUT_QUEUE_2,FANOUT_EXCHAGE,"");


            //创建消费者；并设置消息处理
            DefaultConsumer consumer = new DefaultConsumer(channel){

                
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //交换机
                    System.out.println("交换机为：" + envelope.getExchange());
                    //路由key
                    System.out.println("路由key为：" + envelope.getRoutingKey());
                    //消息id
                    System.out.println("消息id为：" + envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("Consumer2接收到的消息为：" + new String(body, "utf-8"));

                    
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };

            
            channel.basicConsume(FANOUT_QUEUE_2, true, consumer);

        }catch (Exception e){
            e.printStackTrace();
        }

    }
}

（3）Producer

public class Producer {

    //交换机名称
    static final String FANOUT_EXCHAGE = "fanout_exchange";
    //队列名称1
    static final String FANOUT_QUEUE_1 = "fanout_queue_1";
    //队列名称1
    static final String FANOUT_QUEUE_2 = "fanout_queue_2";

    public static void main(String[] args) {

        try {
            //创建连接
            Connection connection = ConnectionUtil.getConnection();

            //创建频道
            Channel channel = connection.createChannel();

            
            channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);

            
            channel.queueDeclare(FANOUT_QUEUE_1,true,false,false,null);
            channel.queueDeclare(FANOUT_QUEUE_2,true,false,false,null);
			
			//队列绑定交换机(绑不绑都可以，因为在消费者方可以随意组合路由key和队列)
            channel.queueBind(FANOUT_QUEUE_1, FANOUT_EXCHAGE, "");
            channel.queueBind(FANOUT_QUEUE_2, FANOUT_EXCHAGE, "");

            for(int i=0;i<30;i++){
                String message = "你好：我是ps" + i;

                
                channel.basicPublish(FANOUT_EXCHAGE,"",null,message.getBytes());

                System.out.println("消息已经发送：" + message);
            }

            //关闭资源
            channel.close();
            connection.close();

        } catch (Exception e) {
            e.printStackTrace();
        }
        
    }
}

4.3.3、测试

首先启动Producer生产者，查看rabbitmq服务中的交换机：

然后启动两个Consumer消费者，查看两个Consumer消费者的控制台：

4.3.4、小结

交换机需要与队列进行绑定，绑定之后；一个消息可以被多个消费者都收到，多个消费者之间不再是竞争的关系。

4.3.5、发布订阅模式与工作队列模式的区别

工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机；
工作队列模式的生产方是面向队列发送消息的（底层使用默认交换机），而发布/订阅模式的生产方是面向交换机发送消息的；
工作队列模式不需要设置，实际上工作队列模式将队列绑定绑定的到默认的交换机，而发布/订阅模式需要设置队列和交换机的绑定。

4.4、Routing路由模式 4.4.1、模式说明

路由模式特点：

队列与交换机的绑定，不能是任意绑定的，而是要指定一个RoutingKey(路由key)；
消息的发送方是在向Exchange发送消息时，也必须指定消息的RoutingKey；
Exchange不再把消息交给每一个绑定的队列，而是根据消息的RoutingKey进行判断，只有队列RoutingKey与消息的RoutingKey完全一致，才会接收到信息。

图解：

P：生产者，向Exchange发送消息，发送消息时，会指定一个routingkey；
X：Exchange（交换机），接收生产者的消息，然后把消息递交给与routingkey完全匹配的队列；
C1：消费者，其所在队列指定了需要routingkey为error的消息；
C2：消费者，其所在队列指定了需要routingkey为info、error、warning的消息。

4.4.2、代码

Producer生产者：

public class Producer {

    //交换机名称
    static final String DIRECT_EXCHAGE = "direct_exchange";
    //队列名称(插入)
    static final String DIRECT_QUEUE_INSERT = "direct_queue_insert";
    //队列名称(更新)
    static final String DIRECT_QUEUE_UPDATE = "direct_queue_update";

    public static void main(String[] args) {
        try {
            //创建连接
            Connection connection = ConnectionUtil.getConnection();

            //创建频道
            Channel channel = connection.createChannel();

            
            channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);

            
            channel.queueDeclare(DIRECT_QUEUE_INSERT,true,false,false,null);
            channel.queueDeclare(DIRECT_QUEUE_UPDATE,true,false,false,null);

            //队列绑定交换机(绑不绑都可以，但是不建议指定routkey，因为在消费者方可以随意组合路由key和队列)
            channel.queueBind(DIRECT_QUEUE_INSERT,DIRECT_EXCHAGE,"");
            channel.queueBind(DIRECT_QUEUE_UPDATE,DIRECT_EXCHAGE,"");

            //发送
            String message = "新增了商品。路由模式；routing key 为 insert ";
            
            channel.basicPublish(DIRECT_EXCHAGE,"insert",null,message.getBytes());
            System.out.println("消息已发送：" + message);

            //发送
            message = "更新了商品。路由模式；routing key 为  ";
            
            channel.basicPublish(DIRECT_EXCHAGE,"updates",null,message.getBytes());
            System.out.println("消息已发送：" + message);

            // 关闭资源
            channel.close();
            connection.close();
        }catch (Exception e){
            e.printStackTrace();
        }

    }
}

注意：

在生产者方，其实channel频道中只是将交换机和所有的队列都绑定在一起，而路由规则key与队列的绑定应该由消费者一方来决定，可以将routkey与队列随意组合调用；
如果在生产者方直接将routkey和指定的队列绑定在交换机中，过于死板了。

Cosumer1：

public class Consumer1 {

    //交换机名称
    static final String DIRECT_EXCHAGE = "direct_exchange";
    //队列名称(插入)
    static final String DIRECT_QUEUE_INSERT = "direct_queue_insert";
    //队列名称(更新)
    static final String DIRECT_QUEUE_UPDATE = "direct_queue_update";

    public static void main(String[] args) {

        try {
            Connection connection = ConnectionUtil.getConnection();

            //创建频道
            Channel channel = connection.createChannel();

            
            channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);

            
            channel.queueDeclare(DIRECT_QUEUE_INSERT,true,false,false,null);

            //绑定指定routingkey队列到交换机
            channel.queueBind(DIRECT_QUEUE_INSERT,DIRECT_EXCHAGE,"insert");

            //创建消费者；并设置消息处理
            DefaultConsumer consumer = new DefaultConsumer(channel){

                
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //交换机
                    System.out.println("交换机为：" + envelope.getExchange());
                    //路由key
                    System.out.println("路由key为：" + envelope.getRoutingKey());
                    //消息id
                    System.out.println("消息id为：" + envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("Consumer1接收到的消息为：" + new String(body, "utf-8"));

                    
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };

            
            channel.basicConsume(DIRECT_QUEUE_INSERT, true, consumer);


        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

Consumer2：

public class Consumer2 {

    //交换机名称
    static final String DIRECT_EXCHAGE = "direct_exchange";
    //队列名称(插入)
    static final String DIRECT_QUEUE_INSERT = "direct_queue_insert";
    //队列名称(更新)
    static final String DIRECT_QUEUE_UPDATE = "direct_queue_update";

    public static void main(String[] args) {

        try {
            Connection connection = ConnectionUtil.getConnection();

            //创建频道
            Channel channel = connection.createChannel();

            
            channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);

            
            channel.queueDeclare(DIRECT_QUEUE_UPDATE,true,false,false,null);

            //绑定指定routingkey队列到交换机
            channel.queueBind(DIRECT_QUEUE_UPDATE,DIRECT_EXCHAGE,"update");

            //创建消费者；并设置消息处理
            DefaultConsumer consumer = new DefaultConsumer(channel){

                
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //交换机
                    System.out.println("交换机为：" + envelope.getExchange());
                    //路由key
                    System.out.println("路由key为：" + envelope.getRoutingKey());
                    //消息id
                    System.out.println("消息id为：" + envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("Consumer2接收到的消息为：" + new String(body, "utf-8"));

                    
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };

            
            channel.basicConsume(DIRECT_QUEUE_UPDATE, true, consumer);


        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

4.4.3、测试

启动所有消费者，然后使用生产者发送消息；在消费者对应的控制台可以查看到生产者发送对应routing key对应的队列的消息；达到按照需要接收的效果。

在执行完测试代码后，其实到RabbitMQ的管理后台找到 Exchanges 选项卡，点击 direct_exchange的交换机，可以查看到如下的绑定：

查看Queues队列选项卡：

direct_queue_update队列同理！！！

查看两个消费者控制台：

4.4.4、小结

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

4.5、Topics通配符模式 4.5.1、模式说明

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

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert
通配符规则：
#：匹配一个或多个词

item.# ：能够匹配 item.insert.abc 或者 item.insert

*：匹配不多不少恰好1个词

item.* ：只能匹配 item.insert

图解：

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

4.5.2、代码

（1）Proucer

public class Producer {

    //交换机名称
    static final String TOPIC_EXCHAGE = "topic_exchange";
    //队列名称
    static final String TOPIC_QUEUE_1 = "topic_queue_1";
    //队列名称
    static final String TOPIC_QUEUE_2 = "topic_queue_2";

    public static void main(String[] args) {

        try {
            //建立连接
            Connection connection = ConnectionUtil.getConnection();

            //创建频道
            Channel channel = connection.createChannel();

            
            channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);

            // 发送信息
            String message = "新增了商品。Topic模式；routing key 为 item.insert" ;
            
            channel.basicPublish(TOPIC_EXCHAGE, "item.insert", null, message.getBytes());
            System.out.println("已发送消息：" + message);

            //修改了商品
            message = "修改了商品。Topic模式；routing key 为 item.update" ;
            
            channel.basicPublish(TOPIC_EXCHAGE, "item.update", null, message.getBytes());
            System.out.println("已发送消息：" + message);

            message = "删除了商品。通配符模式 ；routing key 为 item.delete";
            
            channel.basicPublish(TOPIC_EXCHAGE, "item.delete", null, message.getBytes());
            System.out.println("已发送消息：" + message);


            //关闭资源
            channel.close();
            connection.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

（2）Consumer1

public class Consumer1 {

    //交换机名称
    static final String TOPIC_EXCHAGE = "topic_exchange";
    //队列名称
    static final String TOPIC_QUEUE_1 = "topic_queue_1";
    //队列名称
    static final String TOPIC_QUEUE_2 = "topic_queue_2";

    public static void main(String[] args) {

        try {
            //创建连接
            Connection connection = ConnectionUtil.getConnection();

            //创建频道
            Channel channel = connection.createChannel();

            
            channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);

            
            channel.queueDeclare(TOPIC_QUEUE_1, true, false, false, null);

            
            channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.insert");
            channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.update");

            //创建消费者；并设置消息处理
            DefaultConsumer consumer = new DefaultConsumer(channel){

                
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //交换机
                    System.out.println("交换机为：" + envelope.getExchange());
                    //路由key
                    System.out.println("路由key为：" + envelope.getRoutingKey());
                    //消息id
                    System.out.println("消息id为：" + envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("Consumer1接收到的消息为：" + new String(body, "utf-8"));

                }
            };

            
            channel.basicConsume(TOPIC_QUEUE_1,true,consumer);

        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

关键语句：

channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.insert");
channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.update");

（3）Consumer2

public class Consumer2 {

    //交换机名称
    static final String TOPIC_EXCHAGE = "topic_exchange";
    //队列名称
    static final String TOPIC_QUEUE_1 = "topic_queue_1";
    //队列名称
    static final String TOPIC_QUEUE_2 = "topic_queue_2";

    public static void main(String[] args) {

        try {
            //创建连接
            Connection connection = ConnectionUtil.getConnection();

            //创建频道
            Channel channel = connection.createChannel();

            
            channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);

            
            channel.queueDeclare(TOPIC_QUEUE_2, true, false, false, null);

            
            channel.queueBind(Producer.TOPIC_QUEUE_2, Producer.TOPIC_EXCHAGE, "item.*");

            //创建消费者；并设置消息处理
            DefaultConsumer consumer = new DefaultConsumer(channel){

                
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //交换机
                    System.out.println("交换机为：" + envelope.getExchange());
                    //路由key
                    System.out.println("路由key为：" + envelope.getRoutingKey());
                    //消息id
                    System.out.println("消息id为：" + envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("Consumer2接收到的消息为：" + new String(body, "utf-8"));

                }
            };

            
            channel.basicConsume(TOPIC_QUEUE_2,true,consumer);

        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

关键语句：

channel.queueBind(Producer.TOPIC_QUEUE_2, Producer.TOPIC_EXCHAGE, "item.*");

4.5.3、测试

启动所有消费者，然后使用生产者发送消息；在消费者对应的控制台可以查看到生产者发送对应routing key对应队列的消息；到达按照需要接收的效果；并且这些routing key可以使用通配符。

在执行完测试代码后，其实到RabbitMQ的管理后台找到 Exchanges 选项卡，点击 topic_exchange 的交换机，可以查看到如下的绑定：

查看两个消费者的控制台：

如果将生产者中的其中一个发送routkey修改成三级路径item.abc.insert：

channel.basicPublish(TOPIC_EXCHAGE, "item.abc.insert", null, message.getBytes());

再次启动生产者，查看两个消费者的控制台：

解决办法，将Consumer2中item.*改为item.#,重新启动，查看控制台：

channel.queueBind(Producer.TOPIC_QUEUE_2, Producer.TOPIC_EXCHAGE, "item.#");

4.5.4、小结

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

4.6、模式总结

RabbitMQ工作模式：

1、简单模式 HelloWorld 一个生产者、一个消费者，不需要设置交换机（使用默认的交换
机）
2、工作队列模式 Work Queue 一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）
3、发布订阅模式 Publish/subscribe 需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列
4、路由模式 Routing 需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列
5、通配符模式 Topic 需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

RabbitMQ系列（一）、RabbitMQ入门

大数据系统相关栏目本月热门文章