各大网站rss订阅源地址百度收录关键词查询
一、Topics交换机
之前我们学习了发布/订阅模式、路由模式，其中一个使用了最简单的fanout交换机，一个使用了带个性化的direct交换机，尽管direct在一定程度上提供了个性化操作入口，改善了我们的日志系统，但是还远远不够，因为需求总是千奇百怪的，direct的限制在于：不支持多重标准。
还是以我们的日志系统为例，我们不仅需要根据严重级别来处理日志，还可能需要根据不同的来源作出不同的处理，这时候就需要使用更大的交换机：Topics。Topics交换机可以实现更加复杂的消息发送规则，即发送消息时，指定更为复杂的routingKey。
二、Topics的配置规则
使用Topics的交换机，我们就不能使用任意名字的routingKey，其必须遵守其规则：以一串words命名，使用逗号分隔。
比如：”stock.usd.nyse”, “nyse.vmw”, “quick.orange.rabbit”.
可以随意地增加routingKey的words个数，但是最大长度为255个字节。
Topics的逻辑跟direct大同小异，它与direct一样，可以发给多个特定的消费者，但是其有两点值得关注的：
①：*可以代表任何一个单词
②：#可以代表0个或多个单词
如下图：
我们的生产者打算发送关于动物的消息：
Q1队列绑定了：* . orange . *
Q2队列绑定了：* . * . rabbit、lazy . #
这里说明Q1对橘色的动物感兴趣，而Q2对兔子、懒的动物感兴趣。
一下是一些消息指定的routingKey，以及它将会发送到的队列：
quick.orange.rabbit：二者
“lazy.orange.elephant”：二者
“quick.orange.fox”：Q1
“lazy.brown.fox” ：Q2
“quick.brown.fox”：无（该消息将会被丢弃）
“orange”：无
“quick.orange.male.rabbit”：无（位数过长，消息将被丢弃）
“lazy.orange.male.rabbit”：Q2
三、特殊的Topics
Topics交换机非常的强大，它也可以实现fanout与direct那样的使用效果，当一个队列的routingKey绑定为：“#”时，它接收所有消息；当其绑定为“*”、“#”都没有使用时，其相当于direct。
四、代码实现
Topics交换机非常的强大，它也可以实现fanout与direct那样的使用效果，当一个队列的routingKey绑定为：“#”时，它接收所有消息；当其绑定为“*”、“#”都没有使用时，其相当于direct。
我们将error日志分层处理，sql异常的保存在sql异常相关的磁盘，service异常保存在service相关的目录，为了简单，我们使用打印语句代替将要执行的操作。
EmitTopicLog.java：
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class EmitTopicLog {private static final String EXCHANGE_NAME = "topic_logs";public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, "topic");//debug日志for (int i = 1; i <= 3; i++) {String message = "debug_message" + i;channel.basicPublish(EXCHANGE_NAME, "debug", null, message.getBytes());}//info日志for (int i = 1; i <= 3; i++) {String message = "info_message" + i;channel.basicPublish(EXCHANGE_NAME, "info", null, message.getBytes());}//sql error日志for (int i = 1; i <= 3; i++) {String message = "error_message" + i;channel.basicPublish(EXCHANGE_NAME, "userQuery.sql.error", null, message.getBytes());}//sql error日志for (int i = 1; i <= 3; i++) {String message = "error_message" + i;channel.basicPublish(EXCHANGE_NAME, "userUpdate.sql.error", null, message.getBytes());}//service error日志for (int i = 1; i <= 3; i++) {String message = "error_message" + i;channel.basicPublish(EXCHANGE_NAME, "userLogin.service.error", null, message.getBytes());}channel.close();connection.close();}
}
ReceiveLog1.java：
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ReceiveLog1 {private static final String EXCHANGE_NAME = "topic_logs";public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, "topic");String queueName = channel.queueDeclare().getQueue();channel.queueBind(queueName, EXCHANGE_NAME, "debug");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(message);}};channel.basicConsume(queueName,true,consumer);}
}
ReceiveLog2.java：
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ReceiveLog2 {private static final String EXCHANGE_NAME = "topic_logs";public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, "topic");String queueName = channel.queueDeclare().getQueue();channel.queueBind(queueName, EXCHANGE_NAME, "info");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(message);}};channel.basicConsume(queueName,true,consumer);}
}
ReceiveLog3.java：
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ReceiveLog3 {private static final String EXCHANGE_NAME = "topic_logs";public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, "topic");String queueName = channel.queueDeclare().getQueue();channel.queueBind(queueName, EXCHANGE_NAME, "*.sql.error");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("将"+message+"记录到sql异常相关的目录");}};channel.basicConsume(queueName,true,consumer);}
}
ReceiveLog4.java：
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ReceiveLog4 {private static final String EXCHANGE_NAME = "topic_logs";public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, "topic");String queueName = channel.queueDeclare().getQueue();channel.queueBind(queueName, EXCHANGE_NAME, "*.service.error");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");FileUtil.save(message);System.out.println("将"+message+"记录到service异常相关的目录");}};channel.basicConsume(queueName,true,consumer);}
}
运行结果：
五、总结
将发布/订阅模式、路由模式、Topic模式进行对比，很容易发现以下结论：
| Publish/Subcribe、Routing、Topics都是围绕交换机进行的，差别仅是交换机的类型不同。 |
下篇将和大家一起来探索远程调用RPC