spring集成RabbitMQ配置文件详细解读（生产者和消费者）_java springmvc整合rabbitmq消费者topic配置文件

引入jar包

  <dependency>
      <groupId>com.rabbitmq</groupId>
      <artifactId>amqp-client</artifactId>
      <version>4.0.0</version>
    </dependency>
    
    <dependency>
      <groupId>org.springframework.amqp</groupId>
      <artifactId>spring-rabbit</artifactId>
      <version>1.7.2.RELEASE</version>
    </dependency>
    
1
2
3
4
5
6
7
8
9
10
11
12

rabbitmq属性介绍

概念解释：

Brocker：消息队列服务器实体。

Exchange：消息交换机，指定消息按什么规则，路由到哪个队列。

Queue：消息队列，每个消息都会被投入到一个或者多个队列里。

Binding：绑定，它的作用是把exchange和queue按照路由规则binding起来。

Routing Key：路由关键字，exchange根据这个关键字进行消息投递。

Virtual Host: 虚拟主机，一个broker里可以开设多个vhost，用作不用用户的权限分离。
每个virtual host本质上都是一个RabbitMQ Server（但是一个server中可以有多个virtual host），
拥有它自己若干的个Exchange、Queue和bings rule等等。其实这是一个虚拟概念，类似于权限控制组。
Virtual Host是权限控制的最小粒度。

Producer：消息生产者，就是投递消息的程序。

Consumer：消息消费者，就是接受消息的程序。

Connection: 就是一个TCP的连接。Producer和Consumer都是通过TCP连接到RabbitMQ Server的。
接下来的实践案例中我们就可以看到，producer和consumer与exchange的通信的前提是先建立TCP连接。
仅仅创建了TCP连接，producer和consumer与exchange还是不能通信的。
我们还需要为每一个Connection创建Channel。

Channel: 它是建立在上述TCP连接之上的虚拟连接。数据传输都是在Channel中进行的。AMQP协议规定只有通过
Channel才能执行AMQP的命令。一个Connection可以包含多个Channel。有人要问了，为什么要使用Channel呢，
直接用TCP连接不就好了么？对于一个消息服务器来说，它的任务是处理海量的消息，当有很多线程需要从RabbitMQ
中消费消息或者生产消息，那么必须建立很多个connection，也就是许多个TCP连接。然而对于操作系统而言，建立
和关闭TCP连接是非常昂贵的开销，而且TCP的连接数也有限制，频繁的建立关闭TCP连接对于系统的性能有很大的影
响，如果遇到高峰，性能瓶颈也随之显现。RabbitMQ采用类似NIO的做法，选择TCP连接服用，不仅可以减少性能开
销，同时也便于管理。在TCP连接中建立Channel是没有上述代价的，可以复用TCP连接。对于Producer或者
Consumer来说，可以并发的使用多个Channel进行Publish或者Receive。有实验表明，在Channel中，1秒可以
Publish10K的数据包。对于普通的Consumer或者Producer来说，这已经足够了。除非有非常大的流量时，一个
connection可能会产生性能瓶颈，此时就需要开辟多个connection。
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

逻辑过程

消息队列的使用过程大概如下：

消息接收

客户端连接到消息队列服务器，打开一个channel。
客户端声明一个exchange，并设置相关属性。
客户端声明一个queue，并设置相关属性。
客户端使用routing key，在exchange和queue之间建立好绑定关系。
1
2
3
4

消息发送

客户端投递消息到exchange。
exchange接收到消息后，就根据消息的key和已经设置的binding，进行消息路由，将消息投递到
一个或多个队列里。
1
2
3

配置springRabbitMQ.xml

<!-- 公共部分 -->
<!-- 创建连接类 连接安装好的 rabbitmq -->
<bean id="connectionFactory"  class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
    <constructor-arg value="localhost" />    
    <!-- username,访问RabbitMQ服务器的账户,默认是guest -->
    <property name="username" value="${rmq.manager.user}" />
    <!-- username,访问RabbitMQ服务器的密码,默认是guest -->   
    <property name="password" value="${rmq.manager.password}" />
    <!-- host,RabbitMQ服务器地址，默认值"localhost" -->   
    <property name="host" value="${rmq.ip}" />   
    <!-- port，RabbitMQ服务端口，默认值为5672 -->
    <property name="port" value="${rmq.port}" />
    <!-- channel-cache-size，channel的缓存数量，默认值为25 -->
    <property name="channel-cache-size" value="50" />
    <!-- cache-mode，缓存连接模式，默认值为CHANNEL(单个connection连接，连接之后关闭，自动销毁) -->
    <property name="cache-mode" value="CHANNEL" />   
