rabbitmq入门-路由

创建过绑定（bindings），代码如下：

 

channel.queueBind(queueName, EXCHANGE_NAME, "");

绑定（binding）是指交换器（exchange）和队列（queue）的关系。可以简单理解为：这个队列（queue）对这个交换器（exchange）的消息感兴趣。

 

 

绑定的时候可以带上一个额外的routing_key参数。为了避免与basic_publish的参数混淆，我们把它叫做binding key。以下是如何创建一个带binding key的绑定。

 

channel.queueBind(queueName, EXCHANGE_NAME, "error")

 binding key的含义取决于交换器（exchange）的类型。

 

 

Direct类型的交换器（exchange）

使用的fanout类型的交换器（exchange）扩展性不够——它能做的仅仅是广播。

 

使用direct类型的交换器（exchange）来代替。路由的算法很简单——交换器将会对binding key和routing key进行精确匹配，从而确定消息该分发到哪个队列。

 

多个绑定（Multiple bindings）

多个队列使用相同的binding key是合法的，即一条消息能发送到多个queue

 

发送消息到一个direct exchange，把日志级别作为routing key

channel.exchangeDeclare(EXCHANGE_NAME, "direct");

 发送一条消息

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

 

订阅（Subscribing）

处理接收消息的方式和之前差不多，但是我们为每一个日志级别创建了一个新的绑定

String queueName = channel.queueDeclare().getQueue();

for (String severity : argv) {
	channel.queueBind(queueName, EXCHANGE_NAME, severity);
}

 

代码整合

生产者

package com.duowan.rabbit.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class EmitLogDirect {

	private static final String EXCHANGE_NAME = "direct_logs";

    public static void main(String[] argv)
                  throws java.io.IOException {

        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 getMessage(String[] strings){
        if (strings.length < 1)
            return "Hello World!";
        return joinStrings(strings, " ");
    }
    
    private static String joinStrings(String[] strings, String delimiter) {
        int length = strings.length;
        if (length == 0) return "";
        StringBuilder words = new StringBuilder(strings[0]);
        for (int i = 1; i < length; i++) {
            words.append(delimiter).append(strings[i]);
        }
        return words.toString();
    }
    
    private static String getSeverity(String[] strings){
        if (strings.length < 1)
                    return "info";
        return strings[0];
      }
}

 消费者

package com.duowan.rabbit.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
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);
		}

		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
					+ "'");
		}
	}

}