1.路由交换机
这种类型的工作方式是,消息只去到它绑定的routing key的队列中去
2.实战
(1)图示
(2)代码实现
package com.rabbitmq.six;
import com.rabbitmq.client.Channel;
import com.rabbitmq.utils.MqUtils;
import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;
public class DirectLogs {
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MqUtils.getChannel();
Scanner scanner = new Scanner(System.in);
while(scanner.hasNext()) {
String message = scanner.next();
// 先测试info然后改成warning重启,然后改成error重启测试
channel.basicPublish(EXCHANGE_NAME, "info", null ,message.getBytes("UTF-8"));
System.out.println("生产者发出消息:" + message);
}
}
}
package com.rabbitmq.six;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.utils.MqUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ReceiveLogsDirect01 {
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MqUtils.getChannel();
// 声明交换机-直接交换机(Direct)
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
// 声明队列
channel.queueDeclare("console", false, false,false, null);
// 绑定交换机与队列
channel.queueBind("console", EXCHANGE_NAME, "info");
channel.queueBind("console", EXCHANGE_NAME, "warning");
// 接收消息
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("ReceiveLogsDirect01控制台打印接收到的消息:" + new String(message.getBody(), "UTF-8"));
};
channel.basicConsume("console",true, deliverCallback,consumerTag->{});
}
}
package com.rabbitmq.six;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.utils.MqUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ReceiveLogsDirect02 {
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MqUtils.getChannel();
// 声明交换机-直接交换机(Direct)
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
// 声明队列
channel.queueDeclare("disk", false, false,false, null);
// 绑定交换机与队列
channel.queueBind("disk", EXCHANGE_NAME, "error");
// 接收消息
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("ReceiveLogsDirect02控制台打印接收到的消息:" + new String(message.getBody(), "UTF-8"));
};
channel.basicConsume("disk",true, deliverCallback,consumerTag->{});
}
}
(3)执行效果