RabbitMq 参数配置详解

除了 连接MQ代理、建立交换机和队列、发送消息，还需要 发送方确认消息发送成功、接收方消费后反馈
这里发送消息等不再是Channel而是RabbitTemplate

一、Spring整合RabbitMQ

生产方

1. 依赖

<dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><version>2.1.8.RELEASE</version>
</dependency>

2. properties配置文件

rabbitmq.host=192.168.12.135
rabbitmq.port=5672
rabbitmq.username=host
rabbitmq.password=host
rabbitmq.virtual-host=/test
rabbitmq.publisher-confirms=true //是否开启confirm模式，默认为false
rabbitmq.publisher-returns=true //是否开启return模式，默认为false

3. spring-rabbitmq.xml配置文件

<?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:context="http://www.springframework.org/schema/context"xmlns:rabbit="http://www.springframework.org/schema/rabbit"xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/contexthttps://www.springframework.org/schema/context/spring-context.xsdhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit.xsd"><!--加载配置文件--><context:property-placeholder location="classpath:properties/rabbitmq.properties"/><!-- 定义rabbitmq connectionFactory --><rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"port="${rabbitmq.port}"username="${rabbitmq.username}"password="${rabbitmq.password}"virtual-host="${rabbitmq.virtual-host}"publisher-confirms="${rabbitmq.publisher-confirms}"publisher-returns="${rabbitmq.publisher-returns}" /><!--定义管理交换机、队列--><rabbit:admin connection-factory="connectionFactory"/><!--auto-declare：不存在则自动创建；不绑定到交换机则绑定到默认交换机--><rabbit:queue id="spring_queue" name="spring_queue" auto-declare="true"/><!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~广播；所有队列都能收到消息 start~~~~~~~~~~~~~~~~~~~~~~~~~~~~ --><!--定义广播交换机中的持久化队列，不存在则自动创建--><rabbit:queue id="spring_fanout_queue_1" name="spring_fanout_queue_1" auto-declare="true"/><!--定义广播交换机中的持久化队列，不存在则自动创建--><rabbit:queue id="spring_fanout_queue_2" name="spring_fanout_queue_2" auto-declare="true"/><!--定义广播类型交换机；并绑定上述两个队列--><rabbit:fanout-exchange id="spring_fanout_exchange" name="spring_fanout_exchange" auto-declare="true"><rabbit:bindings><rabbit:binding queue="spring_fanout_queue_1"/><rabbit:binding queue="spring_fanout_queue_2"/></rabbit:bindings></rabbit:fanout-exchange><!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~广播；所有队列都能收到消息 end~~~~~~~~~~~~~~~~~~~~~~~~~~~~ --><!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~通配符；*匹配一个单词，#匹配多个单词 start~~~~~~~~~~~~~~~~~~~~~~~~~~~~ --><!--定义广播交换机中的持久化队列，不存在则自动创建--><rabbit:queue id="spring_topic_queue_star" name="spring_topic_queue_star" auto-declare="true"/><!--定义广播交换机中的持久化队列，不存在则自动创建--><rabbit:queue id="spring_topic_queue_well" name="spring_topic_queue_well" auto-declare="true"/><!--定义广播交换机中的持久化队列，不存在则自动创建--><rabbit:queue id="spring_topic_queue_well2" name="spring_topic_queue_well2" auto-declare="true"/><!--创建主题交换机，绑定队列--><rabbit:topic-exchange id="spring_topic_exchange" name="spring_topic_exchange" auto-declare="true"><rabbit:bindings><rabbit:binding pattern="heima.*" queue="spring_topic_queue_star"/><rabbit:binding pattern="heima.#" queue="spring_topic_queue_well"/><rabbit:binding pattern="itcast.#" queue="spring_topic_queue_well2"/></rabbit:bindings></rabbit:topic-exchange><!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~通配符；*匹配一个单词，#匹配多个单词 end~~~~~~~~~~~~~~~~~~~~~~~~~~~~ --><!--定义rabbitTemplate对象操作可以在代码中方便发送消息，这里需要指定connectionFactory--><rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
</beans>

