SpringBoot 结合 RabbitMQ 实现应用间通信详解

SpringBoot 与 RabbitMQ 实现应用间通信

RabbitMQ 可作为消息队列用于应用程序之间的通信。通过在生产者和消费者代码分别部署于不同应用中，即可完成跨应用通信。

需求描述

用户下单成功后，通知物流系统进行发货。

• 订单系统：生产者
• 物流系统：消费者

创建项目

1. 创建空项目

新建一个 Maven/Gradle 多模块项目。

2. 创建 Module（order-service）

添加 Spring Boot 和 RabbitMQ 依赖。

3. 创建 Module（logistics-service）

添加 Spring Boot 和 RabbitMQ 依赖。

消息类型为字符串

编写订单代码

配置文件 (application.yml)

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.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 "下单成功";
    }
}

编写物流代码

配置文件 (application.yml)

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，将 OrderInfo 类放入其中供双方引用。

import lombok.Data;

@Data
public class OrderInfo {
    private String orderId;
    private String name;
}

编写订单代码

在 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。

解决办法 1：实现序列化接口

修改 OrderInfo 类实现 Serializable 接口：

import lombok.Data;
import java.io.Serializable;

@Data
public class OrderInfo implements Serializable {
    private String orderId;
    private String name;
}

虽然可以发送，但 Payload 不直观。

解决办法 2：设置消息转换类型

使用 Jackson2JsonMessageConverter 使 Payload 更直观。

配置 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;
    }
}

编写物流代码

消费对象类型订单消息

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);
    }
}

注意：当 @RabbitListener 注解声明在类上时，须结合 @RabbitHandler 注解来区分不同的处理方法。

若仍无法解析，原因是消费者端未配置 MessageConverter。需在 logistics-service 的 RabbitMQConfig 中同样添加 Jackson2JsonMessageConverter 和 RabbitTemplate Bean 的配置。

配置完成后，对象类型的订单消息即可正常消费。