一、模式说明

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

Routingkey` 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： `item.insert

通配符规则：

• 匹配一个或多个词
• 匹配不多不少恰好1个词

举例：

item.#`：能够匹配`item.insert.abc` 或者 `item.insert 
item.*`：只能匹配`item.insert

图解：

• 红色Queue：绑定的是 usa.# ，因此凡是以 usa. 开头的 routing key 都会被匹配到
• 黄色Queue：绑定的是 #.news ，因此凡是以 .news 结尾的 routing key 都会被匹配

二、代码

①生产者

使用topic类型的Exchange，发送消息的routing key有3种： item.insert 、 item.update 、 item.delete ：

package com.itheima.rabbitmq.topic; 

import com.itheima.rabbitmq.util.ConnectionUtil; 
import com.rabbitmq.client.BuiltinExchangeType; 
import com.rabbitmq.client.Channel; 
import com.rabbitmq.client.Connection; 

/**
 * 通配符Topic的交换机类型为：topic 
*/ 
public class Producer { 
	//交换机名称 
	static final String TOPIC_EXCHAGE = "topic_exchange"; 
	
	//队列名称 
	static final String TOPIC_QUEUE_1 = "topic_queue_1"; 
	
	//队列名称 
	static final String TOPIC_QUEUE_2 = "topic_queue_2"; 
	public static void main(String[] args) throws Exception { 
		//创建连接 
		Connection connection = ConnectionUtil.getConnection(); 
		// 创建频道 
		Channel channel = connection.createChannel(); 
		/**
		 * 声明交换机 
		 * 参数1：交换机名称 
		 * 参数2：交换机类型，fanout、topic、topic、headers 
		*/ 
		channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC); 
		
		// 发送信息 
		String message = "新增了商品。Topic模式；routing key 为 item.insert " ; 
		channel.basicPublish(TOPIC_EXCHAGE, "item.insert", null, message.getBytes()); 
		System.out.println("已发送消息：" + message); 
		
		// 发送信息 
		message = "修改了商品。Topic模式；routing key 为 item.update" ; 
		channel.basicPublish(TOPIC_EXCHAGE, "item.update", null, message.getBytes()); 
		System.out.println("已发送消息：" + message); 
		
		// 发送信息 
		message = "删除了商品。Topic模式；routing key 为 item.delete" ; 
		channel.basicPublish(TOPIC_EXCHAGE, "item.delete", null, message.getBytes()); 
		System.out.println("已发送消息：" + message); 
		
		// 关闭资源
		channel.close(); 
		connection.close(); 
	} 
}

②消费者1

接收两种类型的消息：更新商品和删除商品

package com.itheima.rabbitmq.topic; 

import com.itheima.rabbitmq.util.ConnectionUtil; 
import com.rabbitmq.client.*; 
import java.io.IOException; 

public class Consumer1 { 
	public static void main(String[] args) throws Exception { 
		Connection connection = ConnectionUtil.getConnection(); 
		
		// 创建频道 
		Channel channel = connection.createChannel();
		 
		//声明交换机 
		channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC); 
		
		// 声明（创建）队列 
		/**
		 * 参数1：队列名称 
		 * 参数2：是否定义持久化队列 
		 * 参数3：是否独占本次连接 
		 * 参数4：是否在不使用的时候自动删除队列 
		 * 参数5：队列其它参数 
		*/ 
		channel.queueDeclare(Producer.TOPIC_QUEUE_1, true, false, false, null); 
		
		//队列绑定交换机 
		channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.update"); 
		channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.delete"); 
		
		//创建消费者；并设置消息处理 
		DefaultConsumer consumer = new DefaultConsumer(channel){ 
			@Override 
			/**
			 * consumerTag 消息者标签，在channel.basicConsume时候可以指定 
			 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) 
			 * properties 属性信息 
			 * body 消息 
			*/
			public void handleDelivery(String consumerTag, Envelope envelope, 
					AMQP.BasicProperties properties, byte[] body) throws IOException { 
				//路由key 
				System.out.println("路由key为：" + envelope.getRoutingKey()); 
				
				//交换机 
				System.out.println("交换机为：" + envelope.getExchange()); 
				
				//消息id 
				System.out.println("消息id为：" + envelope.getDeliveryTag()); 
				
				//收到的消息 
				System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8")); 
			} 
		};
		//监听消息 
		/**
		 * 参数1：队列名称 
		 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 
		 * 参数3：消息接收到后回调 
		*/ 
		channel.basicConsume(Producer.TOPIC_QUEUE_1, true, consumer); 
	} 
}

③消费者2

接收所有类型的消息：新增商品，更新商品和删除商品。

package com.itheima.rabbitmq.topic; 

import com.itheima.rabbitmq.util.ConnectionUtil; 
import com.rabbitmq.client.*; 
import java.io.IOException; 

public class Consumer2 { 
	public static void main(String[] args) throws Exception { 
		Connection connection = ConnectionUtil.getConnection(); 
		
		// 创建频道 
		Channel channel = connection.createChannel(); 
		
		//声明交换机 
		channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC); 
		
		// 声明（创建）队列 
		/**
		 * 参数1：队列名称 
		 * 参数2：是否定义持久化队列 
		 * 参数3：是否独占本次连接 
		 * 参数4：是否在不使用的时候自动删除队列 
		 * 参数5：队列其它参数 
		*/
		channel.queueDeclare(Producer.TOPIC_QUEUE_2, true, false, false, null); 
		
		//队列绑定交换机 
		channel.queueBind(Producer.TOPIC_QUEUE_2, Producer.TOPIC_EXCHAGE, "item.*");
		
		//创建消费者；并设置消息处理 
		DefaultConsumer consumer = new DefaultConsumer(channel){ 
			@Override 
			/**
			 * consumerTag 消息者标签，在channel.basicConsume时候可以指定 
			 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) 
			 * properties 属性信息 
			 * body 消息 
			*/ 
			public void handleDelivery(String consumerTag, Envelope envelope,
					AMQP.BasicProperties properties, byte[] body) throws IOException { 
					
				//路由key 
				System.out.println("路由key为：" + envelope.getRoutingKey()); 
				
				//交换机 
				System.out.println("交换机为：" + envelope.getExchange()); 
				
				//消息id 
				System.out.println("消息id为：" + envelope.getDeliveryTag()); 
				
				//收到的消息 
				System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8")); 
			} 
		};
		//监听消息 
		/**
		 * 参数1：队列名称 
		 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 
		 * 参数3：消息接收到后回调 
		*/ 
		channel.basicConsume(Producer.TOPIC_QUEUE_2, true, consumer); 
	} 
}

三、测试

启动所有消费者，然后使用生产者发送消息；在消费者对应的控制台可以查看到生产者发送对应 routing key 对应队列的消息；到达按照需要接收的效果；并且这些 routing key 可以使用通配符。

在执行完测试代码后，其实到 RabbitMQ 的管理后台找到 Exchanges 选项卡，点击 topic_exchange 的交换机，可以查看到如下的绑定：

总结

Topic主题模式可以实现 Publish/Subscribe发布与订阅模式 和 Routing路由模式 的功能；只是Topic 在配置 routing key 的时候可以使用通配符，显得更加灵活。