RabbitMQ整合Springboot

首先、准备依赖

    4.0.0
    
        org.springframework.boot
        spring-boot-starter-parent
        2.5.9
         
    
    com.example
    rabbitmqboot
    0.0.1-SNAPSHOT
    rabbitmqboot
    rabbitmqboot
    
        1.8
    
    
        
            org.springframework.boot
            spring-boot-starter-amqp
        

        
            org.springframework.boot
            spring-boot-starter-web
        

修改配置文件

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

创建交换机

package com.example.rabbitmqboot.common;

public class Constant {

    public static final String DIRECT_EXCHANGE_A = "EXCHANGE_A";
    public static final String QUEUE_A = "QA";
}

@Configuration
public class AllExchange {

    @Bean("exchange_A")
    public DirectExchange exchangeA() {
        return new DirectExchange(Constant.DIRECT_EXCHANGE_A);
    }
}

创建队列，同时绑定交换机

 @Bean("queueA")
    public Queue queueA () {
        //HashMap arguments = new HashMap<>(3);
        //绑定死信队列
        //arguments.put("x-dead-letter-exchange", Constant.DIRECT_EXCHANGE_Y);
        //arguments.put("x-dead-letter-routing-key", "YD");
        //arguments.put("x-message-ttl", 40000);
        //return QueueBuilder.durable(Constant.QUEUE_A).withArguments(arguments).build();
       return QueueBuilder.durable(Constant.QUEUE_A).build();
    }

    @Bean
    public Binding bindingAToX(@Qualifier("queueA")Queue queue,
                               @Qualifier("exchange_A")DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("XA");
    }

发送消息

package com.example.rabbitmqboot.controller;

import com.example.rabbitmqboot.common.Constant;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Date;

@Slf4j
@RestController
public class MessageController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @RequestMapping("/sendMessage/{message}")
    public ResponseEntity sendMessage(@PathVariable("message") String message) {
        log.info("发送消息: {},当前时间：{}", message, new Date());
        rabbitTemplate.convertAndSend(Constant.DIRECT_EXCHANGE_A, "XA", "交换机Key:XA --" + message);
        return new ResponseEntity<>("ok",HttpStatus.OK);
    }

}

监听获得消息

package com.example.rabbitmqboot.consumer;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.Date;

@Slf4j
@Component
public class Consumer {

    @RabbitListener(queues = "queueA")
    public void getMessage(Message message, Channel channel) {
        try {
            String s = new String(message.getBody(),"UTF-8");
            log.info(" 时间：{} , 消息: {}", new Date(), s);
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
    }

}