rabbitmq exchange

主要介绍rabbitmq常用的三种exchange模式：direct, fanout, topic.

Virtual Host

为了在一个单独的代理上实现多个隔离的环境（用户、用户组、交换机、队列 等），AMQP提供了一个虚拟主机（virtual hosts - vhosts）的概念,默认的virtual host为 “/“.

Default exchange (默认交换机)

默认交换机（default exchange）实际上是一个由消息代理预先声明好的没有名字（名字为空字符串）的直连交换机（direct exchange）。它有一个特殊的属性使得它对于简单应用特别有用处：那就是每个新建队列（queue）都会自动绑定到默认交换机上，绑定的路由键（routing key）名称与队列名称相同。

Direct Exchange

任何发送到Direct Exchange的消息都会被转发到RouteKey中指定的Queue。

• 一般情况可以使用rabbitMQ自带的Exchange：””(该Exchange的名字为空字符串，下文称其为default Exchange)。

• 这种模式下不需要将Exchange进行任何绑定(binding)操作

• 消息传递时需要一个“RouteKey”，可以简单的理解为要发送到的队列名字。

• 如果vhost中不存在RouteKey中指定的队列名，则该消息会被抛弃。

1
2
3
4
5
6
7
8

Channel channel = connection.createChannel();  
channel.exchangeDeclare("exchangeName", "direct"); //direct fanout topic  
channel.queueDeclare("queueName");  
channel.queueBind("queueName", "exchangeName", "routingKey");  
  
byte[] messageBodyBytes = "hello world".getBytes();  
//需要绑定路由键  
channel.basicPublish("exchangeName", "routingKey", MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes);

Fanout Exchange

任何发送到Fanout Exchange的消息都会被转发到与该Exchange绑定(Binding)的所有Queue上。

• 可以理解为路由表的模式

• 这种模式不需要RouteKey

• 这种模式需要提前将Exchange与Queue进行绑定，一个Exchange可以绑定多个Queue，一个Queue可以同多个Exchange进行绑定。

• 如果接受到消息的Exchange没有与任何Queue绑定，则消息会被抛弃。

1
2
3
4
5
6
7
8
9
10
11

Channel channel = connection.createChannel();  
channel.exchangeDeclare("exchangeName", "fanout"); //direct fanout topic  
channel.queueDeclare("queueName");  
channel.queueBind("queueName", "exchangeName", "routingKey");  
  
channel.queueDeclare("queueName1");  
channel.queueBind("queueName1", "exchangeName", "routingKey1");  
  
byte[] messageBodyBytes = "hello world".getBytes();  
//路由键需要设置为空  
channel.basicPublish("exchangeName", "", MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes);

Topic Exchange

任何发送到Topic Exchange的消息都会被转发到所有关心RouteKey中指定话题的Queue上.

• 这种模式较为复杂，简单来说，就是每个队列都有其关心的主题，所有的消息都带有一个“标题”(RouteKey)，Exchange会将消息转发到所有关注主题能与RouteKey模糊匹配的队列。

• 这种模式需要RouteKey，也许要提前绑定Exchange与Queue。

• 在进行绑定时，要提供一个该队列关心的主题，如“#.log.#”表示该队列关心所有涉及log的消息(一个RouteKey为”MQ.log.error”的消息会被转发到该队列)。

• “#”表示0个或若干个关键字，“”表示一个关键字。如“log.”能与“log.warn”匹配，无法与“log.warn.timeout”匹配；但是“log.#”能与上述两者匹配。

• 同样，如果Exchange没有发现能够与RouteKey匹配的Queue，则会抛弃此消息。

1
2
3
4
5
6
7

Channel channel = connection.createChannel();  
channel.exchangeDeclare("exchangeName", "topic"); //direct fanout topic  
channel.queueDeclare("queueName");  
channel.queueBind("queueName", "exchangeName", "routingKey.*");  
  
byte[] messageBodyBytes = "hello world".getBytes();  
channel.basicPublish("exchangeName", "routingKey.one", MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes);

概述

需要对Post请求的表单数据进行简单校验。

Maven 配置

• Bean Validation API 1.1:

1
2
3
4
5

<dependency>
    <groupId>javax.validation</groupId>
    <artifactId>validation-api</artifactId>
    <version>1.1.0.Final</version>
</dependency>

• Hibernate Validator 5.0.1.Final:

1
2
3
4
5

<dependency>
    <groupId>org.hibernate</groupId>
    <artifactId>hibernate-validator</artifactId>
    <version>5.0.1.Final</version>
 </dependency>

Spring MVC

配置

在application-context.xml文件中加入下面一行：

1

<mvc:annotation-driven />

controller

1. 在需要校验的model对象前面加上@Valid注解(javax.validation.Valid):

