RabbitMQ routing模式

本文主要是介绍RabbitMQ routing模式，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

首先理解一下routing模式

原理同路由器，根据不同的rountKey可以指定发送到某一“IP”地址的接收者，例如发送者指定发送给IP为192.168.1.1的接收者。但是允许存在不同的接收者之间公用同一IP，即不同接收者允许有多个IP地址。

也可以多个queue绑定为同一routKey

还是先上代码

package com.tencent.routing;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;public class EmitLogDirect {private static final String EXCHANGE_NAME = "direct_logs";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, "direct");String severity = getSeverity(argv);String message = getMessage(argv);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)//如果数据不够 则默认发送hello Worldreturn "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();}
}

     来看一下发送端producing端的代码，开始new了ConnectionFactory并open了相应的Connection和Channel.

     关键代码出现了declare了一个Exchange 

  channel.exchangeDeclare(EXCHANGE_NAME, "direct");

    exchange有direct, topic, headers 和 fanout这4种模式,fanout是在Publish/Subscribe模式下使用，即所有的queue都能接受到exchange分发的massage，

现在要讨论的routing模式，则采用的是direct(直发)模式。

  然后channel在publish message的时候

 channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());

  这里的severity对应的是下图中的'orange' ‘black’和'green' 也就是routKey

   下图中有2个queue  Q1和Q2 绑定(binding)到了exchange X上面

     也就是说exchange决定了分发到哪个绑定了的queue

package com.tencent.routing;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;public class ReceiveLogsDirect {private static final String EXCHANGE_NAME = "direct_logs";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, "direct");String queueName = channel.queueDeclare().getQueue();if (argv.length < 1){System.err.println("Usage: ReceiveLogsDirect [info] [warning] [error]");System.exit(1);}for(String severity : argv){    channel.queueBind(queueName, EXCHANGE_NAME, severity);//argv是用户的输入，指定这个Receive什么severity标签的数据//argv指定error 该ReceivelLogDirect就只能接受binding到了error的Message}System.out.println(" [*] Waiting for messages. To exit press CTRL+C");QueueingConsumer consumer = new QueueingConsumer(channel);channel.basicConsume(queueName, true, consumer);while (true) {QueueingConsumer.Delivery delivery = consumer.nextDelivery();String message = new String(delivery.getBody());String routingKey = delivery.getEnvelope().getRoutingKey();System.out.println(" [x] Received '" + routingKey + "':'" + message + "'");   }}
}

   在consumer端，会为每一个severity创建一个新的绑定，也就是

   for(String severity : argv)

       channel.queueBind(queueName, EXCHAGNE_NAME ，severity)

  参考 http://www.rabbitmq.com/tutorials/tutorial-four-java.html

这篇关于RabbitMQ routing模式的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---