Direct exchange
前言
我们构建了一个简单的日志记录系统。我们能够向许多接收者广播日志消息。在本节我们将向其中添加一些特别的功能-比方说我们只让某个消费者订阅发布的部分消息。例如我们只把严重错误消息定向存储到日志文件(以节省磁盘空间)，同时仍然能够在控制台上打印所有日志消息。
绑定是交换机和队列之间的桥梁关系。也可以这么理解：
队列只对它绑定的交换机的消息感兴趣。绑定用参数：routingKey来表示也可称该参数为binding key，创建绑定我们用代码:channel.queueBind(queueName, EXCHANGE_NAME, "routingKey");绑定之后的意义由其交换类型决定。
Direct exchange介绍
我们希望将日志消息写入磁盘的程序仅接收严重错误(errros)，而不存储哪些警告(warning)或信息(info)日志消息避免浪费磁盘空间。Fanout这种交换类型并不能给我们带来很大的灵活性-它只能进行无意识的广播，在这里我们将使用direct这种类型来进行替换，这种类型的工作方式是，消息只去到它绑定的routingKey队列中去。
在上面这张图中，我们可以看到X绑定了两个队列，绑定类型是direct。队列Q1绑定键为orange，队列Q2绑定键有两个:一个绑定键为black，另一个绑定键为green.
在这种绑定情况下，生产者发布消息到exchange上，绑定键为orange的消息会被发布到队列
Q1。绑定键为blackgreen和的消息会被发布到队列Q2，其他消息类型的消息将被丢弃。
多重绑定
当然如果exchange的绑定类型是direct，但是它绑定的多个队列的key如果都相同，在这种情况下虽然绑定类型是direct但是它表现的就和fanout有点类似了，就跟广播差不多，如下图所示。
实战代码
工具类：
public class untils {
public static Channel getChannel() throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.231.132");
factory.setUsername("admin");
factory.setPassword("123");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
return channel;
}
}消费者1：
public class ReceiveLogsDirect01 {
private static final String EXCHANG_NAME="direct_logs";
public static void main(String[] args) throws Exception{
Channel channel = untils.getChannel();
/**
* 声明交换机
*/
channel.exchangeDeclare(EXCHANG_NAME, BuiltinExchangeType.DIRECT);
String queueName="disk";
channel.queueDeclare(queueName,false,false,false,null);
channel.queueBind(queueName,EXCHANG_NAME,"error");
System.out.println("正在准备接收消息....");
DeliverCallback deliverCallback=(Consumer,delivert)->
{
String s = new String(delivert.getBody());
System.out.println("错误消息--->："+s);
};
channel.basicConsume(queueName,true,deliverCallback,consumerTag ->{} );
}
}消费者2：
public class ReceiveLogsDirect02 {
private static final String EXCHANG_NAME="direct_logs";
public static void main(String[] args) throws Exception{
Channel channel = untils.getChannel();
/**
* 声明交换机
*/
channel.exchangeDeclare(EXCHANG_NAME, BuiltinExchangeType.DIRECT);
String queueName="console";
channel.queueDeclare(queueName,false,false,false,null);
channel.queueBind(queueName,EXCHANG_NAME,"info");
channel.queueBind(queueName,EXCHANG_NAME,"warning");
System.out.println("正在准备接收消息....");
DeliverCallback deliverCallback=(Consumer, delivert)->
{
String s = new String(delivert.getBody());
System.out.println("info warning消息--->："+s);
};
channel.basicConsume(queueName,true,deliverCallback,consumerTag ->{} );
}
}生产者：
public class EmitLogDirect {
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] args)throws Exception {
Channel channel = untils.getChannel();
//创建多个bindingKeyMap
Map<String,String> bindingKeyMap=new HashMap<>();
bindingKeyMap.put("info","普通info消息");
bindingKeyMap.put("warning","警告warning消息");
bindingKeyMap.put("error","错误error消息");
//debug没有的消费者接收这个消息，所以就失去
bindingKeyMap.put("debug","调试debug消息");
for (Map.Entry<String,String> bindingKeyEntry:bindingKeyMap.entrySet())
{
String bindingKey=bindingKeyEntry.getKey();
String message=bindingKeyEntry.getValue();
channel.basicPublish(EXCHANGE_NAME,bindingKey,null,message.getBytes(StandardCharsets.UTF_8));
System.out.println("生产者发出消息"+message);
}
}
}结果：
- 点赞
- 收藏
- 关注作者
评论（0）