文档章节

轻松搞定RabbitMQ(六)——主题

白志华
 白志华
发布于 2015/10/18 10:55
字数 1481
阅读 56
收藏 0

       翻译地址:http://www.rabbitmq.com/tutorials/tutorial-five-java.html

       在上一篇博文中,我们进一步改良了日志系统。使用Direct类型的转换器,使得接收者有能力进行选择性的接收日志,,而非fanout那样,只能够无脑的转发,如果你还不了解,请阅读:轻松搞定RabbitMQ(四)——发布/订阅

       虽然使用Direct类型的转换器改进了日志系统。但它仍然有一定的局限性——不能根据多重条件进行路由选择。

       在我们的日志系统中,我们可能不仅仅根据日志严重性订阅日志,也想根据发送源订阅。你可能从unix工具syslog了解过这个概念,它可以根据严重性(info/warning/crit…)和设备(auth/cron/kern…)转发日志。

       这将给我们更多的灵活性——我们可以订阅来自元‘cron’的致命错误日志,同时也可以订阅‘kern’的所有日志。

       为了在我们的日志系统中实现上述需求,我们需要了解更复杂的主题类型的转发器——Topic Exchange。


Topic exchange(主题转发器)

       发送给主题转发器的消息不能是任意设置的选择键,必须是用小数点隔开的一系列的标识符。这些标识符可以是随意,但是通常跟消息的某些特性相关联。一些合法的路由选择键比如“socket.usd.nyse”,"nyse.vmw","quick.orange.rabbit",你愿意用多少单词都可以,只要不超过上限的255个字节。

       绑定键也必须以相同的格式。主题转发器的逻辑类似于direct类型的转发器。消息通过一个特定的路由键发送到所有与绑定键匹配的队列中。需要注意的是,关于绑定键有两种特殊的情况:*(星号)可以代替任意一个标识符 ;#(井号)可以代替零个或多个标识符。

       举一个简单例子,如下图:


       在上图例子中,我们发送描述动物的消息。消息会转发给包含3个单词(2个小数点)的路由键绑定的队列中。绑定键中的第一个单词描述的是速度,第二个是颜色,第三个是物种:“速度.颜色.物种”。

       我们创建3个绑定:Q1绑定键是“*.orange.*”,Q2绑定键是“*.*.rabbit”,Q3绑定键是“lazy.#”。这些绑定可以概括为:Q1只对橙色的动物感兴趣。Q2则是关注兔子和所有懒的动物。

       路由键为“quick.orange.rabbit”的消息会被路由到2个队列中去。而“lazy.orange.elephant”的消息同样会发往2个队列。另外“quick.orange.fox” 仅仅发往第一个队列,而"lazy.brown.fox"则只发往第二个队列。“quick.brown.fox”则所有的绑定键都不匹配而被丢弃。

       如果我们违反约定,发送了只带1个或者4个标识符的选择键,像“orange”或者“quick.orange.male.rabbit”,会发生什么呢?这些消息都不匹配任何绑定,所以将被丢弃。

       另外,“lazy.orange.male.rabbit”,尽管有4个标识符,但是仍然匹配最后一个绑定键,所以会发送到第二个队列中。

       注:主题类型的转发器非常强大,可以实现其他类型的转发器。当队列绑定#绑定键,可以接受任何消息,类似于fanout转发器。当特殊字符*和#不包含在绑定键中,这个主题转发器就像一个direct类型的转发器。


完整实例

       我们将主题类型的转发器应用到日志系统中,路由格式为:“<设备>.<严重级别>”。

发送端(EmitLogTopic.java)

public class EmitLogDirect {
	private final static String EXCHANGE_NAME = "topic_logs";

	public static void main(String[] args) throws IOException {
		/**
		 * 创建连接连接到MabbitMQ
		 */
		ConnectionFactory factory = new ConnectionFactory();
		// 设置MabbitMQ所在主机ip或者主机名
		factory.setHost("127.0.0.1");
		// 创建一个连接
		Connection connection = factory.newConnection();
		// 创建一个频道
		Channel channel = connection.createChannel();
		// 指定转发——广播
		channel.exchangeDeclare(EXCHANGE_NAME, "topic");

		//所有设备和日志级别
		String[] facilities ={"auth","cron","kern","auth.A"};
		String[] severities={"error","info","warning"};
		
		for(int i=0;i<4;i++){
			for(int j=0;j<3;j++){
			//每一个设备,每种日志级别发送一条日志消息
			String routingKey = facilities[i]+"."+severities[j%3];
			
			// 发送的消息
			String message =" Hello World!"+Strings.repeat(".", i+1);
			//参数1:exchange name
			//参数2:routing key
			channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
			System.out.println(" [x] Sent [" + routingKey +"] : '"+ message + "'");
			}
		}
		// 关闭频道和连接
		channel.close();
		connection.close();
	}
}

