详解SpringBoot+RabbitMQ完成应用通信
目录
应用通信
作为⼀个消息队列, RabbitMQ也可以⽤作应⽤程序之间的通信. 上述代码⽣产者和消费者代码放在不同的应⽤中即可完成不同应⽤程序的通信.
接下来我们来看, 基于SpringBoot+RabbitMQ完成应⽤间的通信.
需求描述
⽤⼾下单成功之后, 通知物流系统, 进⾏发货。
订单系统作为⼀个⽣产者, 物流系统作为⼀个消费者。
创建项目
创建空项目
创建Module（order-service）
添加依赖
创建Module（logistics-service）
添加依赖
消息类型为字符串
编写订单代码
添加配置
spring: application: name: order-service rabbitmq: addresses: amqp://study:[email protected]:5672/order
声明队列
import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.QueueBuilder; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitMQConfig { @Bean("orderQueue") public Queue orderQueue() { return QueueBuilder.durable("order.create").build(); } }生产订单
import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import java.util.Random; import java.util.UUID; @RequestMapping("/order") @RestController public class OrderController { @Autowired private RabbitTemplate rabbitTemplate; @RequestMapping("create") public String create(){ String orderId= UUID.randomUUID().toString(); rabbitTemplate.convertAndSend("","order.create","订单信息，订单ID:"+orderId); return "下单成功"; } }编写物流代码
添加配置
spring: application: name: logistics-service rabbitmq: addresses: amqp://study:[email protected]:5672/order server: port: 9090
声明队列
import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.QueueBuilder; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitMQConfig { @Bean("order.create") public Queue createOrder() { return QueueBuilder.durable("order.create").build(); } }消费订单
import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component public class OrderListener { @RabbitListener(queues = "order.create") public void handMessage(String orderInfo) { System.out.println("接收到订单消息"+orderInfo); } }生产订单
消费订单
可以看到，订单已经被成功消费了，结果符合预期。
消息类型为对象
下面的代码是以上面代码的基础，并进行增添删除编写而成的。
新增Module
因为订单系统和物流系统在生产和消费对象类型的消息时，会用到同一个结构体，为了便于代码编写，新增Module，把OrderInfo放到新增的Module中。
import lombok.Data; @Data public class OrderInfo{ private String orderId; private String name; }
编写订单代码
在order-service项目的OrderController中添加如下代码：
@RequestMapping("create2") public String create2(){ OrderInfo orderInfo=new OrderInfo(); orderInfo.setOrderId(UUID.randomUUID().toString()); orderInfo.setName("商品"+new Random().nextInt(100)); rabbitTemplate.convertAndSend("","order.create",orderInfo); return "下单成功"; }生产对象类型订单消息
此时我们发现抛异常了，异常信息为SimpleMessageConverter只支持String, byte[] and Serializable类型的payloads，但是接收到的是order.model.OrderInfo对象。
解决办法1（实现序列化接口）
修改order-service下的OrderInfo类
import lombok.Data; import java.io.Serializable; @Data public class OrderInfo implements Serializable { private String orderId; private String name; }
虽然成功生成了对象类型的订单消息，但是我们看到消息的Payload并不直观，下面我们将采用两一种解决办法。
解决办法2（设置消息转换类型）
修改order-service中的OrderInfo
import lombok.Data; @Data public class OrderInfo{ private String orderId; private String name; }查看RabbitTemplated的setMessageConverter方法，查看MessageConverter接口的实现类，我们将使用Jackson2JsonMessageConverter。
修改order-service中的RabbitMQConfig
import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.QueueBuilder; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitMQConfig { @Bean("orderQueue") public Queue orderQueue() { return QueueBuilder.durable("order.create").build(); } @Bean public Jackson2JsonMessageConverter jsonMessageConverter(){ return new Jackson2JsonMessageConverter(); } @Bean public RabbitTemplate rabbitTemplate(ConnectionFactory factory, Jackson2JsonMessageConverter jsonMessageConverter){ RabbitTemplate rabbitTemplate = new RabbitTemplate(factory); rabbitTemplate.setMessageConverter(jsonMessageConverter); return rabbitTemplate; } }再次生产对象类型订单消息
此时再次查看队列中的消息，我们可以看到消息的Payload是非常直观的！！！
编写物流代码
修改logistics-service中的代码
import logistics.model.OrderInfo; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component @RabbitListener(queues = "order.create") public class OrderListener { public void handMessage(String orderInfo) { System.out.println("接收到订单消息"+orderInfo); } public void handMessage(OrderInfo orderInfo) { System.out.println("接收到订单消息"+orderInfo); } }消费对象类型订单消息
但是过了许久，我们也没有看到队列中的消息被消费！！！
解决办法
原因是因为@RabbitListener(queues = "order.create")注解声明在类上面时，须结合@RabbitHandler注解来使用！！！
修改logistics-service中的代码
import logistics.model.OrderInfo; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component @RabbitListener(queues = "order.create") public class OrderListener { @RabbitHandler public void handMessage(String orderInfo) { System.out.println("接收到订单消息"+orderInfo); } @RabbitHandler public void handMessage(OrderInfo orderInfo) { System.out.println("接收到订单消息"+orderInfo); } }此时消费消息，但是发现又抛异常了
原因是因为，我们的确是针对生产对象类型消息的一方设置了MessageConverter，但是我们没有给消费对象类型消息的一方设置MessageConverter，导致消费对象类型消息的一方没有办法正确解析出消息。
解决办法，依旧是添加MessageConverter！！！
修改logistics-service中的RabbitMQConfig
import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.QueueBuilder; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitMQConfig { @Bean("order.create") public Queue createOrder() { return QueueBuilder.durable("order.create").build(); } @Bean public Jackson2JsonMessageConverter jsonMessageConverter(){ return new Jackson2JsonMessageConverter(); } @Bean public RabbitTemplate rabbitTemplate(ConnectionFactory factory, Jackson2JsonMessageConverter jsonMessageConverter){ RabbitTemplate rabbitTemplate = new RabbitTemplate(factory); rabbitTemplate.setMessageConverter(jsonMessageConverter); return rabbitTemplate; } }再次消费消息
此时我们可以看到，对象类型的订单消息可以正常消费了，符合预期。