文档章节

轻松搞定RabbitMQ(五)——路由选择

白志华
 白志华
发布于 2015/10/18 10:56
字数 1376
阅读 47
收藏 1
点赞 0
评论 0

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

       在前篇博文中,我们建立了一个简单的日志系统。可以广播消息给多个消费者。本篇博文,我们将添加新的特性——我们可以只订阅部分消息。比如:我们可以接收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)

public class EmitLogDirect {
	private final static String EXCHANGE_NAME = "direct_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, "direct");

		//所有日志严重性级别
		String[] severities={"error","info","warning"};
		for(int i=0;i<3;i++){
			String severity = severities[i%3];//每一次发送一条不同严重性的日志
			
			// 发送的消息
			String message = "Hello World"+Strings.repeat(".", i+1);
			//参数1:exchange name
			//参数2:routing key
			channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
			System.out.println(" [x] Sent '" + severity +"':'"+ message + "'");
		}
		// 关闭频道和连接
		channel.close();
		connection.close();
	}
}
消费者1(ReceiveLogs2Console.java)

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("127.0.0.1");
		// 打开连接和创建频道,与发送端一样
		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) {
			  @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 = "direct_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, "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();
		}
	}  
}
       最终结果:


       罗哩罗嗦的说这么多,其实就是说了这么一件事:我们可以使用Direct exchange+routingKey来过滤自己感兴趣的消息。一个队列可以绑定多个routingKey。这就是我们今天的主题——路由选择。

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

共有 人打赏支持
白志华
粉丝 29
博文 260
码字总数 57524
作品 0
长沙
程序员
消息中间件—RabbitMQ(初探篇)

文章摘要:本篇文章为RabbitMQ的入门文章,不像其他一些程序代码和应用实战性的文章会带着大家从一个“Hello World”的简单例子出发,在该篇幅中主要给大家讲下RabbitMQ消息队列的起源、为何...

癫狂侠 ⋅ 05/23 ⋅ 0

SpringBoot整合高级消息队列RabbitMQ及原理

一 SpringBoot 与消息概述 xhesrc="https://img-blog.csdn.net/20180619001205459?watermark/2/text/aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4OTc0NjM0/font/5a6L5L2T/fontsize/400/fill/I0JBQk......

刘信坚 ⋅ 06/19 ⋅ 0

消息中间件—RabbitMQ(集群原理与搭建篇)

摘要:实际生产应用中都会采用消息队列的集群方案,如果选择RabbitMQ那么有必要了解下它的集群方案原理 一般来说,如果只是为了学习RabbitMQ或者验证业务工程的正确性那么在本地环境或者测试...

癫狂侠 ⋅ 05/25 ⋅ 0

SpringBoot笔记(十一)RabbitMQ

安装Erlang RabbitMQ基于Erlang,所以得先安装Erlang http://www.erlang.org/downloads 根据自己的系统选择下载,安装完了,配置一下path即可 windows默认安装路径: 验证: 有时候可能需要重...

世外大帝 ⋅ 04/25 ⋅ 0

介绍Spring Cloud Stream与RabbitMQ集成

Spring Cloud Stream是一个建立在Spring Boot和Spring Integration之上的框架,有助于创建事件驱动或消息驱动的微服务。在本文中,我们将通过一些简单的例子来介绍Spring Cloud Stream的概念...

RaiseHead ⋅ 05/13 ⋅ 0

ActiveMQ RabbitMQ KafKa对比

前言: ActiveMQ和 RabbitMq 以及Kafka在之前的项目中都有陆续使用过,当然对于三者没有进行过具体的对比,以下摘抄了一些网上关于这三者的对比情况,我自己看过之后感觉还 是可以的,比较清...

xiaomin0322 ⋅ 05/11 ⋅ 0

Spring Boot+RabbitMQ学习笔记

RabbitMQ RabbitMQ 是一个由 Erlang 语言开发的 的开源实现。最初起源于金融系统,用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗。 消息队列 先从基础开始,加...

听_风 ⋅ 06/07 ⋅ 0

RabbitMQ在Ubuntu上的环境搭建

环境配置 Ubuntu Server 18.04 RabbitMQ 3.6.10 安装之前 我们使用apt-get进行RabbitMQ安装,在安装之前,强烈建议您把apt源换位国内,大大增加下载安装的速度,点击查看:《Ubuntu apt-get和...