消费者1(ReceiveLogs2Console.java)

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

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

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

		String[] routingKeys ={"auth.*","*.info","#.warning"};//关注所有的授权日志、所有info和waring级别的日志
		for (String routingKey : routingKeys) {
			//关注所有级别的日志(多重绑定)
			channel.queueBind(queueName, EXCHANGE_NAME, routingKey);
		}
		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");
			    System.out.println(" [x] Received ["  + envelope.getRoutingKey() + "] :'" + message + "'");
			  }
			};
			channel.basicConsume(queueName, true, consumer);
	}
}

消费者2(ReceiveLogs2File.java)

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

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

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

	    String severity="kern.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();
		}
	}  
}

       最终结果:





本文转载自:http://blog.csdn.net/xiaoxian8023/article/details/48806871

白志华
粉丝 34
博文 265
码字总数 57524
作品 0
长沙
程序员
私信 提问
go封装rabbitmq的 发布/订阅 解耦组件

关于rabbitmq 的入门和使用,网络上的资料甚多,在此不再赘述。可以看文章末尾的参考资料。 对rabbitmq的golang客户端二度封装,通过订阅/发布的模式使消息的获取和接收进行解耦,从而支持多...

歪文
07/25
0
0
我的RabbitMQ的学习成果

背景 在研发分布式事务的最终一致性事务模式时,使用了RabbitMQ。 在这之前也接触过RabbitMQ,但没有特别深入的去了解它的特性与原理。这次决定系统的学习一次,所以业余时间阅读大神们的书籍...

XuePeng77
04/15
257
0
你不知道的RabbitMQ集群架构全解

你不知道的RabbitMQ集群架构全解 RabbitMQ系列文章 RabbitMQ在Ubuntu上的环境搭建 深入了解RabbitMQ工作原理及简单使用 RabbitMQ交换器Exchange介绍与实践 RabbitMQ事务和Confirm发送方消息确...

王磊的博客
2018/07/25
0
0
了解消息队列中间件——RabbitMQ

了解消息队列中间件 1. 消息:指的是在应用之间传送的数据,比如json字符串、纯文本字符串等 2. 消息队列中间件:指利用高效可靠的消息传递机制进行与平台无关的数据交流,并基于数据通信来进...

江左煤郎
01/13
45
0
阿里大牛带你轻松实现RabbitMQ 延时消息

RabbitMQ 延时消息的实现(上) 我们在实际业务中有一些需要延时发送消息的场景,例如: 家里有一台智能热水器,需要在30分钟后启动 未付款的订单,15分钟后关闭 注意这里的场景是延时,不是...

Java架构
01/23
0
0

没有更多内容

加载失败,请刷新页面

加载更多

uni app 零基础小白到项目实战

$emit 子组件传给父组件$ref 父组件操作子组件 公用模板 uni-app全局变量的几种实现方法 const websiteUrl = 'http'const now = Date.now || function() { return new Date().getTime......

达达前端小酒馆
21分钟前
4
0
Tomcat是如何实现异步Servlet的

前言 通过我之前的Tomcat系列文章,相信看我博客的同学对Tomcat应该有一个比较清晰的了解了,在前几篇博客我们讨论了Tomcat在SpringBoot框架中是如何启动的,讨论了Tomcat的内部组件是如何设...

木木匠
45分钟前
18
0
mysql中间件分享(Mysql-prxoy,Atlas,DBProxy,Amoeba,cobar,TDDL)

hello 各位小伙伴大家好,我是小栈君,这期我们分享关于mysql中间件的研究,也就是数据层的读写分离和负载均衡,希望能够在实际的应用中能够帮助到各位小伙伴。 下期我们将继续分享go语言的系...

IT干货栈
今天
10
0
OSChina 周一乱弹 —— 人生,还真是到处是意外

Osc乱弹歌单(2019)请戳(这里) 【今日歌曲】 @这次装个文艺青年吧 :#今日歌曲推荐# 分享lil peep的单曲《High School》 《High School》- lil peep 手机党少年们想听歌,请使劲儿戳(这里...

小小编辑
今天
749
12
Spring使用ThreadPoolTaskExecutor自定义线程池及实现异步调用

多线程一直是工作或面试过程中的高频知识点,今天给大家分享一下使用 ThreadPoolTaskExecutor 来自定义线程池和实现异步调用多线程。 一、ThreadPoolTaskExecutor 本文采用 Executors 的工厂...

CREATE_17
今天
12
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部