消息中间件——RabbitMQ（十）RabbitMQ整合SpringBoot实战！（全）

时间  2019-11-16
标签 消息 中间件 rabbitmq 整合 springboot 实战 栏目 RabbitMQ 繁體版
原文   原文链接

求关注
RabbitMQ整合SpringBoot实战！（全）

前言

1. SpringBoot整合配置详解

  • publisher-confirms,实现一个监听器用于监听Broker端给咱们返回的确认请求：RabbitTemplate.ConfirmCallbackphp

  • publisher-returns,保证消息对Broker端是可达的，若是出现路由键不可达的状况，则使用监听器对不可达的消息进行后续的处理，保证消息的路由成功：RabbitTemplate.ReturnCallbackjava

注意一点，在发送消息的时候对template进行配置mandatory=true保证监听有效 生产端还能够配置其余属性，好比发送重试，超时时间，次数，间隔等git

2. 代码演示

2.1 生产端

2.1.1 新建项目springboot-producer

pom.xmlgithub

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>

	<groupId>com.cp</groupId>
	<artifactId>springboot-producer</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<packaging>jar</packaging>

	<name>springboot-producer</name>
	<description>springboot-producer</description>

	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.0.4.RELEASE</version>
		<relativePath/> <!-- lookup parent from repository -->
	</parent>

	<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
		<java.version>1.8</java.version>
	</properties>

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

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

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>
</project>


复制代码

RabbitSender.java 消息生产者面试

@Component
public class RabbitSender {

	//自动注入RabbitTemplate模板类
	@Autowired
	private RabbitTemplate rabbitTemplate;  
	
	//回调函数: confirm确认
	final ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
		@Override
		public void confirm(CorrelationData correlationData, boolean ack, String cause) {
			System.err.println("correlationData: " + correlationData);
			System.err.println("ack: " + ack);
			if(!ack){
				//能够进行日志记录、异常处理、补偿处理等
				System.err.println("异常处理....");
			}else {
				//更新数据库，可靠性投递机制
			}
		}
	};
	
	//回调函数: return返回
	final ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {
		@Override
		public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText, String exchange, String routingKey) {
			System.err.println("return exchange: " + exchange + ", routingKey: " 
				+ routingKey + ", replyCode: " + replyCode + ", replyText: " + replyText);
		}
	};
	
	//发送消息方法调用: 构建Message消息
	public void send(Object message, Map<String, Object> properties) throws Exception {
		MessageHeaders mhs = new MessageHeaders(properties);
		Message msg = MessageBuilder.createMessage(message, mhs);
		rabbitTemplate.setConfirmCallback(confirmCallback);
		rabbitTemplate.setReturnCallback(returnCallback);
		//id + 时间戳 全局惟一 用于ack保证惟一一条消息,这边作测试写死一个。可是在作补偿策略的时候，必须保证这是全局惟一的消息
		CorrelationData correlationData = new CorrelationData("1234567890");
		rabbitTemplate.convertAndSend("exchange-1", "springboot.abc", msg, correlationData);
	}
	
}


复制代码

application.propertiesspring

spring.rabbitmq.addresses=localhost:5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/vhost_cp
spring.rabbitmq.connection-timeout=15000

spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.template.mandatory=true

复制代码

2.1.2 操做管控台

添加Exchange 数据库

Exchange-1
添加成功

添加Queue apache

Queue-1

Exchange绑定Queue 编程

绑定
绑定成功

修改routingKey，springboot改成spring，则进入的是returnCallback方法springboot

returnCallback打印日志-错误

这时候咱们发现报错了

correlationData: CorrelationData [id=1234567890]
ack: false
异常处理....

复制代码

returnCallback打印日志-正确

2.1.3 解决ack为false问题

这是因为咱们在测试方法中进行测试，当测试方法结束，rabbitmq相关的资源也就关闭了，虽然咱们的消息发送出去，但异步的ConfirmCallback却因为资源关闭而出现了上面的问题。 加入Thread.sleep()便可解决。