1
2
3
4
5
6

@RequestMapping(
      value = "/v1/live/send/gift", method = RequestMethod.POST)
  @Response
public WebResult requestSendGift(HttpServletRequest request, @Valid @ModelAttribute SendGiftForm sendGiftForm, BindingResult bindingResult) {
   }

1. 在controller方法里面加上参数判断：

1
2
3

if (bindingResult.hasErrors()) {
        throw new IllegalArgumentException("request params must be not null.");
  }

Validator Model

在需要校验的model字段前面加上注解(hibernate validator):

1
2
3
4
5
6
7

public class SendGiftForm {
  @NotNull
  private Long roomId; 
  @NotNull
  private Long liveId; 
  
 }

参考链接：

Spring Form Validation

概述

rabbitmq 是目前使用最为普及的消息队列组件，基于 AMQP 的 rabbitmq 在各方面设计都比较完善，同时，它具有非常丰富的功能与特性，可以支持各种实际的适用场景。

但是rabbitmq并不直接支持延时队列的功能，本文我们就来介绍一下，如何使用 rabbitmq 实现一个延时队列。

延时队列的简易实现

使用redis集群来实现了这个功能，redis中存储了下单时间，以分钟为粒度扫描相应的key，即可扫出所有下单时间超过指定时间间隔的数据.

rabbitmq 与消息过期时间 – TTL

为队列设置消息过期时间

rabbitmq 支持在创建队列时对队列设置消息过期时间：

1
2
3

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-message-ttl", 60000);
channel.queueDeclare("myqueue", false, false, false, args);

失效消息转发队列 – DLX

一旦上述消息过期时间设置生效，某条消息达到消息过期时间，那么他将会成为一条“dead-lettered”，此外，被拒绝的消息如果requeue属性为 false，或者消息所在队列已达到最大长度，那么他也将成为“dead-lettered”.

如果设置了DLX规则，即失效消息转发规则，那么失效的消息就会被转发到相应的exchange和queue.

通过代码设置失效消息转发队列

1
2
3
4
5

channel.exchangeDeclare("some.exchange.name", "direct");
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-dead-letter-exchange", "some.exchange.name");
channel.queueDeclare("myqueue", false, false, false, args);

这样，一旦消息失效，则消息会被自动转发到你设置的x-dead-letter-exchange上的同名队列.

也可以通过下面的代码指定具体转发的目标 routing-key：

1

args.put("x-dead-letter-routing-key", "some-routing-key");

spring-rabbit 配置

可以使用spring-rabbit配置替代代码方式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

<rabbit:queue name="q.with.dlx">
    <rabbit:queue-arguments> 
        <entry key="x-dead-letter-exchange" value="dlx"/>
        <entry key="x-message-ttl" value="10000" value-type="java.lang.Long"/>
    </rabbit:queue-arguments>
</rabbit:queue>
<rabbit:queue name="dlq"/>
<rabbit:direct-exchange name="dlx">
    <rabbit:bindings>
        <rabbit:binding key="q.with.dlx" queue="dlq"/>
    </rabbit:bindings>
</rabbit:direct-exchange>

This assumes you routed the original message using the default direct exchange (routing by queue name). Hence the dead letter routing uses the same routing key (queue name). If you route using an explicit routing key, you would use that.

参考链接

RabbitMQ TTL

RabbitMQ DLX

Spring AMQP

深入理解Java内存模型（一）

原文发表于InfoQ：http://www.infoq.com/cn/articles/java-memory-model-1

并发编程模型的分类

在并发编程中，我们需要处理两个关键问题：线程之间如何通信及线程之间如何同步（这里的线程是指并发执行的活动实体）。通信是指线程之间以何种机制来交换信息。在命令式编程中，线程之间的通信机制有两种：共享内存和消息传递。
在共享内存的并发模型里，线程之间共享程序的公共状态，线程之间通过写-读内存中的公共状态来隐式进行通信。在消息传递的并发模型里，线程之间没有公共状态，线程之间必须通过明确的发送消息来显式进行通信。
同步是指程序用于控制不同线程之间操作发生相对顺序的机制。在共享内存并发模型里，同步是显式进行的。程序员必须显式指定某个方法或某段代码需要在线程之间互斥执行。在消息传递的并发模型里，由于消息的发送必须在消息的接收之前，因此同步是隐式进行的。
Java的并发采用的是共享内存模型，Java线程之间的通信总是隐式进行，整个通信过程对程序员完全透明。如果编写多线程程序的Java程序员不理解隐式进行的线程之间通信的工作机制，很可能会遇到各种奇怪的内存可见性问题。

Java内存模型的抽象

