spring集成RabbitMQ配置文件详解（生产者和消费者）

1，首先引入配置文件org.springframework.amqp，如下：

<dependency>

            <groupId>org.springframework.amqp</groupId>

            <artifactId>spring-rabbit</artifactId>

            <version>1.7.1.RELEASE</version>

        </dependency>

2，准备工作：安装好rabbitmq,并在项目中增加配置文件 rabbit.properties 内容如下：

rmq.ip=192.188.113.114

rmq.port=5672

rmq.producer.num=20

rmq.manager.user=admin

rmq.manager.password=admin

3，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。

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

客户端连接到消息队列服务器，打开一个channel。

客户端声明一个exchange，并设置相关属性。

客户端声明一个queue，并设置相关属性。

客户端使用routing key，在exchange和queue之间建立好绑定关系。

消息发布

客户端投递消息到exchange。

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

AMQP 里主要要说两个组件：

Exchange 和 Queue

绿色的X就是Exchange ，红色的是Queue，这两者都在Server端，又称作Broker

这部分是RabbitMQ实现的，而蓝色的则是客户端，通常有Producer和Consumer两种类型。

4，配置spring-rabbitmq.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>

5，生产者（发送端）代码：

@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;

}

6，消费端代码：TestHandler 类

public class TestHandler  {

    @Override

    public void handlerTest(CommonMessage commonMessage) {

        System.out.println("DetailQueueConsumer: " + new String(message.getBody()));

    }

}

其他exchangeType介绍：

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>

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>

routing key绑定如下图：

本文转自：

https://blog.csdn.net/nandao158/article/details/81065892

https://www.cnblogs.com/LipeiNet/p/6079427.html

https://www.toutiao.com/a6598154241037042189/?tt_from=mobile_qq&utm_campaign=client_share&timestamp=1536277584&app=news_article&utm_source=mobile_qq&iid=43157585039&utm_medium=toutiao_android&group_id=6598154241037042189

spring集成RabbitMQ配置文件详解（生产者和消费者）

spring集成RabbitMQ配置文件详解（生产者和消费者）的相关教程结束。

相关推荐

塔瑞斯世界职业技能详解全职业技能都有哪些

塔瑞斯世界职业大全一览全职业效果详解

塔瑞斯世界游侠和法师技能详解游侠和法师怎么选择

塔瑞斯世界各职业强度详解各职业强度如何

幻兽帕鲁帕鲁矿碎片获取方法详解

dnf锁1锁2什么意思锁1锁2洗衣粉含义详解

弈仙牌修为机制详解修为怎么增加

弈仙牌琴剑阵打法思路攻略详解

spring集成RabbitMQ配置文件详解（生产者和消费者）

spring集成RabbitMQ配置文件详解（生产者和消费者）的相关教程结束。

相关推荐

塔瑞斯世界职业技能详解 全职业技能都有哪些

塔瑞斯世界职业大全一览 全职业效果详解

塔瑞斯世界游侠和法师技能详解 游侠和法师怎么选择

塔瑞斯世界各职业强度详解 各职业强度如何

幻兽帕鲁帕鲁矿碎片获取方法详解

dnf锁1锁2什么意思 锁1锁2洗衣粉含义详解

弈仙牌修为机制详解 修为怎么增加

弈仙牌琴剑阵打法思路攻略详解

塔瑞斯世界职业技能详解全职业技能都有哪些

塔瑞斯世界职业大全一览全职业效果详解

塔瑞斯世界游侠和法师技能详解游侠和法师怎么选择

塔瑞斯世界各职业强度详解各职业强度如何

dnf锁1锁2什么意思锁1锁2洗衣粉含义详解

弈仙牌修为机制详解修为怎么增加