可以支持模糊匹配的路由key

特殊字符：
# 0个，一个或者多个
* 最少要一个

image.png

发送消息

image.png

实现

主题模式(Topics)

增加模糊路由，模糊匹配
*：一个
#：多个

image.png

生产者

  1. public class Producer {
  2.     private static Logger log = LoggerFactory.getLogger(Producer.class);
  4.     public static void main(String[] args) {
  6.         //    步骤
  7.         //    1. 创建连接工厂
  8.         log.info("1. 创建连接工厂");
  9.         ConnectionFactory connectionFactory = new ConnectionFactory();
  10.         //配置相关配置项
  11.         connectionFactory.setHost("112.74.175.76");
  12.         connectionFactory.setPort(5672);
  13.         connectionFactory.setUsername("admin");
  14.         connectionFactory.setPassword("admin");
  15.         connectionFactory.setVirtualHost("/");
  17.         //声明连接和通道
  18.         Connection connection = null;
  19.         Channel channel = null;
  20.         try {
  21.             //     2. 创建连接 Connection
  22.             log.info("2. 创建连接 Connection");
  23.             connection = connectionFactory.newConnection("生产者");
  25.             //    3. 通过连接获取通道Chanel
  26.             log.info("3. 通过连接获取通道Chanel");
  27.             channel = connection.createChannel();
  29.             //    5. 准备消息
  30.             log.info("5. 准备消息");
  31.             String msg = "Hello topic-exchange "+System.currentTimeMillis();
  33.             //    6. 准备交换机
  34.             log.info("6. 准备交换机");
  35.             String exchangeName = "topic-exchange";
  37.             //    7. 定义路由key
  38.             log.info("7. 定义路由key");
  39.             String routingKey = "com.order.user.test";
  41.             //    8. 定义交换机类型
  42.             log.info("8. 定义交换机类型");
  43.             String exchangeType = "topic";
  45.             /**
  46.              * @param 交换机
  47.              * @param 队列名称/路由key
  48.              * @param 属性配置
  49.              * @param 发送消息内容
  50.              */
  51.             channel.basicPublish(exchangeName,routingKey,null,msg.getBytes());
  52.             log.info("消息发送成功：{}",msg);
  53.         } catch (Exception e) {
  54.             e.printStackTrace();
  55.         } finally {
  56.             log.info("关闭连接");
  57.             //    7. 关闭通道
  59.             if (ObjectUtil.isNotEmpty(channel) && channel.isOpen()) {
  61.                 try {
  62.                     channel.close();
  63.                 } catch (IOException | TimeoutException e) {
  64.                     e.printStackTrace();
  65.                 }
  66.             }
  68.             //    8. 关闭连接
  69.             if (ObjectUtil.isNotEmpty(connection) && connection.isOpen()) {
  70.                 try {
  71.                     connection.close();
  72.                 } catch (IOException e) {
  73.                     e.printStackTrace();
  74.                 }
  75.             }
  76.         }
  77.     }
  78. }

image.png

消费者

  1. public class Consumer {
  2.     private static Logger log = LoggerFactory.getLogger(Consumer.class);
  3.     private static Runnable runnable = new Runnable() {
  4.         @Override
  5.         public void run() {
  6.             //    步骤
  7.             //    1. 创建连接工厂
  8.             log.info("1. 创建连接工厂");
  9.             ConnectionFactory connectionFactory = new ConnectionFactory();
  10.             //配置相关配置项
  11.             connectionFactory.setHost("112.74.175.76");
  12.             connectionFactory.setPort(5672);
  13.             connectionFactory.setUsername("admin");
  14.             connectionFactory.setPassword("admin");
  15.             connectionFactory.setVirtualHost("/");
  17.             //获取队列的名称
  18.             final String queueName = Thread.currentThread().getName();
  20.             //声明连接和通道
  21.             Connection connection = null;
  22.             Channel channel = null;
  23.             try {
  24.                 //     2. 创建连接 Connection
  25.                 log.info("2. 创建连接 Connection");
  26.                 connection = connectionFactory.newConnection("消费者-1");
  28.                 //    3. 通过连接获取通道Chanel
  29.                 log.info("3. 通过连接获取通道Chanel");
  30.                 channel = connection.createChannel();
  32.                 //    4. 通过通道创建交换机，声明队列，绑定关系，路由key，发送消息，接收消息
  33.                 //声明队列
  34.                 log.info("4. 通过通道接收消息");
  36.                 log.info("参数传递队列名称：{}", queueName);
  38.                 //5. 声明队列存储消息，如果不存在则被创建，rabbitmq不允许存在两个相同的队列名称
  39.                 log.info("5. 声明队列存储消息");
  41.                 //6. 定义接收消息的回调
  42.                 log.info("6. 定义接收消息的回调");
  43.                 Channel callbackChannel = channel;
  44.                 /**
  45.                  * @param queue 队列名称
  46.                  * @param durable 是否持久化
  47.                  * @param exclusive 是否排他，是否是私有的，如果为true，则会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
  48.                  * @param autoDelete  是否自动删除 当最后一个消费者断开连接之后是否自动删除消息
  49.                  * @param arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的生命周期
  50.                  */
  51.                 callbackChannel.basicConsume(queueName, true, new DeliverCallback() {
  52.                     @Override
  53.                     public void handle(String s, Delivery delivery) throws IOException {
  54.                         log.info(String.valueOf(delivery.getEnvelope().getDeliveryTag()));
  55.                         log.info("{}：收到的消息是：{}",queueName, new String(delivery.getBody(), StandardCharsets.UTF_8));
  56.                     }
  57.                 }, new CancelCallback() {
  58.                     @Override
  59.                     public void handle(String s) throws IOException {
  60.                         log.error("接受消息失败:{}", s);
  61.                     }
  62.                 });
  64.                 System.out.println("开始接收消息");
  65.                 //阻断执行
  66.                 // System.in.read();
  67.             } catch (Exception e) {
  68.                 e.printStackTrace();
  69.             } finally {
  70.                 log.info("关闭连接");
  71.                 //    7. 关闭通道
  73.                 if (ObjectUtil.isNotEmpty(channel) && channel.isOpen()) {
  75.                     try {
  76.                         channel.close();
  77.                     } catch (IOException | TimeoutException e) {
  78.                         e.printStackTrace();
  79.                     }
  80.                 }
  82.                 //    8. 关闭连接
  83.                 if (ObjectUtil.isNotEmpty(connection) && connection.isOpen()) {
  84.                     try {
  85.                         connection.close();
  86.                     } catch (IOException e) {
  87.                         e.printStackTrace();
  88.                     }
  89.                 }
  90.             }
  91.         }
  92.     };
  94.     public static void main(String[] args) {
  95.         //采用多线程的方式进行创建，使用线程名作为队列名
  96.         new Thread(runnable,"q1").start();
  97.         new Thread(runnable,"q2").start();
  98.         new Thread(runnable,"q3").start();
  99.         new Thread(runnable,"q4").start();
  100.     }
  101. }

image.png