@Test
public void testSender1() throws Exception {
	 Map<String, Object> properties = new HashMap<>();
	 properties.put("number", "12345");
	 properties.put("send_time", simpleDateFormat.format(new Date()));
	 rabbitSender.send("Hello RabbitMQ For Spring Boot!", properties);
	 Thread.sleep(2000);
}

复制代码

2.2 消费端

消费端核心配置：

##签收模式-手工签收 spring.rabbitmq.listener.simple.acknowledge-mode=manual ##设置监听限制：最大10，默认5 spring.rabbitmq.listener.simple.concurrency=5 spring.rabbitmq.listener.simple.max-concurrency=10

  • 首先配置手工确认模式，用于ACK的手工处理，这样咱们能够保证消息的可靠性送达，或者再消费端消费失败的时候能够作到重回队列（不建议）、根据业务记录日志等处理。

  • 能够设置消费端的监听个数和最大个数，用于监控消费端的并发状况

@RabbitListener注解使用

  • 消费端监听@RabbitListener注解,这个对于在实际工做中很是的好用
  • @RabbitListener是一个组合注解，里面能够注解配置
  • @QueueBinding、@Queue、@Exchange直接经过这个组合注解一次性搞定消费端交换机、队列、绑定、路由、而且配置监听功能等。

@RabbitListener注解

好比在方法onMessage上加@RabbitListener注解，同时须要加另一个注解@RabbitHandler,代码被消费者监听。

创建绑定，在Value上写上队列，设置Exchange，是否持久化，设置Exchange的类型、表达式设置为true以及路由key。经过这种简单的方式，就能够完成以前很复杂的代码逻辑。同时建议将配置放入到配置文件中，动态获取。若是mq中没有相应的队列、Exchange等，注解声明也能够建立它们，你们能够自行测试！

2.2.1 新建项目springboot-consumer

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>

	<groupId>com.cp</groupId>
	<artifactId>springboot-consumer</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<packaging>jar</packaging>

	<name>springboot-consumer</name>
	<description>springboot-consumer</description>

	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.0.2.RELEASE</version>
		<relativePath/> <!-- lookup parent from repository -->
	</parent>

	<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
		<java.version>1.8</java.version>
	</properties>

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

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

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>


</project>



复制代码

RabbitReceiver.java 消息生产者

@Component
public class RabbitReceiver {

	
	@RabbitListener(bindings = @QueueBinding(
			value = @Queue(value = "queue-1", 
			durable="true"),
			exchange = @Exchange(value = "exchange-1", 
			durable="true", 
			type= "topic", 
			ignoreDeclarationExceptions = "true"),
			key = "springboot.*"
			)
	)
	@RabbitHandler
	public void onMessage(Message message, Channel channel) throws Exception {
		System.err.println("--------------------------------------");
		System.err.println("消费端Payload: " + message.getPayload());
		Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
		//手工ACK,获取deliveryTag
		channel.basicAck(deliveryTag, false);
	}
}


复制代码

application.properties

spring.rabbitmq.addresses=localhost:5672
spring.rabbitmq.username=user_cp
spring.rabbitmq.password=123456
spring.rabbitmq.virtual-host=/vhost_cp
spring.rabbitmq.connection-timeout=15000

spring.rabbitmq.listener.simple.acknowledge-mode=manual
spring.rabbitmq.listener.simple.concurrency=5
spring.rabbitmq.listener.simple.max-concurrency=10

复制代码

运行Application，查看以前在生产端发送的消息，是否能被消费。

打印结果

打印结果
这里以前因为我测试的时候多发了消息，因此消费的时候会有这么多。

3. 优化代码

  • 自定义Java对象消息
  • @RabbitListener注解中的配置改成动态配置

@Payload：指定具体的消息体Body。 @Headers: 获取Headers。

3.1 消费端优化

一、先定义一个Order对象

public class Order implements Serializable {

	private String id;
	private String name;
	
	public Order() {
	}
	public Order(String id, String name) {
		super();
		this.id = id;
		this.name = name;
	}
	public String getId() {
		return id;
	}
	public void setId(String id) {
		this.id = id;
	}
	public String getName() {
		return name;
	}
	public void setName(String name) {
		this.name = name;
	}
}


