时间戳

首页 > 技术文章 > RabbitMQ/JAVA （路由选择）

RabbitMQ/JAVA （路由选择）

tinmh 2016-12-06 14:11 原文

上篇博文中，我们建立了一个简单的日志系统。可以广播消息给多个消费者。本篇博文，我们将添加新的特性——我们可以只订阅部分消息。比如：我们可以接收Error级别的消息写入文件。同时仍然可以在控制台打印所有日志。

Bindings（绑定）

 在上一篇博客中我们已经使用过绑定。类似下面的代码：

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

绑定表示转换器与队列之间的关系。可以简单的认为：队列对该转发器上的消息感兴趣。

绑定可以设定额外的routingKey参数。为了与避免basicPublish方法（发布消息的方法）的参数混淆，我们准备把它称作绑定键（binding key）。下面展示如何使用绑定键（binding key）来创建一个绑定：

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

绑定键关键取决于转换器的类型。对于fanout类型，忽略此参数。

Direct exchange（直接转发）

前面讲到我们的日志系统广播消息给所有的消费者。我们想对其扩展，根据消息的严重性来过滤消息。例如：我们希望将致命错误的日志消息记录到文件，而不是把磁盘空间浪费在warn和info类型的日志上。我们使用的fanout转发器，不能给我们太多的灵活性。它仅仅只是盲目的广播而已。我们使用direct转发器进行代替，其背后的算法很简单——消息会被推送至绑定键（binding key）和消息发布附带的选择键（routing key）完全匹配的队列。

在上图中，我们可以看到direct类型的转发器与2个队列进行了绑定。第一个队列使用的绑定键是orange，第二个队列绑定键为black和green。这样当消息发布到转发器是，附带orange绑定键的消息将被路由到队列Q1中去。附带black和green绑定键的消息被路由到Q2中去。其他消息全部丢弃。

Multiple bindings（多重绑定）

使用一个绑定键绑定多个队列是完全合法的。如上图，绑定键black绑定了2个队列——Q1和Q2。

 

Emitting logs(发送日志)

我们将这种模式用于日志系统，发送消息给direct类型的转发器。我们将 提供日志严重性做为绑定键。那样，接收程序可以选择性的接收严重性的消息。首先关注发送日志的代码：

像往常一样首先创建一个转换器：

channel.exchangeDeclare(EXCHANGE_NAME, "direct");

然后为发送消息做准备：

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

为了简化代码，我们假定日志的严重性是“info”、“warning”、“error”中之一。

Subscribing(订阅)

接收消息跟前面博文中的一样。我们仅需要修改一个地方：为每一个我们感兴趣的严重性的消息，创建一个新的绑定。

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

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

完整例子：

发送端代码（EmitLogDirect.java）



package ERROR;




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("115.159.181.204");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();




channel.exchangeDeclare(EXCHANGE_NAME, "direct");




 String[] severities = {"error","info","warning"};
        for(int i=0 ;i<3;i++){
        	String severity =severities[i%3];
        	String message="Hellp World!";




 channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
            System.out.println(" [x] Sent '" + severity + "':'" + message + "'");
        };





        channel.close();
        connection.close();
    }
    //..
}


 




消费者1（ReceiveLogs2Console.java）




package ERROR;


import java.io.IOException;


import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;



public class ReceiveLogs2Console {
	private static final String EXCHANGE_NAME = "direct_logs";


	public static void main(String[] argv) throws IOException, InterruptedException {
		ConnectionFactory factory = new ConnectionFactory();
		 factory.setHost("115.159.181.204");
	        factory.setPort(5672);
	        factory.setUsername("admin");
	        factory.setPassword("admin");
		// 打开连接和创建频道，与发送端一样
		Connection connection = factory.newConnection();
		final Channel channel = connection.createChannel();


		channel.exchangeDeclare(EXCHANGE_NAME, "direct");
		// 声明一个随机队列
		String queueName = channel.queueDeclare().getQueue();


		//所有日志严重性级别
		String[] severities={"error","info","warning"};
		for (String severity : severities) {
			//关注所有级别的日志（多重绑定）
			channel.queueBind(queueName, EXCHANGE_NAME, severity);
		}
		System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
		
		// 创建队列消费者
		final Consumer consumer = new DefaultConsumer(channel) {
			  public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
			    String message = new String(body, "UTF-8");
			    System.out.println(" [x] Received '"  + envelope.getRoutingKey() + "':'" + message + "'");
			  }
			};
			channel.basicConsume(queueName, true, consumer);
	}
}




消费者2（ReceiveLogs2File.java）




package ERROR;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;


import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;


public class ReceiveLogs2File {
	private static final String EXCHANGE_NAME = "direct_logs";


	public static void main(String[] argv) throws IOException, InterruptedException {
		ConnectionFactory factory = new ConnectionFactory();
		 factory.setHost("115.159.181.204");
	        factory.setPort(5672);
	        factory.setUsername("admin");
	        factory.setPassword("admin");
		// 打开连接和创建频道，与发送端一样
		Connection connection = factory.newConnection();
		final Channel channel = connection.createChannel();


		channel.exchangeDeclare(EXCHANGE_NAME, "direct");
		// 声明一个随机队列
		String queueName = channel.queueDeclare().getQueue();
	    
	    String severity="error";//只关注error级别的日志，然后记录到文件中去。
	    channel.queueBind(queueName, EXCHANGE_NAME, severity);
	    
		System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
		
		// 创建队列消费者
		final 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");
			    //记录日志到文件：
			    print2File( "["+ envelope.getRoutingKey() + "] "+message);
			  }
			};
			channel.basicConsume(queueName, true, consumer);
	}
	
	private static void print2File(String msg) {
		try {
			String dir = ReceiveLogs2File.class.getClassLoader().getResource("").getPath();
			String logFileName = new SimpleDateFormat("yyyy-MM-dd").format(new Date());
			File file = new File(dir, logFileName + ".log");
			FileOutputStream fos = new FileOutputStream(file, true);
			fos.write((new SimpleDateFormat("HH:mm:ss").format(new Date())+" - "+msg + "\r\n").getBytes());
			fos.flush();
			fos.close();
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}  
}




运行结果：






 







	

		


推荐阅读