当路由关键字非常多的时候没法管理，所以使用通配符匹配关键字。
*：唯一匹配的一个单词
#：匹配0个活多个单词
package com.ckfuture.topic.send; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.nio.charset.StandardCharsets; /** * 主题队列-消息生产者 */ public class Send { //定义交换机名称 private final static String EXCHANGE_NAME = "exchange_topic"; public static void main(String[] argv) throws Exception { //创建连接工厂 ConnectionFactory factory = new ConnectionFactory(); //连接工厂的地址 factory.setHost("localhost"); try ( //连接工厂创建连接 Connection connection = factory.newConnection(); //创建信道 Channel channel = connection.createChannel()) { //绑定交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC); String infoMessage = "普通信息！"; String errorMessage = "错误信息!"; String warningMessage = "警告信息!"; //路由key String infoRoutingKey="info.message.orange"; String errorRoutingKey="error.rabbit.lazy"; String warningRoutingKey="orange.warning.message"; //发送消息 channel.basicPublish(EXCHANGE_NAME, infoRoutingKey, null, infoMessage.getBytes(StandardCharsets.UTF_8)); channel.basicPublish(EXCHANGE_NAME, errorRoutingKey, null, errorMessage.getBytes(StandardCharsets.UTF_8)); channel.basicPublish(EXCHANGE_NAME, warningRoutingKey, null, warningMessage.getBytes(StandardCharsets.UTF_8)); System.out.println(" [x] Sent ‘" + infoMessage + "‘"); System.out.println(" [x] Sent ‘" + errorMessage + "‘"); System.out.println(" [x] Sent ‘" + warningMessage + "‘"); } } }
package com.ckfuture.topic.recv; import com.rabbitmq.client.*; /** * 主题队列-消息消费者 */ public class Recv01 { private final static String EXCHANGE_NAME = "exchange_topic"; public static void main(String[] argv) throws Exception { //创建连接工厂 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); //连接工厂创建连接 Connection connection = factory.newConnection(); //创建信道 Channel channel = connection.createChannel(); //绑定交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC); //获取队列（排他队列） String queueName = channel.queueDeclare().getQueue(); //绑定队列和交换机 String routingKey="#.message.#"; channel.queueBind(queueName,EXCHANGE_NAME,routingKey); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received ‘" + message + "‘"); }; channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { }); } }
package com.ckfuture.topic.recv; import com.rabbitmq.client.*; /** * 主题队列-消息消费者 */ public class Recv02 { private final static String EXCHANGE_NAME = "exchange_topic"; public static void main(String[] argv) throws Exception { //创建连接工厂 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); //连接工厂创建连接 Connection connection = factory.newConnection(); //创建信道 Channel channel = connection.createChannel(); //绑定交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC); //获取队列（排他队列） String queueName = channel.queueDeclare().getQueue(); //绑定队列和交换机 String routingKey="*.rabbit.*"; channel.queueBind(queueName,EXCHANGE_NAME,routingKey); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received ‘" + message + "‘"); }; channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { }); } }
在管理控制台查看：
查看详情：
package com.ckfuture.topic.send; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.nio.charset.StandardCharsets; /** * 主题队列-消息生产者 */ public class Send { //定义交换机名称 private final static String EXCHANGE_NAME = "exchange_topic"; public static void main(String[] argv) throws Exception { //创建连接工厂 ConnectionFactory factory = new ConnectionFactory(); //连接工厂的地址 factory.setHost("localhost"); try ( //连接工厂创建连接 Connection connection = factory.newConnection(); //创建信道 Channel channel = connection.createChannel()) { //绑定交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC); String infoMessage = "普通信息！"; String errorMessage = "错误信息!"; String warningMessage = "警告信息!"; //路由key String infoRoutingKey="info.message.orange"; String errorRoutingKey="error.rabbit.lazy"; String warningRoutingKey="orange.warning.message"; //发送消息 channel.basicPublish(EXCHANGE_NAME, infoRoutingKey, null, infoMessage.getBytes(StandardCharsets.UTF_8)); channel.basicPublish(EXCHANGE_NAME, errorRoutingKey, null, errorMessage.getBytes(StandardCharsets.UTF_8)); channel.basicPublish(EXCHANGE_NAME, warningRoutingKey, null, warningMessage.getBytes(StandardCharsets.UTF_8)); System.out.println(" [x] Sent ‘" + infoMessage + "‘"); System.out.println(" [x] Sent ‘" + errorMessage + "‘"); System.out.println(" [x] Sent ‘" + warningMessage + "‘"); } } }
Recv01按照匹配规则可以接收两条消息：
Recv02按照匹配规则可以接收一条消息：
原文：https://www.cnblogs.com/ckfuture/p/14407534.html