复制代码

注意：咱们在传输对象的时候，必须序列化。不然会传输失败。

二、RabbitReceiver添加监听

/** * * spring.rabbitmq.listener.order.queue.name=queue-2 spring.rabbitmq.listener.order.queue.durable=true spring.rabbitmq.listener.order.exchange.name=exchange-2 spring.rabbitmq.listener.order.exchange.durable=true spring.rabbitmq.listener.order.exchange.type=topic spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=true spring.rabbitmq.listener.order.key=springboot.* * @param order * @param channel * @param headers * @throws Exception */
	@RabbitListener(bindings = @QueueBinding(
			value = @Queue(value = "${spring.rabbitmq.listener.order.queue.name}", 
			durable="${spring.rabbitmq.listener.order.queue.durable}"),
			exchange = @Exchange(value = "${spring.rabbitmq.listener.order.exchange.name}", 
			durable="${spring.rabbitmq.listener.order.exchange.durable}", 
			type= "${spring.rabbitmq.listener.order.exchange.type}", 
			ignoreDeclarationExceptions = "${spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions}"),
			key = "${spring.rabbitmq.listener.order.key}"
			)
	)
	@RabbitHandler
	public void onOrderMessage(@Payload com.cp.springboot.entity.Order order, Channel channel, @Headers Map<String, Object> headers) throws Exception {
		System.err.println("--------------------------------------");
		System.err.println("消费端order: " + order.getId());
		Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
		//手工ACK
		channel.basicAck(deliveryTag, false);
	}

复制代码

已经将配置写入到了application.properties中，进行动态获取。也能够像咱们公司同样放入到配置中心当中。例如：携程开源配置中心Apollo

三、application.properties

spring.rabbitmq.listener.order.queue.name=queue-2
spring.rabbitmq.listener.order.queue.durable=true
spring.rabbitmq.listener.order.exchange.name=exchange-2
spring.rabbitmq.listener.order.exchange.durable=true
spring.rabbitmq.listener.order.exchange.type=topic
spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=true
spring.rabbitmq.listener.order.key=springboot.*

复制代码

3.2 生产端优化

一、一样是一个Order对象，必须跟消费端的保持一致。

二、RabbitSender添加发送消息

//发送消息方法调用: 构建自定义对象消息
public void sendOrder(Order order) throws Exception {
	rabbitTemplate.setConfirmCallback(confirmCallback);
	rabbitTemplate.setReturnCallback(returnCallback);
	//id + 时间戳 全局惟一 
	CorrelationData correlationData = new CorrelationData("0987654321");
	rabbitTemplate.convertAndSend("exchange-2", "springboot.def", order, correlationData);
}


复制代码

三、添加测试方法

@Test
public void testSender2() throws Exception {
	 Order order = new Order("001", "第一个订单");
	 rabbitSender.sendOrder(order);
	 //防止资源提早关闭，ConfirmCallback异步回调失败
	 Thread.sleep(2000);
}

复制代码

3.3测试

运行testSender2()方法。

生产端打印消息

生产端打印消息

消费端打印消息

消费端打印消息

至此，RabbitMQ整合SpringBoot完毕，在实际工做中，使用场景也是差很少的。

文末

欢迎关注我的微信公众号：Coder编程 获取最新原创技术文章和免费学习资料，更有大量精品思惟导图、面试资料、PMP备考资料等你来领，方便你随时随地学习技术知识！ 新建了一个qq群：315211365，欢迎你们进群交流一块儿学习。谢谢了！也能够介绍给身边有须要的朋友。

文章收录至 Github: github.com/CoderMerlin… Gitee: gitee.com/573059382/c… 欢迎关注并star~

微信公众号

参考文章：

《RabbitMQ消息中间件精讲》

推荐文章：

消息中间件——RabbitMQ（七）高级特性全在这里!（上）

消息中间件——RabbitMQ（八）高级特性全在这里!（下）

消息中间件——RabbitMQ（九）RabbitMQ整合Spring AMQP实战！（全）

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程