vipstone ⋅ 06/14 ⋅ 0

RabbitMQ消息中间件介绍

一、基础介绍 随着分布式应用的发展消息队列中间件成为C/S架构中解耦的一个重要环节,传统的消息传输模型中,C端发出消息,S端必须在线,否则将无法继续进行,而在拥有消息中间件的模型下消息...

qiao645 ⋅ 昨天 ⋅ 0

RabbitMQ系列(一):Windows下RabbitMQ安装及入门

1.Windows下安装RabbitMQ需要以下几个步骤 (1):下载erlang,原因在于RabbitMQ服务端代码是使用并发式语言erlang编写的,下载地址:http://www.erlang.org/downloads,双击.exe文件进行安装就...

xiaomin0322 ⋅ 05/09 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

MySQL 数据库设计总结

规则1:一般情况可以选择MyISAM存储引擎,如果需要事务支持必须使用InnoDB存储引擎。 注意:MyISAM存储引擎 B-tree索引有一个很大的限制:参与一个索引的所有字段的长度之和不能超过1000字节...

OSC_cnhwTY ⋅ 今天 ⋅ 0

多线程(四)

线程池和Exector框架 什么是线程池? 降低资源的消耗 提高响应速度,任务:T1创建线程时间,T2任务执行时间,T3线程销毁时间,线程池没有或者减少T1和T3 提高线程的可管理性。 线程池要做些什...

这很耳东先生 ⋅ 今天 ⋅ 0

使用SpringMVC的@Validated注解验证

1、SpringMVC验证@Validated的使用 第一步:编写国际化消息资源文件 编写国际化消息资源ValidatedMessage.properties文件主要是用来显示错误的消息定制 [java] view plain copy edit.userna...

瑟青豆 ⋅ 今天 ⋅ 0

19.压缩工具gzip bzip2 xz

6月22日任务 6.1 压缩打包介绍 6.2 gzip压缩工具 6.3 bzip2压缩工具 6.4 xz压缩工具 6.1 压缩打包介绍: linux中常见的一些压缩文件 .zip .gz .bz2 .xz .tar .gz .tar .bz2 .tar.xz 建立一些文...

王鑫linux ⋅ 今天 ⋅ 0

6. Shell 函数 和 定向输出

Shell 常用函数 简洁:目前没怎么在Shell 脚本中使用过函数,哈哈,不过,以后可能会用。就像java8的函数式编程,以后获取会用吧,行吧,那咱们简单的看一下具体的使用 Shell函数格式 linux ...

AHUSKY ⋅ 今天 ⋅ 0

单片机软件定时器

之前写了一个软件定时器,发现不够优化,和友好,现在重写了 soft_timer.h #ifndef _SOFT_TIMER_H_#define _SOFT_TIMER_H_#include "sys.h"typedef void (*timer_callback_function)(vo...

猎人嘻嘻哈哈的 ⋅ 今天 ⋅ 0

好的资料搜说引擎

鸠摩搜书 简介:鸠摩搜书是一个电子书搜索引擎。它汇集了多个网盘和电子书平台的资源,真所谓大而全。而且它还支持筛选txt,pdf,mobi,epub、azw3格式文件。还显示来自不同网站的资源。对了,...

乔三爷 ⋅ 今天 ⋅ 0

Debian下安装PostgreSQL的表分区插件pg_pathman

先安装基础的编译环境 apt-get install build-essential libssl1.0-dev libkrb5-dev 将pg的bin目录加入环境变量,主要是要使用 pg_config export PATH=$PATH:/usr/lib/postgresql/10/bin 进......

玛雅牛 ⋅ 今天 ⋅ 0

inno安装

#define MyAppName "HoldChipEngin" #define MyAppVersion "1.0" #define MyAppPublisher "Hold Chip, Inc." #define MyAppURL "http://www.holdchip.com/" #define MyAppExeName "HoldChipE......

backtrackx ⋅ 今天 ⋅ 0

Linux(CentOS)下配置php运行环境及nginx解析php

【part1:搭建php环境】 1.选在自己需要安装的安装包版本,wget命令下载到服务器响应目录 http://php.net/releases/ 2.解压安装包 tar zxf php-x.x.x 3.cd到解压目录执行如下操作 cd ../php-...

硅谷课堂 ⋅ 今天 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部