在前面的教程里，我们构建了一个简单的日志记录系统。我们已经能够向许多消费者传送日志消息啦。
在本期，我们将会做一些修改，为它添加一个特性——让它可以只订阅消息的一部分。比如，我们能够只把致命的错误消息指向日志文件(节省磁盘的空间)，同时，我们还可以在控制台打印出所有的日志消息。
绑定
在前面几期的例子中，我们已经用了绑定。回忆一下这段代码：
channel.queueBind(queueName,EXCHANGE_NAME,"");
绑定是交换器和队列之前的一种联系。我们可以简单的解读为：队列对这个交换器里的信息感兴趣(手动滑稽)
绑定可以接受一个额外的路由键(RoutingKey)参数。为了避免与basic_publish参数混淆，我们可以称呼它为绑定键。这就是我们如何用key创建一个绑定。
channel.queueBind(queueName,EXCHANGE_NAME,"black");
绑定键的含义取决于交换类型。我们之前使用的是fanout类型，只不过我们忽略了他的value值。
直接交换——Direct Exchange
上一期，我们的日志系统可以把消息发送给所有的消费者。我们希望能够扩展这个功能，我们希望能基于这些日志的严重程度去过滤这些消息。举个栗子：我们可能想要一个能够吧日志消息写入磁盘的程序，这个程序只接受一种日志——报关键错误的日志。因为我们不需要在警告和不同信息上浪费磁盘空间。
我们之前使用的是fanout交换类型，这时候来看，他就显得不那么灵活了——它只能无意识的进行广播，也就是给所有人都发。这次我们将会使用direct交换类型，dierct交换类型背后的路由算法很简单：消息会传递给那个与自己的Routingkey完全匹配的队列。
为了明白这一点，请看图：
在这个图里，我们可以看到一个X，这个X是一个direct类型的交换器。第一个队列与一个橙色orange绑定，第二个队列有俩绑定，黑色black和绿色green。这样的交换机制下，将一条路由键是orange的消息发布到交换器中，它将与Q1绑定。将一条路由键是黑色或者绿色的消息发到交换器，它将与Q2绑定。所有其他的消息就会被丢弃。
多重绑定——Multiple bindings
用相同的绑定键绑定多个队列将会是完全合法的。在我们的示例中，我们可以在X和Q1之间添加黑色绑定键。在这样的情况下，direct exchange将会表现为fanout exchange，并将消息广播到所有匹配的队列中去。一个带有黑色路由键的消息将会被发送到Q1和Q2
发送日志——Emitting logs
我们将会给我们的日志系统使用这个模型。我们将会用direct类型代替之前的fanout类型。我们将会把日志的严重程度作为路由键。这样接受程序将会能够通过控制来选择它想要接受的消息。让我们关注一下发送日志。
像往常一样，我们首先需要创建一个交换器：
channel.exchangeDeclare(EXCHANGE_NAME,"direct");
我们准备发送信息：
channel.basicPublish(EXCHANGE_NAME,severity,null,message.getBytes());
为了简化问题，我们假设把“严重性”划分为“info(信息)”、“warn(警告)”、“error(错误)”三个级别。
订阅——Subscribing
接收消息的程序基本上与以前相同，不过有个例外——我们将会为我们感兴趣的“严重性”创建一个信息的绑定。
String queueName = channel.queueDeclare.getQueue();for(String serverity : argv){ channel.queueBind(queueName,EXCHANGE_NAME,severity);}
完整程序
EmitLogDirect.java
public class EmitLogDirect {
//设置交换器的名字
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] args) throws IOException {
//获取连接
Connection connection = ConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//声明交换器,给它名字,设置交换类型为direct
channel.exchangeDeclare(EXCHANGE_NAME,"direct");
//代码里手动设置严重性
String severity = getSeverity(args);
//待传递的消息内容
String message = getMessage(args);
//发送消息
channel.basicPublish(EXCHANGE_NAME,severity,null,message.getBytes());
System.out.println("[x] Sent '"+severity+"':'"+message+"'");
//关闭连接通道
channel.close();
connection.close();
}
private static String getSeverity(String[] strings) {
if (strings.length<1){
return "info";
}
return strings[0];
}
private static String getMessage(String[] strings) {
if (strings.length<2){
return "hello world";
}
return joinStrings(strings," ",1);
}
private static String joinStrings(String[] strings, String delimiter,int startIndex) {
int length = strings.length;
if (length == 0){
return "";
}
if (length<startIndex){
return "";
}
StringBuilder words = new StringBuilder(strings[startIndex]);
for (int i = startIndex+1 ; i < length; i++){
words.append(delimiter).append(strings[i]);
}
return words.toString();
}
}
ReceiveLogDirect.java
public class ReceiveLogsDirect {
//设置交换器的名字
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] args) throws IOException {
//获取连接
Connection connection = ConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//声明交换器,给它名字,设置交换类型为direct
channel.exchangeDeclare(EXCHANGE_NAME,"direct");
//得到队列的名字
String queueName = channel.queueDeclare().getQueue();
//截一下输入错误的情况
if (args.length<1){
System.err.println("Usage: ReceiveLogsDirect [info] [warning] [error]");
System.exit(1);
}
//同时我们看看我们到底需要什么程度的日志
String wholeSeverity = "";
//根据输入，确定程序想要收集的"严重性"，进行绑定。
for (String severity : args){
channel.queueBind(queueName,EXCHANGE_NAME,severity);
wholeSeverity = wholeSeverity+" "+severity;
}
System.out.println("[*] Waiting for"+wholeSeverity+" message.To exit press CTRL+C");
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("[x] Received'"+envelope.getRoutingKey()+"':'"+message+"'");
}
};
channel.basicConsume(queueName,true,consumer);
}
}
再次证明自己不是单纯的粘代码！！！
这里我测试的时候发现原版有一点不人性化，所以就自己在receive这边加了个wholeServerity，启动程序的时候告诉我们它可以收取到的严重性到底是什么程度。
还是给大家粘一下结果好啦！
结果
发送程序args设置示范：(请按照爱好,自行设置)
接受程序args设置示范：(请按照爱好,自行设置)
我用上面的发送程序，发送了一条Error的日志信息
只接受严重性为Error的程序结果：
接受严重性为info和warn的程序结果：
接受全部严重性程序的结果：
十分清晰明了！！！如果有哪些地方我理解的不到位，或者大家还是不懂的话。发评论我们可以探讨下~~
