SpringBoot 整合RabbitMq

一、准备环境

1、SpringBoot 导入RabbitMq依赖

        <!--rabbitmq依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <!--web依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

这里要导入web依赖，因为如果没有web依赖，会出现NoClassDefFoundError错误

2、配置RabbitMq连接

# rabbitmq配置
# 连接地址
spring.rabbitmq.host=127.0.0.1
# 端口号 （默认是5672,默认与实际相同，可以不用配置）
spring.rabbitmq.port=5672
# 用户名
spring.rabbitmq.username=guest
# 密码
spring.rabbitmq.password=guest
# 虚拟主机（默认是 / ）
spring.rabbitmq.virtual-host=/

二、开始使用RabbitMq

1、生产者消息推送

SpringBoot整合RabbitMq主要用RabbitTemplate来实现对队列的操作

• 以java序列化的方式存储在队列

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void sendDirect() {
        Map<String,Object> map = new HashMap<>();
        map.put("imei","第一个mq消息");
        map.put("cmd","0D00");
        // convertAndSend(交换器，路由键，消息体)
        rabbitTemplate.convertAndSend("my_exchanges","news",map);
       // rabbitTemplate.convertAndSend("my_exchanges","news",new DeviceCmd("12345678","0D00"));
    }

默认以java序列化方式存储

• RabbitMq可以把数据序列化为json格式存储在队列里
  重写的MessageConverter messageConverter()方法使其返回Jackson2JsonMessageConverter()方法，，并注入spring容器

@Configuration
public class MyAMQPConfig {
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}

然后运行生产者推送消息的方法 结果：

以json格式存储

• 广播模式推送消息
  广播模式交换器（fanout）不需要识别路由键，会将消息推送到所有绑定的队列

    // 测试广播模式推送消息
    @Test
    public void sendFanout(){
        // 广播模式推送消息，可以不用写 路由键，会默认发送到交换器绑定的所有队列里
        rabbitTemplate.convertAndSend("my_exchange_fanout","",new DeviceCmd("54321","0D01"));
    }

2、消费者接收消息

• 消费者手动触发消费
  消费者接收消息，返回类型是Object类型，可以手动转换消息类型，每次接收一条消息
  注意：如果消息体无法转换成待转类型，会报消息转换异常

    @Test
    public void receive(){
        // 接收并转换消息
        // receiveAndConvert("queueName"); 从指定队列中取出消息
        Object o = rabbitTemplate.receiveAndConvert("my_queue_old");
        System.out.println(o);
//        DeviceCmd deviceCmd = (DeviceCmd) rabbitTemplate.receiveAndConvert("my_queue_old");
//        System.out.println(deviceCmd.getClass());
//        System.out.println(deviceCmd.getCmd() + " -- " + deviceCmd.getImei());
    }

• 消费者监听队列消费
  1）首先要在项目启动时，开启RabbitMq 在启动类上添加@EnableRabbit注解

@EnableRabbit
@SpringBootApplication
public class DemoApplication {
    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }
}

2）在项目中添加@RabbitListener注解，来实现队列的监听
注意：监听方法所在类必须在Spring容器中

@Component
public class DeviceCmdService {

    @RabbitListener(queues = {"test.queue"})
    public void receive(byte[] body){
        String s = new String(body);
        System.out.println(s);
    }
}

监听队列也可以直接将消息转换为对象或自己想要的类型，如果消息无法转换，会出现消息转换异常

    @RabbitListener(queues = {"test.queue"})
    public void receive(DeviceCmd deviceCmd){
        System.out.println(deviceCmd);
    }

3、队列管理AmqpAdmin

SpringBoot整合RabbitMq根据AmqpAdmin类来实现对交换器，队列进行创建、删除和绑定操作

    @Autowired
    private AmqpAdmin amqpAdmin;
    @Test
    public void createExchange(){
        // 创建交换机 new FanoutExchange()、new TopicExchange() new CustomExchange()
        amqpAdmin.declareExchange(new DirectExchange("amqpAdmin.exchange"));
        // 创建队列  new Queue("队列名称")
        amqpAdmin.declareQueue(new Queue("amqpAdmin.queue"));

        // 绑定队列与交换机  new Binding("目标名称",目标类行，"交换机名称"，"路由键"，)
        amqpAdmin.declareBinding(
                new Binding("amqpAdmin.queue",Binding.DestinationType.QUEUE,"amqpAdmin.exchange","amqpAdmin.key",null));

    }