SpringBoot中配置RabbitMQ

Posted 2022-04-27 程序员超时空

RabbitMQ介绍与原理见：消息队列之 RabbitMQ

RabbitMQ安装见：RabbitMQ安装

依赖：

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

在Spring Boot中配置RabbitMQ比较简单，首先配置yml文件，properties也可以。

application-rabbitmq.yml

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    connection-timeout: 60s
    listener:
      direct:
        acknowledge-mode: none

RabbitMQ默认的ExChange消息类型是direct，而且不用自己配置交换机exchange、路由键routing-key及bind信息。发送消息amqpTemplate.convertAndSend(routingKey, data)时，将队列名字作为routing-key和消息一块发送。在consumer监听相应队列即可。

编写配置类RabbitMqOperator：

package com.qiqi.mq;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

/**
 * 消息队列
 */
@Component
public class RabbitMqOperator 

    private AmqpTemplate amqpTemplate;

    @Autowired
    public RabbitMqOperator(AmqpTemplate amqpTemplate) 
        this.amqpTemplate = amqpTemplate;
    

    /**
     * 异步发送字节数组到rabbitmq
     * @param data
     */
    public void push(String routingKey, byte[] data)
        amqpTemplate.convertAndSend(routingKey, data);
    


    /**
     * 异步发送int类型消息到rabbitmq
     * @param data
     */
    public void pushInt(String routingKey, int data)
        amqpTemplate.convertAndSend(routingKey, data);
    

    /**
     * 定义一个名为：first_queue 的队列
     * @return
     */
    @Bean
    public Queue firstQueue() 
        return new Queue("first_queue");
    

    /**
     * 定义一个名为：second_queue 的队列
     * @return
     */
    @Bean
    public Queue secondQueue() 
        return new Queue("second_queue");
    

生产者：

public class Productor

    //消息队列
    private RabbitMqOperator rabbitMqOperator;

    @Autowired
    public Productor(RabbitMqOperator operator) 
        this.rabbitMqOperator = operator;
        Assert.notNull(this.rabbitMqOperator, "rabbitmq init fail.");
    

    /**
     * 数据发送到rabbitMq
     */
    public void sendData() 

        byte[] data = new byte[11];
       // data[10] = ;
        //发送数据到first_queue队列
        rabbitMqOperator.push("first_queue" ,data);

        for (int i = 0; i < 100; i++)
            //发送数据到second_queue队列
            rabbitMqOperator.pushInt("second_queue",i);
        
    

消费者：

package com.qiqi.mq;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * 消费者
 */
@Component
public class Consumer 

    private static final Logger log = LoggerFactory.getLogger(Consumer.class);

    /**
     * @RabbitHandler 指定消息的处理方法
     * @param message
     */
    @RabbitListener(queues = "first_queue")@RabbitListener 监听 first_queue 队列
    @RabbitHandler
    public void process(String message) 
         //接收first_queue队列消息
        log.info("接收的消息为: ", message);
    

    @RabbitListener(queues = "second_queue")@RabbitListener 监听 second_queue 队列
    @RabbitHandler
    public void process1(String message) 
         //接收second_queue队列消息
        log.info("2接收的消息为: ", message);
    

输出例子：