</bean>
<!--或者这样配置，connection-factory元素实际就是注册一个org.springframework.amqp.rabbit.connection.CachingConnectionFactory实例
<rabbit:connection-factory id="connectionFactory" host="${rmq.ip}" port="${rmq.port}"
username="${rmq.manager.user}" password="${rmq.manager.password}" />-->
<rabbit:admin connection-factory="connectionFactory"/>

<!--定义消息队列,durable:是否持久化，如果想在RabbitMQ退出或崩溃的时候，
不会失去所有的queue和消息，需要同时标志队列(queue)和交换机(exchange)是持久化的，
即rabbit:queue标签和rabbit:direct-exchange中的durable=true,而消息(message)
默认是持久化的可以看类org.springframework.amqp.core.MessageProperties中的属性
public static final MessageDeliveryMode DEFAULT_DELIVERY_MODE = MessageDeliveryMode.PERSISTENT;exclusive: 仅创建者可以使用的私有队列，断开后自动删除；auto_delete: 当所有消费客户端连接断开后，是否自动删除队列 -->
<rabbit:queue name="spittle.alert.queue.1" id="queue_1" durable="true" auto-delete="false" exclusive="false" />
<rabbit:queue name="spittle.alert.queue.2" id="queue_2" durable="true" auto-delete="false" exclusive="false" />
<rabbit:queue name="spittle.alert.queue.3" id="queue_3" durable="true" auto-delete="false" exclusive="false" />

<!--绑定队列,rabbitmq的exchangeType常用的三种模式：direct，fanout，topic三种,我们用direct模式，
即rabbit:direct-exchange标签，Direct交换器很简单，如果是Direct类型，
就会将消息中的RoutingKey与该Exchange关联的所有Binding中的BindingKey进行比较，
如果相等，则发送到该Binding对应的Queue中。有一个需要注意的地方：如果找不到指定的exchange，就会报错。
但routing key找不到的话，不会报错，这条消息会直接丢失，所以此处要小心,
auto-delete:自动删除，如果为Yes，则该交换机所有队列queue删除后，自动删除交换机，默认为false -->
<rabbit:direct-exchange id="spittle.fanout" name="spittle.fanout" durable="true" auto-delete="false">
    <rabbit:bindings>
        <rabbit:binding queue="spittle.alert.queue.1" key="{alert.queue.1}"></rabbit:binding>
        <rabbit:binding queue="spittle.alert.queue.2" key="{alert.queue.2}"></rabbit:binding>
        <rabbit:binding queue="spittle.alert.queue.3" key="{alert.queue.3}"></rabbit:binding>
    </rabbit:bindings>
</rabbit:fanout-exchange>

<!-- 生产者部分 -->
<!-- 发送消息的producer类，也就是生产者 -->
<bean id="msgProducer" class="com.asdf.sdf.ClassA">
    <!-- value中的值就是producer中的的routingKey，也就是队列名称，
    它与上面的rabbit:bindings标签中的key必须相同 -->
    <property name="queueName" value="{alert.queue.1}"/>
</bean>

<!-- spring amqp默认的是jackson 的一个插件,目的将生产者生产的数据转换为json存入消息队列，
由于fastjson的速度快于jackson,这里替换为fastjson的一个实现 -->
<bean id="jsonMessageConverter" class="com.jy.utils.FastJsonMessageConverter"></bean>
<!-- 或者配置jackson -->
<!--
<bean id="jsonMessageConverter" class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" />
-->

<rabbit:template exchange="test-exchange" id="rabbitTemplate" connection-factory="connectionFactory" message-converter="jsonMessageConverter" />

<!-- 消费者部分 -->
<!-- 自定义接口类 -->
<bean id="testHandler" class="com.rabbit.TestHandler"></bean>

