RabbitMQ的通配符(即topic)模式示例代码

1、RabbitMQ官网通配符(即topic)模式的架构图

2、pom文件需要引入有关MQ的2个依赖包

 
   org.springframework.amqp
     spring-rabbit
     1.4.0.RELEASE
 
 
     com.rabbitmq
     amqp-client
     3.4.1
 

3、MQ获取连接工具类代码如下：

package com.rf.rabiitmq.util;

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

/**
 * @description: mq连接工具类
 * @author: xiaozhi
 * @create: 2020-04-21 09:03
 */
public class ConnectionUtil {
    /** 
    * @Description: 获取mq连接方法
    * @Param: [] 
    * @Author: xz  
    * @return: com.rabbitmq.client.Connection
    * @Date: 2020/4/21 9:19
    */ 
    public static Connection getConnection() throws Exception {
        //定义连接工厂
        ConnectionFactory connectionFactory=new ConnectionFactory();
        //以下信息，在安装mq后，登录mq客户端进行配置的信息
        connectionFactory.setHost("localhost");//设置服务地址
        connectionFactory.setPort(5672);//设置端口
        connectionFactory.setVirtualHost("xzVirtualHosts");//设置虚拟主机名称
        connectionFactory.setUsername("xz");//设置用户名
        connectionFactory.setPassword("xz");//设置密码
        //通过连接工厂获取连接
        Connection connection = connectionFactory.newConnection();
        return connection;
    }
}

4、消息发送者，即服务端代码如下：

package com.rf.rabiitmq.topic;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rf.rabiitmq.util.ConnectionUtil;
/**
 * @description: 通配符模式  消息发布者
 * @author: xiaozhi
 * @create: 2020-04-22 09:16
 */
public class TopicSend {
    private final static String EXCHANGE_NAME="topic_exchange_name";//声明交换机名称
    public static void main(String[] args) throws  Exception{
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME,"topic");
        String message="删除员工id=102的信息";
        channel.basicPublish(EXCHANGE_NAME,"item.delete",null,message.getBytes());
        System.out.println("后台系统发布消息："+message);
        channel.close();
        connection.close();
    }
}

5、消息接收者，即客户端1代码如下：

package com.rf.rabiitmq.topic;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
import com.rf.rabiitmq.util.ConnectionUtil;
/**
 * @description: 客户端1
 * @author: xiaozhi
 * @create: 2020-04-22 09:21
 */
public class TopicRescv1 {
    private final static String QUEUE_NAME="topic_queue_name1";
    private final static String EXCHANGE_NAME="topic_exchange_name";//声明交换机名称
    public static void main(String[] args) throws Exception{
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        // 绑定队列到交换机,并指定路由key,#表示匹配一个或一个以上，*表示只匹配一个
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"item.#");
        // 同一时刻服务器只会发一条消息给消费者
        channel.basicQos(1);
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
        channel.basicConsume(QUEUE_NAME,false,queueingConsumer);
        while(true){
            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println("客户端1接收消息："+message);
            Thread.sleep(10);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
        }
    }

}

6、消息接收者，即客户端2代码如下：

package com.rf.rabiitmq.topic;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
import com.rf.rabiitmq.util.ConnectionUtil;
/**
 * @description: 客户端2
 * @author: xiaozhi
 * @create: 2020-04-22 09:22
 */
public class TopicRescv2 {
    private final static String QUEUE_NAME="topic_queue_name2";
    private final static String EXCHANGE_NAME="topic_exchange_name";//声明交换机名称
    public static void main(String[] args) throws Exception{
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        // 绑定队列到交换机,并指定路由key,#表示匹配一个或一个以上，*表示只匹配一个
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"item.select");
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"item.update");
        // 同一时刻服务器只会发一条消息给消费者
        channel.basicQos(1);
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
        channel.basicConsume(QUEUE_NAME,false,queueingConsumer);
        while(true){
            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println("客户端2接收消息："+message);
            Thread.sleep(10);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
        }
    }
}

7、先启动服务端、客户端1和客户端2代码程序，之后关闭服务端、客户端1和客户端2代码程序

• 原因：为了先创建队列和交换机

8、启动客户端代码程序：

1）、启动消息接收者，即客户端1代码程序。输出如下图： 2）、启动消息接收者，即客户端2代码程序。输出如下图：

9、启动服务端代码程序

1）、启动消息发送者，即服务端的代码程序。控制台输出如下图： 2）查看客户端1即(消费者1)和客户端2(即消费者2)的控制台输出如下：

10、结论

• 客户端1即(消费者1)接收到消息
• 客户端2即(消费者2)未接收到消息

11、结论的原因

1）、客户端1即(消费者1)绑定队列到交换机中路由key指定了以item开头，以#结尾的所有可能方式，如下图：

注：

• #表示匹配一个或一个以上
• *表示只匹配一个 2）、客户端2即(消费者2)绑定队列到交换机中路由key指定了以item开头，以select、update结尾的2种方式,如下图： 3）、服务端程序代码发送消息时绑定路由key是以item开头，以delete结尾，所以匹配到了客户端1中绑定的路由key指定了以item开头，以#结尾的所有可能方式；而客户端2只有select和update,所以客户端1接收到了消息，如下图：