Topics(动态路由)

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

*：匹配不多不少刚好一个单词。
#：匹配一个或多个词。
举例：
audit.#可以匹配audit.x.y或者dudit.x。
audit.*只能匹配audit.x.

1.开发生产者

public class Provider {public static void main(String[] args) throws IOException {Connection connection = RabbitMqUtil.getConnection();Channel channel = connection.createChannel();String exchange = "topicexchange";//将通道声明指定交换机 参数1：指明交换机名称  参数2：交换机的类型 fanout广播类型channel.exchangeDeclare(exchange,"topic");//发送消息String routingKey = "user.save.okk";channel.basicPublish(exchange,routingKey,null,("这是topicsexchange发布的"+routingKey+"消息").getBytes());//释放资源RabbitMqUtil.closeConnectionAndChannel(channel,connection);}
}

2.开发消费者(user.*)

public class Consumer1 {public static void main(String[] args) throws IOException {//获取连接对象Connection connection = RabbitMqUtil.getConnection();//获取连接通道Channel channel = connection.createChannel();String exchange = "topicexchange";//通道绑定交换机channel.exchangeDeclare(exchange,"topic");//临时队列String queueName = channel.queueDeclare().getQueue();//基于RoutingKey绑定交换机和队列channel.queueBind(queueName,exchange,"user.*");channel.basicConsume(queueName,true,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("consumer1得到："+new String(body));}});//注意这里不能关闭通道和连接，因为要一直监听}
}

3.开发消费者(user.#)

public class Consumer2 {public static void main(String[] args) throws IOException {//获取连接对象Connection connection = RabbitMqUtil.getConnection();//获取连接通道Channel channel = connection.createChannel();String exchange = "topicexchange";//通道绑定交换机channel.exchangeDeclare(exchange,"topic");//临时队列String queueName = channel.queueDeclare().getQueue();//基于RoutingKey绑定交换机和队列channel.queueBind(queueName,exchange,"user.#");channel.basicConsume(queueName,true,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("consumer2得到："+new String(body));}});//注意这里不能关闭通道和连接，因为要一直监听}
}