Topic类型的Exchange是要进行路由键匹配的。此时需要通过路由键将队列绑定要一个交换器上。规则如下：

符号“#”匹配一个或多个词，例如：“logs.#”能够匹配到“logs.error”、“logs.info.toc”

符号“*”只能匹配一个词。例如：“logs.*” 只能匹配到“logs.error”,不能匹配到“logs.info.toc” 。

如下图：

下面就直接上代码吧！！

Consumer：

package topic;import java.io.IOException;import java.util.concurrent.TimeoutException;import com.rabbitmq.client.AMQP;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.Consumer;import com.rabbitmq.client.DefaultConsumer;import com.rabbitmq.client.Envelope;public class TopicConsumer {private static final String EXCHANGE_NAME = "exchange_topic";public static void main(String[] argv) throws IOException, TimeoutException  {new ExchangeTopic("logs.info");new ExchangeTopic("logs.*");new ExchangeTopic("logs.#");}static class ExchangeTopic{public  ExchangeTopic(final String routingKey) throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();//rabbitmq监听IPfactory.setHost("192.168.249.128");//rabbitmq监听默认端口factory.setPort(5672);//设置访问的用户factory.setUsername("test");factory.setPassword("test");Connection connection = factory.newConnection();Channel channel = connection.createChannel();//声明路由名字和类型channel.exchangeDeclare(EXCHANGE_NAME, "topic", false, true, null);//队列名称String queueName = routingKey + ".queue";//创建队列channel.queueDeclare(queueName, false, false, true, null);//把队列绑定到路由上channel.queueBind(queueName, EXCHANGE_NAME, routingKey);System.out.println(" [routingKey = "+ routingKey +"] Waiting for msg....");Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body, "UTF-8");System.out.println("[routingKey = "+ routingKey +"] Received msg is '" + message + "'");}};channel.basicConsume(queueName, true, consumer);}}}

Producer：

package topic;import java.io.IOException;import java.util.concurrent.TimeoutException;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;public class TopicProducer {private static final String EXCHANGE_NAME = "exchange_topic";public static void main(String[] argv) throws Exception{new ExchangeTopic("logs.info", "logs Info test ！！");new ExchangeTopic("logs.error", "logs error test ！！");new ExchangeTopic("logs.error.toc", "logs error toc test ！！");}static class ExchangeTopic{public ExchangeTopic(String routingKey,String message) throws IOException, TimeoutException{ConnectionFactory factory = new ConnectionFactory();//rabbitmq监听IPfactory.setHost("192.168.249.128");//rabbitmq监听默认端口factory.setPort(5672);//设置访问的用户factory.setUsername("test");factory.setPassword("test");Connection connection = factory.newConnection();Channel channel = connection.createChannel();//声明路由名字和类型channel.exchangeDeclare(EXCHANGE_NAME, "topic", false, true, null);channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());System.out.println("[routingKey = "+routingKey+"] Sent msg is '" + message + "'");channel.close();connection.close();}}}

结果：

从结果中可以看出，路由键为"logs.info"只匹配到了Producer路由键是"logs.info"的信息，路由键为"logs.*"则匹配到了Producer路由键是"logs.info","logs.error"两条信息，而路由键为"logs.#"则匹配到了Producer路由键是"logs.info","logs.error","logs.error.toc"三条信息。

本文由职坐标整理并发布，希望对同学们有所帮助。了解更多详情请关注职坐标系统运维之Exchange频道！