<!-- 用于消息的监听的代理类MessageListenerAdapter -->
<bean id="testQueueListenerAdapter" class="org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter" >
        <!-- 类名 -->
　　　　<constructor-arg ref="testHandler" />
        <!-- 方法名 -->
　　　　<property name="defaultListenerMethod" value="handlerTest"></property>
                <property name="messageConverter" ref="jsonMessageConverter"></property>
</bean>

<!-- 配置监听acknowledeg="manual"设置手动应答，它能够保证即使在一个worker处理消息的时候用CTRL+C
来杀掉这个worker，或者一个consumer挂了(channel关闭了、connection关闭了或者TCP连接断了)，也不会丢
失消息。因为RabbitMQ知道没发送ack确认消息导致这个消息没有被完全处理，将会对这条消息做re-queue处理。
如果此时有另一个consumer连接，消息会被重新发送至另一个consumer会一直重发,直到消息处理成功,
监听容器acknowledge="auto" concurrency="30"设置发送次数,最多发送30次 -->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto" concurrency="20">
        <rabbit:listener queues="spittle.alert.queue.1" ref="testQueueListenerAdapter" />
　　　　<rabbit:listener queues="spittle.alert.queue.2" ref="testQueueListenerAdapter" />
　　　　<rabbit:listener queues="spittle.alert.queue.2" ref="testQueueListenerAdapter" />
</rabbit:listener-container>
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
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

生产者代码

@Resource  
private RabbitTemplate rabbitTemplate;
private String queueName;  
public void sendMessage(CommonMessage msg){
         try {  
              logger.error("发送信息开始");
              System.out.println(rabbitTemplate.getConnectionFactory().getHost());  
             //发送信息  queueName交换机,就是上面的routingKey msg.getSource() 为 test_key 
             rabbitTemplate.convertAndSend(queueName,msg.getSource(), msg);
             //如果是普通字符串消息需要先序列化，再发送消息
             //rabbitTemplate.convertAndSend(queueName,msg.getSource(), SerializationUtils.serialize(msg));
             logger.error("发送信息结束");
         } catch (Exception e) {  
             e.printStackTrace();
         }
    }
public void setQueueName(String queueName) {
    this.queueName = queueName;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

消费端代码

public class TestHandler  {
    @Override
    public void handlerTest(CommonMessage commonMessage) {
        System.out.println("DetailQueueConsumer: " + new String(message.getBody()));
    }
}
1
2
3
4
5
6

Fanout介绍

<!-- Fanout 扇出，顾名思义，就是像风扇吹面粉一样，吹得到处都是。如果使用fanout类型的exchange，
那么routing key就不重要了。因为凡是绑定到这个exchange的queue，都会受到消息。 -->
<rabbit:fanout-exchange name="delayed_message_exchange" durable="true" auto-delete="false" id="delayed_message_exchange">  
        <rabbit:bindings>  
            <rabbit:binding queue="test_delay_queue"/>  
        </rabbit:bindings>  
</rabbit:fanout-exchange>
1
2
3
4
5
6
7

topic介绍

如果说direct是将消息放到exchange绑定的一个queue里（一对一）；fanout是将消息放到exchange绑定的所有queue里（一对所有）；那么topic类型的exchange就可以实现（一对部分），应用场景就是打印不同级别的错误日志，我们的系统出错后会根据不同的错误级别生成error_levelX.log日志，我们在后台首先要把所有的error保存在一个总的queue（绑定了一个.error的路由键）里，然后再按level分别存放在不同的queue。*

<!-- 发送端不是按固定的routing key发送消息，而是按字符串“匹配”发送，接收端同样如此 -->
<rabbit:topic-exchange name="message-exchange" durable="true" auto-delete="false" id="message-exchange">
        <rabbit:bindings>
            <rabbit:binding queue="Q1" pattern="error.*.log" />
            <rabbit:binding queue="Q2" pattern="error.level1.log" />
            <rabbit:binding queue="Q3" pattern="error.level2.log" />
        </rabbit:bindings>
</rabbit:topic-exchange>
1
2
3
4
5
6
7
8

routing key绑定如下图：