4. 发送消息

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring/spring-rabbitmq.xml")
public class ProducerTest {@Autowiredprivate RabbitTemplate rabbitTemplate;//定义confirm和callback的回调函数@Beforepublic void callback(){//1. 设置confirmCallBackrabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {/*** 发送消息后此方法就会被执行* @param correlationData ：消息相关配置信息* @param b ：交换机是否接收到消息* @param s ：失败原因，成功为null*/@Overridepublic void confirm(CorrelationData correlationData, boolean b, String s) {if(b){System.out.println("confirm方法被执行，消息发送成功");}else {System.out.println("confirm方法被执行，消息发送失败");System.out.println("错误信息: "+s);//可以自己定义一些逻辑，让消息再次发送，具体怎么搞没研究过}}});rabbitTemplate.setMandatory(true);//2.设置ReturnCallBack,rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {/**** @param message   消息对象* @param replyCode 错误码* @param replyText 错误信息* @param exchange  交换机* @param routingKey 路由键*/@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {System.out.println("return 执行了....");System.out.println(message);System.out.println(replyCode);System.out.println(replyText);System.out.println(exchange);System.out.println(routingKey);//处理，会让交换机将消息给其它队列}});}/*** 只发队列消息* 默认交换机类型为 direct* 交换机的名称为空，路由键为队列的名称*/@Testpublic void queueTest(){//路由键与队列同名rabbitTemplate.convertAndSend("spring_queue", "只发队列spring_queue的消息。");}/*** 发送广播* 交换机类型为 fanout* 绑定到该交换机的所有队列都能够收到消息*/@Testpublic void fanoutTest(){/*** 参数1：交换机名称* 参数2：路由键名（广播设置为空）* 参数3：发送的消息内容*/rabbitTemplate.convertAndSend("spring_fanout_exchange", "", "发送到spring_fanout_exchange交换机的广播消息");}/*** 通配符* 交换机类型为 topic* 匹配路由键的通配符，*表示一个单词，#表示多个单词* 绑定到该交换机的匹配队列能够收到对应消息*/@Testpublic void topicTest(){/*** 参数1：交换机名称* 参数2：路由键名* 参数3：发送的消息内容*/rabbitTemplate.convertAndSend("spring_topic_exchange", "heima.bj", "发送到spring_topic_exchange交换机heima.bj的消息");rabbitTemplate.convertAndSend("spring_topic_exchange", "heima.bj.1", "发送到spring_topic_exchange交换机heima.bj.1的消息");rabbitTemplate.convertAndSend("spring_topic_exchange", "heima.bj.2", "发送到spring_topic_exchange交换机heima.bj.2的消息");rabbitTemplate.convertAndSend("spring_topic_exchange", "itcast.cn", "发送到spring_topic_exchange交换机itcast.cn的消息");}
}

消费方

1. 依赖

<dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><version>2.1.8.RELEASE</version>
</dependency>

2. properties配置文件

rabbitmq.host=192.168.12.135
rabbitmq.port=5672
rabbitmq.username=host
rabbitmq.password=host
rabbitmq.virtual-host=/test
rabbitmq.publisher-confirms=true //是否开启confirm模式，默认为false
rabbitmq.publisher-returns=true //是否开启return模式，默认为false

<beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:context="http://www.springframework.org/schema/context"xmlns:rabbit="http://www.springframework.org/schema/rabbit"xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/contexthttps://www.springframework.org/schema/context/spring-context.xsdhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit.xsd"><!--加载配置文件--><context:property-placeholder location="classpath:properties/rabbitmq.properties"/><!-- 定义rabbitmq connectionFactory --><rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"port="${rabbitmq.port}"username="${rabbitmq.username}"password="${rabbitmq.password}"virtual-host="${rabbitmq.virtual-host}"/><!--右边为自定义的监听器，在3中致谢了第一个可以类比于`channel.basicConsume()`的参数（callback）`DefaultConsumer类`**--><bean id="springQueueListener" class="com.rabbitmq.listener.SpringQueueListener"/><bean id="fanoutListener1" class="com.rabbitmq.listener.FanoutListener1"/><bean id="fanoutListener2" class="com.rabbitmq.listener.FanoutListener2"/><bean id="topicListenerStar" class="com.rabbitmq.listener.TopicListenerStar"/><bean id="topicListenerWell" class="com.rabbitmq.listener.TopicListenerWell"/><bean id="topicListenerWell2" class="com.rabbitmq.listener.TopicListenerWell2"/><!-- 自动确认，在这里才指定监听从哪个队列出来的消息 --><rabbit:listener-container connection-factory="connectionFactory" auto-declare="true"><rabbit:listener ref="springQueueListener" queue-names="spring_queue"/><rabbit:listener ref="fanoutListener1" queue-names="spring_fanout_queue_1"/><rabbit:listener ref="fanoutListener2" queue-names="spring_fanout_queue_2"/><rabbit:listener ref="topicListenerStar" queue-names="spring_topic_queue_star"/><rabbit:listener ref="topicListenerWell" queue-names="spring_topic_queue_well"/><rabbit:listener ref="topicListenerWell2" queue-names="spring_topic_queue_well2"/></rabbit:listener-container>
</beans>