在java中，所有实例域、静态域和数组元素存储在堆内存中，堆内存在线程之间共享（本文使用“共享变量”这个术语代指实例域，静态域和数组元素）。局部变量（Local variables），方法定义参数（java语言规范称之为formal method parameters）和异常处理器参数（exception handler parameters）不会在线程之间共享，它们不会有内存可见性问题，也不受内存模型的影响。
Java线程之间的通信由Java内存模型（本文简称为JMM）控制，JMM决定一个线程对共享变量的写入何时对另一个线程可见。从抽象的角度来看，JMM定义了线程和主内存之间的抽象关系：线程之间的共享变量存储在主内存（main memory）中，每个线程都有一个私有的本地内存（local memory），本地内存中存储了该线程以读/写共享变量的副本。本地内存是JMM的一个抽象概念，并不真实存在。它涵盖了缓存，写缓冲区，寄存器以及其他的硬件和编译器优化。Java内存模型的抽象示意图如下：

从上图来看，线程A与线程B之间如要通信的话，必须要经历下面2个步骤：

1. 首先，线程A把本地内存A中更新过的共享变量刷新到主内存中去。
2. 然后，线程B到主内存中去读取线程A之前已更新过的共享变量。

下面通过示意图来说明这两个步骤：

如上图所示，本地内存A和B有主内存中共享变量x的副本。假设初始时，这三个内存中的x值都为0。线程A在执行时，把更新后的x值（假设值为1）临时存放在自己的本地内存A中。当线程A和线程B需要通信时，线程A首先会把自己本地内存中修改后的x值刷新到主内存中，此时主内存中的x值变为了1。随后，线程B到主内存中去读取线程A更新后的x值，此时线程B的本地内存的x值也变为了1。
从整体来看，这两个步骤实质上是线程A在向线程B发送消息，而且这个通信过程必须要经过主内存。JMM通过控制主内存与每个线程的本地内存之间的交互，来为java程序员提供内存可见性保证。

Java 类加载器

类加载器基本概念

类加载器（class loader）用来加载 Java 类到 Java 虚拟机中。一般来说，Java 虚拟机使用 Java 类的方式如下：Java 源程序（.java 文件）在经过 Java 编译器编译之后就被转换成 Java 字节代码（.class 文件）。类加载器负责读取 Java 字节代码，并转换成 java.lang.Class类的一个实例。每个这样的实例用来表示一个 Java 类。通过此实例的 newInstance()方法就可以创建出该类的一个对象。实际的情况可能更加复杂，比如 Java 字节代码可能是通过工具动态生成的，也可能是通过网络下载的。

java.lang.ClassLoader类介绍

java.lang.ClassLoader类的基本职责就是根据一个指定的类的名称，找到或者生成其对应的字节代码，然后从这些字节代码中定义出一个 Java 类，即 java.lang.Class类的一个实例。除此之外，ClassLoader还负责加载 Java 应用所需的资源，如图像文件和配置文件等。为了完成加载类的这个职责，ClassLoader提供了一系列的方法，比较重要的方法如表 1所示。

表 1. ClassLoader 中与加载类相关的方法

类加载器的树状组织结构
Java 中的类加载器大致可以分成两类，一类是系统提供的，另外一类则是由 Java 应用开发人员编写的。系统提供的类加载器主要有下面三个：

• 引导类加载器（bootstrap class loader）：它用来加载 Java 的核心库，是用原生代码来实现的，并不继承自 java.lang.ClassLoader。

• 扩展类加载器（extensions class loader）：它用来加载 Java 的扩展库。Java 虚拟机的实现会提供一个扩展库目录。该类加载器在此目录里面查找并加载 Java 类。

• 系统类加载器（system class loader）：它根据 Java 应用的类路径（CLASSPATH）来加载 Java 类。一般来说，Java 应用的类都是由它来完成加载的。可以通过 ClassLoader.getSystemClassLoader()来获取它。

除了系统提供的类加载器以外，开发人员可以通过继承 java.lang.ClassLoader类的方式实现自己的类加载器，以满足一些特殊的需求(如tomcat)。
除了引导类加载器之外，所有的类加载器都有一个父类加载器。通过 表 1中给出的 getParent()方法可以得到。对于系统提供的类加载器来说，系统类加载器的父类加载器是扩展类加载器，而扩展类加载器的父类加载器是引导类加载器；对于开发人员编写的类加载器来说，其父类加载器是加载此类加载器 Java 类的类加载器。因为类加载器 Java 类如同其它的 Java 类一样，也是要由类加载器来加载的。一般来说，开发人员编写的类加载器的父类加载器是系统类加载器。类加载器通过这种方式组织起来，形成树状结构。树的根节点就是引导类加载器。图 1中给出了一个典型的类加载器树状组织结构示意图，其中的箭头指向的是父类加载器。