Java

消息中间件(AMQP)

略

对比AMQP和JMS可以发现,AMQP有很多的优势:

  1. AMQP为消息定义了线路层(wire-level protocll)的协议,而JMS使用了API规范;
  2. JMS规定了它的发送的消息被相同的JMS所使用,而AMQP可以给不同的AMQP来实现;
  3. AMQP甚至可以跨语言和平台,只要遵循了它的规范,都可以使用;

AMQP简介

JMS使用有三个角色:生产者,消费者,传递消息的通道.生产者和消费者解耦了,但是他们俩和通道依然存在耦合,这样一来通道就具有双重责任,既要传递消息,又要确保消息传递的目的地和方式.

AMQP与其不同,它的生产者不会直接把消息放到队列中.他在生产者和通道中加入了一个间接的机制:Exchange.

生产者会发消息给Exchange.Exchange会绑定到一个或多个队列上,它负责把消息路由到队列上,然后消费者从队列中取出数据并处理.

Exchange不仅仅是一个简单的中介,它也会有不同的处理路由的方法,AMQP定义了4个不同类型的Exchange,每个有不同的路由算法.根据算法不同,消息会被筛选的方式选择队列然后进入队列.四个标准的Exchange分别为:

  1. Direct: 如果消息的 routing key 与 binding 的 routing key 直接匹配,就会路由到这个队列上;
  2. Topic: 如果消息的 routing key 与 binding 的 routing key 符合通配符匹配的话,就会路由到这个队列上;
  3. Headers: 如果消息参数表中的头信息和值都与binding参数列表中匹配,就会路由到这个队列上;
  4. Fanout: 不管routing和头信息是啥,都会路由到所有的队列上;

根据Exchange的不同,可以设计出各种方式的路由模式,而不限于P2P(point-to-point)和发布/订阅模式,并且这些路由算法对于生产者和消费者的编写没有任何影响,唯一的注意的是:生产者发送消息给Exchange时会带一个runting key.

配置Spring支持AMQP

RabbitMQ实现了AMQP,SpringAMQP为RabbitMQ提供了支持,包括连接工厂,模板,命名空间

发送

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
       http://www.springframework.org/schema/context
       http://www.springframework.org/schema/context/spring-context.xsd">

  <rabbit:connection-factory
            id="connectionFactory"
            host="127.0.0.1"
            port="5672"
            username="guest"
            password="guest"/>

  	<rabbit:admin connection-factory="connectionFactory"/>

    <rabbit:queue id="myQueue1" name="myQueue1"/>
    <rabbit:queue id="myQueue2" name="myQueue2"/>

    <rabbit:fanout-exchange name="myFanoutExchange">
        <rabbit:bindings>
          	<rabbit:binding queue="myQueue1"/>
            <rabbit:binding queue="myQueue2"/>
        </rabbit:bindings>
    </rabbit:fanout-exchange>

    <rabbit:direct-exchange name="myDirectExchange">
        <rabbit:bindings>
          	<rabbit:binding queue="myQueue2" key="ab#"/>
            <rabbit:binding queue="myQueue1" key="#ab" exchange="myFanoutExchange"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>


    <rabbit:template id="template" connection-factory="connectionFactory"/>

</beans>

接收

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
       http://www.springframework.org/schema/context
       http://www.springframework.org/schema/context/spring-context.xsd">

    <context:property-placeholder location="classpath*:rabbitmq.properties"/>

    <rabbit:connection-factory
            id="connectionFactory"
            host="127.0.0.1"
            port="5672"
            username="guest"
            password="guest"/>

    <rabbit:queue id="myQueue1" name="myQueue1"/>

    <bean class="MyListener" id="listener"/>

    <rabbit:listener-container connection-factory="connectionFactory">
        <rabbit:listener ref="listener" method="myMethod" queues="myQueue1"/>
    </rabbit:listener-container>

</beans>

区别

RabbitMQ几乎与ActiveMQ的配置相同,下面列举一下区别:

  1. 发送端:RabbitMQ有Exchange需要配置,通过Exchange来连接每个队列,使用不同的Exchange效果不同,Exchange之间也可以互相连接在一起,可以通过Exchange来实现复杂的消息发布的方式.
  2. RabbitMQTemplate的send方法,需要传递三个参数,一是Exchange,二是routing key,三是发送的消息(注意要和消费者myMethod方法的参数一致).
  3. RabbitMQ的监听器中没有目的地属性,有一个queues属性可以配置监听的队列

其他方面RabbitMQ与ActiveMQ在Spring中的配置都一样.RabbitMQ就比ActiveMQ多了一个Exchange.