RabbitMQ Exchange分发消息时根据类型的不同分发策略有区别，目前共四种类型：direct、fanout、topic、headers 。headers 匹配 AMQP 消息的 header 而不是路由键，此外 headers 交换器和 direct 交换器完全一致，但性能差很多，目前几乎用不到了。下面分别以实例的方式对这几种exchange进行讲解。
direct
首先我们以路由的方式对消息进行过滤，代码如下：
生产者
public class RoutingSendDirect {
private static final String EXCHANGE_NAME = "direct_test";
private static final String[] routingKeys = new String[]{"info" ,"warning", "error"};
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("localhost");
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME,"direct");
for(String key : routingKeys){
String message = "RoutingSendDirect Send the message level:" + key;
channel.basicPublish(EXCHANGE_NAME,key,null,message.getBytes());
System.out.println("RoutingSendDirect Send"+key +"':'" + message);
}
channel.close();
connection.close();
}
}
消费者
public class ReceiveDirect1 {
// 交换器名称
private static final String EXCHANGE_NAME = "direct_test";
// 路由关键字
private static final String[] routingKeys = new String[]{"info" ,"warning"};
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("localhost");
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME,"direct");
//获取匿名队列名称
String queueName=channel.queueDeclare().getQueue();
for(String key : routingKeys){
channel.queueBind(queueName,EXCHANGE_NAME,key);
System.out.println("ReceiveDirect1 exchange:"+EXCHANGE_NAME+"," +
" queue:"+queueName+", BindRoutingKey:" + key);
}
System.out.println("ReceiveDirect1 Waiting for messages");
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = new String(body,"UTF-8");
System.out.println("ReceiveDirect1 Received '" + envelope.getRoutingKey() + "':'" + msg + "'");
}
};
channel.basicConsume(queueName, true, consumer);
}
}
public class ReceiveDirect2 {
// 交换器名称
private static final String EXCHANGE_NAME = "direct_test";
// 路由关键字
private static final String[] routingKeys = new String[]{"error"};
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("localhost");
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME,"direct");
//获取匿名队列名称
String queueName=channel.queueDeclare().getQueue();
for(String key : routingKeys){
channel.queueBind(queueName,EXCHANGE_NAME,key);
System.out.println("ReceiveDirect2 exchange:"+EXCHANGE_NAME+"," +
" queue:"+queueName+", BindRoutingKey:" + key);
}
System.out.println("ReceiveDirect2 Waiting for messages");
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = new String(body,"UTF-8");
System.out.println("ReceiveDirect2 Received '" + envelope.getRoutingKey() + "':'" + msg + "'");
}
};
channel.basicConsume(queueName, true, consumer);
}
}
运行结果如下：
RoutingSendDirect Sendinfo':'RoutingSendDirect Send the message level:info
RoutingSendDirect Sendwarning':'RoutingSendDirect Send the message level:warning
RoutingSendDirect Senderror':'RoutingSendDirect Send the message level:error ReceiveDirect1 exchange:direct_test, queue:amq.gen-HsUrzbjzto-K7HeigXSEfQ, BindRoutingKey:info
ReceiveDirect1 exchange:direct_test, queue:amq.gen-HsUrzbjzto-K7HeigXSEfQ, BindRoutingKey:warning
ReceiveDirect1 Waiting for messages
ReceiveDirect1 Received 'info':'RoutingSendDirect Send the message level:info'
ReceiveDirect1 Received 'warning':'RoutingSendDirect Send the message level:warning' ReceiveDirect2 exchange:direct_test, queue:amq.gen-i3NY12l3DqWjGapaBOCdwQ, BindRoutingKey:error
ReceiveDirect2 Waiting for messages
ReceiveDirect2 Received 'error':'RoutingSendDirect Send the message level:error'
fanout
fanout和别的ＭＱ的发布/订阅模式类似，实例如下：
生产者
public class Pub {
private static final String EXCHANGE_NAME = "logs";
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory=new ConnectionFactory();
factory.setHost("localhost");
Connection connection=factory.newConnection();
Channel channel=connection.createChannel();
//fanout表示分发，所有的消费者得到同样的队列信息
channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
//分发信息
for (int i=;i<;i++){
String message="Hello World"+i;
channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes());
System.out.println("Pub Sent '" + message + "'");
}
channel.close();
connection.close();
}
}
消费者
public class Sub {
private static final String EXCHANGE_NAME = "logs";
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
//产生一个随机的队列名称
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, EXCHANGE_NAME, "");//对队列进行绑定
System.out.println("Sub Waiting for messages");
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("Sub Received '" + message + "'");
}
};
channel.basicConsume(queueName, true, consumer);//队列会自动删除
}
}
Topics
这种应该属于模糊匹配，* ：可以替代一个词，#：可以替代0或者更多的词，现在我们继续看看代码来理解
生产者
public class TopicSend {
private static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = null;
Channel channel = null;
try{
ConnectionFactory factory=new ConnectionFactory();
factory.setHost("localhost");
connection=factory.newConnection();
channel=connection.createChannel();
//声明一个匹配模式的交换机
channel.exchangeDeclare(EXCHANGE_NAME,"topic");
//待发送的消息
String[] routingKeys=new String[]{
"quick.orange.rabbit",
"lazy.orange.elephant",
"quick.orange.fox",
"lazy.brown.fox",
"quick.brown.fox",
"quick.orange.male.rabbit",
"lazy.orange.male.rabbit"
};
//发送消息
for(String severity :routingKeys){
String message = "From "+severity+" routingKey' s message!";
channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
System.out.println("TopicSend Sent '" + severity + "':'" + message + "'");
}
}catch (Exception e){
e.printStackTrace();
if (connection!=null){
channel.close();
connection.close();
}
}finally {
if (connection!=null){
channel.close();
connection.close();
}
}
}
}
消费者
public class ReceiveLogsTopic1 {
private static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//声明一个匹配模式的交换机
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
String queueName = channel.queueDeclare().getQueue();
//路由关键字
String[] routingKeys = new String[]{"*.orange.*"};
//绑定路由
for (String routingKey : routingKeys) {
channel.queueBind(queueName, EXCHANGE_NAME, routingKey);
System.out.println("ReceiveLogsTopic1 exchange:" + EXCHANGE_NAME + ", queue:" + queueName + ", BindRoutingKey:" + routingKey);
}
System.out.println("ReceiveLogsTopic1 Waiting for messages");
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("ReceiveLogsTopic1 Received '" + envelope.getRoutingKey() + "':'" + message + "'");
}
};
channel.basicConsume(queueName, true, consumer);
}
}
public class ReceiveLogsTopic2 {
private static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] argv) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 声明一个匹配模式的交换器
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
String queueName = channel.queueDeclare().getQueue();
// 路由关键字
String[] routingKeys = new String[]{"*.*.rabbit", "lazy.#"};
// 绑定路由关键字
for (String bindingKey : routingKeys) {
channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
System.out.println("ReceiveLogsTopic2 exchange:"+EXCHANGE_NAME+", queue:"+queueName+", BindRoutingKey:" + bindingKey);
}
System.out.println("ReceiveLogsTopic2 Waiting for messages");
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws UnsupportedEncodingException {
String message = new String(body, "UTF-8");
System.out.println("ReceiveLogsTopic2 Received '" + envelope.getRoutingKey() + "':'" + message + "'");
}
};
channel.basicConsume(queueName, true, consumer);
}
}