翻译官网的文章已经翻译了几天了,这份官方文档写的总体算是很简洁易懂。它让我们很快的入门并了解了RabbitMQ的运作原理和使用方式。本篇最后介绍一下Exchange的另外两种类别,即direct和topic。对官网文档的翻译工作也将告一段落,接下来会探讨异步消息队列在spring项目中的集成。但现在还是先把RabbitMQ的另外两种使用方式来介绍完。
Direct类型的Exchange
一.基础知识点
1.direct类别的Exchange.
记得上一章节我们使用的是fanout类别的Exchange,它的特点是消费者不加区分地订阅所有消息,即所有的消费者都能订阅并收到广播到这个Exchange的消息。这一节我们希望稍作改动,让不同消费者能够订阅不同的消息。要实现这个功能,需要使用Direct的Exchange.
其运作原理如图所示。
首先,将Exchange设为direct类型,可用这句话来实现:channel.exchangeDeclare(EXCHANGE_NAME, "direct")。
然后,我们发现,queue和Producer之间,除了多出Exchange,还多出了另外一条线索,也就是所谓的routeKey。其实它就是对Exchange的进一步分类,这样做了以后,当我们广播消息的时候,消息就进行了两层分类,第一层是Exchange(图中的X),第二层就是RouteKey(图中的Orange,green等)。我们使用如下语句来发布消息,channel.basicPublish(EXCHANGE_NAME, orange, null, message.getBytes());其中第二个参数就是RouteKey.
再然后,消费者如果想要接收指定的消息,只需要在符合一定条件的queue的集合中去寻找即可。比如说,我想接收颜色为orange的消息,则去Q1中寻找,我想接收颜色为black或者green的消息,就去Q2中寻找。当然Q1和Q2需要进行绑定才能将特定的消息放进来。使用如下语句即可以将Queue与Exchange和RouteKey绑定。channel.queueBind(queueName, EXCHANGE_NAME, orange);其中第三个参数就是RouteKey。
二.Demo
话不多说,看下面的demo,这个demo希望对不同的日志信息做不同的处理,相信读者很快就会理解其中原理。
首先是生产者类。
package com.xdx.learn; import java.io.IOException;
import java.util.concurrent.TimeoutException; import net.sf.json.JSONObject; import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory; public class EmitLogDirect {
private final static String EXCHANGE_NAME="direct_logs";//交换机名称为log public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory=new ConnectionFactory();
factory.setHost("192.168.1.195");//服务器ip
factory.setPort(5672);//端口
factory.setUsername("xdx");//登录名
factory.setPassword("xxxxx");//密码
Connection connection=factory.newConnection();//建立连接
Channel channel=connection.createChannel();//建立频道
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);//在频道里声明一个交换机,类型定位direct
System.out.println(channel+"发布20条info日志消息");
for(int i=0;i<20;i++){
String message="info日志消息"+i;
channel.basicPublish(EXCHANGE_NAME, "info", null, message.getBytes());//发布消息,发布到EXCHANGE_NAME,并且routeKey标为info
System.out.println(message);
}
System.out.println(channel+"发布10条error日志消息");
for(int i=0;i<10;i++){
String message="error日志消息"+i;
channel.basicPublish(EXCHANGE_NAME, "error", null, message.getBytes());//发布消息,发布到EXCHANGE_NAME,并且routeKey标为error
System.out.println(message);
}
System.out.println(channel+"发布5条debug日志消息");
for(int i=0;i<5;i++){
String message="debug日志消息"+i;
channel.basicPublish(EXCHANGE_NAME, "debug", null, message.getBytes());//发布消息,发布到EXCHANGE_NAME,并且routeKey标为error
System.out.println(message);
}
channel.close();
connection.close();
}
}
再看消费者类。
package com.xdx.learn; import java.io.IOException;
import java.util.concurrent.TimeoutException; import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
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 ReceiveLogsDirect {
private final static String EXCHANGE_NAME = "direct_logs";
public static void main(String[] args) throws IOException, TimeoutException {
// 下面的配置与生产者相对应
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.1.195");// 服务器ip
factory.setPort(5672);// 端口
factory.setUsername("xdx");// 登录名
factory.setPassword("xxxxx");// 密码
Connection connection = factory.newConnection();// 连接
final Channel channel = connection.createChannel();// 频道
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
final String queueName = channel.queueDeclare().getQueue();// 生成一个独立的,非持久的,自动删除的queue
channel.queueBind(queueName, EXCHANGE_NAME, "info");// 绑定queue和exchange,还有routekey。这样队列中就有通过EXCHANGE_NAME发布的消息。
channel.queueBind(queueName, EXCHANGE_NAME, "debug");
System.out.println(" messages from channel:"+ channel+",queue:"+ queueName
+ ". To exit press CTRL+C");
// defaultConsumer实现了Consumer,我们将使用它来缓存生产者发送过来储存在队列中的消息。当我们可以接收消息的时候,从中获取。
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body)
throws IOException {
String message = new String(body, "UTF-8");
System.out.println("channel:"+channel+",queue:"+queueName+",consumer:"+this.getConsumerTag()+" Received '" + message + "'");
// channel.basicAck(envelope.getDeliveryTag(),false);
try {
Thread.sleep(300);
} catch (Exception e) {
}
}
};
channel.basicConsume(queueName, true, consumer);//自动回复,消息发出后队列自动消除
} }
首先我们先运行消费者类,这样它就可以守株待兔,订阅消息了。然后再运行生产者类。
生产者类的控制台打印出如下信息:
消费者类的控制台打印出如下信息。
三.总结
1.在生产者类中,我们将消息分为三种类别,info,error,debug,并分别广播进了以这三个类别作为routeKey的Exchange中。这是第一步,对消息做分类。
2.在消费者类中,我们想要查看info和debug这两种类型的消息,所以我们将queue与这两种类别的Exchange进行了绑定,通过channel.queueBind(queueName, EXCHANGE_NAME, "info");channel.queueBind(queueName, EXCHANGE_NAME, "debug");这两条代码。这样,这个queue中就存满了这两种类型的信息,消费者就可以使用它们了。至于error类别的日志消息,因为没有消费者订阅,所以它们会被丢弃。
应该来说demo是非常简单也非常恰当的解释了Direct类型的Exchange的运作机制。
Topic类型的Exchange
topic类型的Exchange其实是在Direct类型上进行了扩展,如果direct类型的Exchange只能从一个维度来对Exchange进行区分,那么topic类型的Exchange以一种更为灵活的方式对Exchange进行了多维度的区分。关键还是在RouteKey上。
一.基础知识点
来看topic类型的Exchange的示意图。
可以看到,现在的RouteKey不在是一个单一的一个字段来描述,而是可以有多个字段来共同描述。每个字段代表一个维度。例如上述的例子,我们描述一种动物,第一个字段代表速度,第二个字段代表颜色,第三个字段代表的是物种。字段之间用点号隔开,即<speed>.<colour>.<species>。并且,它还为我们提供了两个很好用的通配符。
*:代表一个字段
#:代表0个或者多个字段。
所以*.orange.*代表的是所以橙色的动物,*.*.rabbit代表的是所以兔子,而lazy.#代表的是所有反应缓慢的动物。那么:quick.orange.rabbit匹配Q1和Q2,lazy.orange.elephant也是,quick.orange.fox匹配Q1.lazy.brown.fox匹配Q2.
值得注意的是,如果我将一个routeKey写为#,则这时候Exchange相当于fanout类型,而如果routeKey中即没有*又没有#,则相当于Direct类型了。
二.Demo
话不多说,还是来看一个demo吧。在这个demo中,我们将消息分为两个维度<project><severity>。
生产者类
package com.xdx.learn; import java.io.IOException;
import java.util.concurrent.TimeoutException; import net.sf.json.JSONObject; import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory; public class EmitLogTopic {
private final static String EXCHANGE_NAME="topic_logs";//交换机名称为log public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory=new ConnectionFactory();
factory.setHost("192.168.1.195");//服务器ip
factory.setPort(5672);//端口
factory.setUsername("xdx");//登录名
factory.setPassword("xxxx");//密码
Connection connection=factory.newConnection();//建立连接
Channel channel=connection.createChannel();//建立频道
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);//在频道里声明一个交换机,类型定位topic
System.out.println(channel+"发布3条project1.info日志消息");
for(int i=0;i<3;i++){
String message="project1.info日志消息"+i;
channel.basicPublish(EXCHANGE_NAME, "project1.info", null, message.getBytes());//发布消息,发布到EXCHANGE_NAME,并且routeKey标为info
System.out.println(message);
}
System.out.println(channel+"发布3条project1.error日志消息");
for(int i=0;i<3;i++){
String message="project1.error日志消息"+i;
channel.basicPublish(EXCHANGE_NAME, "project1.error", null, message.getBytes());//发布消息,发布到EXCHANGE_NAME,并且routeKey标为error
System.out.println(message);
}
System.out.println(channel+"发布3条project1.debug日志消息");
for(int i=0;i<3;i++){
String message="project1.debug日志消息"+i;
channel.basicPublish(EXCHANGE_NAME, "project1.debug", null, message.getBytes());//发布消息,发布到EXCHANGE_NAME,并且routeKey标为error
System.out.println(message);
}
System.out.println(channel+"发布3条project2.info日志消息");
for(int i=0;i<3;i++){
String message="project2.info日志消息"+i;
channel.basicPublish(EXCHANGE_NAME, "project2.info", null, message.getBytes());//发布消息,发布到EXCHANGE_NAME,并且routeKey标为info
System.out.println(message);
}
System.out.println(channel+"发布3条project2.error日志消息");
for(int i=0;i<3;i++){
String message="project2.error日志消息"+i;
channel.basicPublish(EXCHANGE_NAME, "project2.error", null, message.getBytes());//发布消息,发布到EXCHANGE_NAME,并且routeKey标为error
System.out.println(message);
}
System.out.println(channel+"发布3条project2.debug日志消息");
for(int i=0;i<3;i++){
String message="project2.debug日志消息"+i;
channel.basicPublish(EXCHANGE_NAME, "project2.debug", null, message.getBytes());//发布消息,发布到EXCHANGE_NAME,并且routeKey标为error
System.out.println(message);
}
channel.close();
connection.close();
}
}
消费者类
package com.xdx.learn; import java.io.IOException;
import java.util.concurrent.TimeoutException; import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
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 ReceiveLogsTopic {
private final static String EXCHANGE_NAME = "topic_logs";
public static void main(String[] args) throws IOException, TimeoutException {
// 下面的配置与生产者相对应
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.1.195");// 服务器ip
factory.setPort(5672);// 端口
factory.setUsername("xdx");// 登录名
factory.setPassword("xxxx");// 密码
Connection connection = factory.newConnection();// 连接
final Channel channel = connection.createChannel();// 频道
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
final String queueName = channel.queueDeclare().getQueue();// 生成一个独立的,非持久的,自动删除的queue
channel.queueBind(queueName, EXCHANGE_NAME, "*.info");// 绑定queue和exchange,还有routekey。这样队列中就有通过EXCHANGE_NAME发布的消息。
channel.queueBind(queueName, EXCHANGE_NAME, "project1.*");
System.out.println(" messages from channel:"+ channel+",queue:"+ queueName
+ ". To exit press CTRL+C");
// defaultConsumer实现了Consumer,我们将使用它来缓存生产者发送过来储存在队列中的消息。当我们可以接收消息的时候,从中获取。
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body)
throws IOException {
String message = new String(body, "UTF-8");
System.out.println("channel:"+channel+",queue:"+queueName+",consumer:"+this.getConsumerTag()+" Received '" + message + "'");
// channel.basicAck(envelope.getDeliveryTag(),false);
try {
Thread.sleep(300);
} catch (Exception e) {
}
}
};
channel.basicConsume(queueName, true, consumer);//自动回复,消息发出后队列自动消除
} }
运行消费者类,再运行生产者类,分别打印出。
生产者类:
生产者类:
大家可以根据代码和结果来验证我们的假设,这里就不做总结了。