3. 定义监听器

不管是自动还是手动确认，都会执行监听类，其实与其说是监听类可以说是消费逻辑，监听器是负责接收消息对消息消费的，设置手动确认时会在里面自定义一些确认的逻辑。

实现MessageListener，重写onMessage方法即可，其它监听器一样，这里不赘述。这个可以类比 channel.basicConsume()的参数（callback）DefaultConsumer类

public class SpringQueueListener implements MessageListener {
//Message类中封装了消息的相关信息，这个监听器类会在接收消息后自动执行@Overridepublic void onMessage(Message message) {try {String msg = new String(message.getBody(), "utf-8");System.out.printf("接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",message.getMessageProperties().getReceivedExchange(),message.getMessageProperties().getReceivedRoutingKey(),message.getMessageProperties().getConsumerQueue(),msg);} catch (Exception e) {e.printStackTrace();}}
}

消费端测试代码

拦截器会自动执行消费消息，这里只是保证消费端处于启动状态，加载spring-rabbitmq-consumer.xml配置文件

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-consumer.xml")
public class ConsumerTest {@Testpublic void test1(){boolean flag = true;while (true){}}
}

---

二、Spring Boot整合RabbitMQ

生产者

1. 依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2. application.yml配置文件

RabbitMQ相关参数配置见：Springboot-RabbitMq 参数配置详解

spring:rabbitmq:host: localhostport: 5672virtual-host: /itcastusername: heimapassword: heimaspring.rabbitmq.publisher-confirms: 是否启用【发布确认】，默认false
spring.rabbitmq.publisher-returns: 是否启用【发布返回】，默认false

3. 配置类

定义交换机和队列

@Configuration
public class RabbitMQConfig {public static final String EXCHANGE_NAME = "boot_topic_exchange";public static final String QUEUE_NAME = "boot_queue";//1.交换机@Bean("bootExchange")public Exchange bootExchange(){return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();}//2.Queue 队列@Bean("bootQueue")public Queue bootQueue(){return QueueBuilder.durable(QUEUE_NAME).build();}//3. 队列和交互机绑定关系 Binding/*1. 知道哪个队列2. 知道哪个交换机3. routing key4. @Qualifier指定了名称，而Queue指定了类型，相当于和@Authoud的组合*/@Beanpublic Binding bindQueueExchange(@Qualifier("bootQueue") Queue queue, @Qualifier("bootExchange") Exchange exchange){//需要指定参数就and()不需要就noargs()；return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();}}

4. 发送消息

通过RabbitTemplate来完成

import com.itheima.rabbitmq.config.RabbitMQConfig;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMQTest {@Autowiredprivate RabbitTemplate rabbitTemplate;//发送消息@Testpublic void test(){rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE, "item.insert", "商品新增，routing key 为item.insert");rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE, "item.update", "商品修改，routing key 为item.update");rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE, "item.delete", "商品删除，routing key 为item.delete");}

消费者

1. 依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2. application.yml配置文件

RabbitMQ相关参数配置见：Springboot-RabbitMq 参数配置详解

spring:rabbitmq:host: localhostport: 5672virtual-host: /itcastusername: heimapassword: heimaspring.rabbitmq.listener.simple.acknowledge-mode: manual --设置为手动
spring.rabbitmq.listener.direct.acknowledge-mode: manual

3. 定义监听器

Springboot整合RabbitMQ 消费者消息接收确认

消费消息

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class MyListener {/*** 监听某个队列的消息，会自动执行。* @param message 接收到的消息*/@RabbitListener(queues = "item_queue")public void myListener1(String message){System.out.println("消费者接收到的消息为：" + message);}
}