RabbitMQ学习(四)之路由（java）

转载来自:http://blog.csdn.net/zhu_tianwei/article/details/40887755

使用direct类型实现：消息会被推送至绑定键（binding key）和消息发布附带的选择键（routing key）完全匹配的队列。例如：将不同的日志发送到不同的消费端。



1.发送日志端SendLogDirect.Java

[java]
view plain
copy
print?





package cn.slimsmart.rabbitmq.demo.route;  
  
import java.util.Random;  
import java.util.UUID;  
  
import com.rabbitmq.client.AMQP;  
import com.rabbitmq.client.Channel;  
import com.rabbitmq.client.Connection;  
import com.rabbitmq.client.ConnectionFactory;  
  
//随机发送6条随机类型（routing key）的日志给转发器~~  
public class SendLogDirect {  
      
     //交换名称  
     private static final String EXCHANGE_NAME = "ex_logs_direct";    
     //日志分类  
     private static final String[] SEVERITIES = { "info", "warning", "error" };    
          
    public static void main(String[] args) throws Exception {  
        //创建连接和频道    
        ConnectionFactory factory = new ConnectionFactory();    
        factory.setHost("192.168.101.174");  
        // 指定用户 密码  
        factory.setUsername("admin");  
        factory.setPassword("admin");  
        // 指定端口  
        factory.setPort(AMQP.PROTOCOL.PORT);  
        Connection connection = factory.newConnection();    
        Channel channel = connection.createChannel();    
        // 声明转发器的类型    
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");    
    
        //发送6条消息    
        for (int i = 0; i < 6; i++)    
        {    
            String severity = getSeverity();    
            String message = severity + "_log :" + UUID.randomUUID().toString();    
            // 发布消息至转发器，指定routingkey    
            channel.basicPublish(EXCHANGE_NAME, severity, null, message  .getBytes());    
            System.out.println(" [x] Sent '" + message + "'");    
        }    
    
        channel.close();    
        connection.close();    
    }  
      
    /**  
     * 随机产生一种日志类型  
     *   
     * @return  
     */    
    private static String getSeverity()    
    {    
        Random random = new Random();    
        int ranVal = random.nextInt(3);    
        return SEVERITIES[ranVal];    
    }    
  
}  

2.接收日志端ReceiveLogsDirect.java

[java]
view plain
copy
print?





package cn.slimsmart.rabbitmq.demo.route;  
  
import java.util.Random;  
  
import com.rabbitmq.client.AMQP;  
import com.rabbitmq.client.Channel;  
import com.rabbitmq.client.Connection;  
import com.rabbitmq.client.ConnectionFactory;  
import com.rabbitmq.client.QueueingConsumer;  
  
//接收端随机设置一个日志严重级别（binding_key）。。。  
public class ReceiveLogsDirect {  
      
    private static final String EXCHANGE_NAME = "ex_logs_direct";    
    private static final String[] SEVERITIES = { "info", "warning", "error" };    
  
    public static void main(String[] args) throws Exception {  
          // 创建连接和频道    
        ConnectionFactory factory = new ConnectionFactory();    
        factory.setHost("192.168.101.174");  
        // 指定用户 密码  
        factory.setUsername("admin");  
        factory.setPassword("admin");  
        // 指定端口  
        factory.setPort(AMQP.PROTOCOL.PORT);  
        Connection connection = factory.newConnection();    
        Channel channel = connection.createChannel();    
        // 声明direct类型转发器    
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");    
    
        String queueName = channel.queueDeclare().getQueue();    
        String severity = getSeverity();    
        // 指定binding_key    
        channel.queueBind(queueName, EXCHANGE_NAME, severity);    
        System.out.println(" [*] Waiting for "+severity+" logs. 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());    
    
            System.out.println(" [x] Received '" + message + "'");    
        }    
    }    
    
    /**  
     * 随机产生一种日志类型  
     *   
     * @return  
     */    
    private static String getSeverity()    
    {    
        Random random = new Random();    
        int ranVal = random.nextInt(3);    
        return SEVERITIES[ranVal];    
    }    
}  



启动几个接收端服务，再启动发送端，接收端对应绑定的键收到对应的消息。注：发送消息时可以设置routing_key，接收队列与转发器间可以设置binding_key，接收者接收与binding_key与routing_key相同的消息。