Rabbit与springboot整合-1

news2025/1/14 18:05:20

1、整体结构

2、pom引入

3、配置文件

4、代码

公共类

controller类

JSON转换类

监听-接收发送消息类

1、整体结构

2、pom引入

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

3、配置文件

4、代码

公共类

controller类

import com.rabbitmqprovider.commons.CommonUtils;
import com.rabbitmqprovider.vo.User;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class RabbitMQController {
    Logger LOG = LoggerFactory.getLogger(RabbitMQController.class);
    @Autowired
    AmqpAdmin amqpAdmin;
    /**
     * 交换机创建
     */
    @GetMapping("/createExchange")
    public void createExchange(){
        // 第一个参数为交换机名字，第二个参数为是否持久化，第三个参数为不使用交换机时删除
        DirectExchange directExchange = new DirectExchange(CommonUtils.dirExchange,true,false);
        amqpAdmin.declareExchange(directExchange);
        System.out.println("交换机创建成功");
    }
    /**
     * 绑定队列
     */
    @GetMapping("/createQueue")
    public void createQueue() {
        /**
         * 第一个参数为队列名字，
         * 第二个参数为是否持久化，
         * 第三个参数为是否排他（true：一个连接只能有一个队列，false：一个连接可以有多个（推荐））
         * 第四个参数为不使用队列时自动删除
         */
        Queue queue = new Queue(CommonUtils.queueStr,true,false,false);
        amqpAdmin.declareQueue(queue);
        System.out.println("队列创建成功");
    }

    /**
     * 绑定交换机和队列
     */
    @GetMapping("/createBinding")
    public void createBinding() {
        /**
         * 第一个参数为目的地，就是交换机或者队列的名字
         * 第二个参数为目的地类型，交换机还是队列
         * 第三个参数为交换机，QUEUE-队列  EXCHANGE-交换机
         * 第四个参数为路由键，匹配的名称
         */
        Binding binding = new Binding(CommonUtils.queueStr,
                Binding.DestinationType.QUEUE,
                CommonUtils.dirExchange,CommonUtils.routingKey,null);
        amqpAdmin.declareBinding(binding);
        System.out.println("绑定成功");
    }

    @Autowired
    RabbitTemplate rabbitTemplate;

    /**
     * 发送消息
     * 结果:"这是一条消息"
     */
    @GetMapping("/sendMessageTest")
    public String sendMessageTest(){
        // 消息类型为object 发送对象也是可以的
        String msg = "这是一条消息";
        // 第一个参数为发送消息到那个交换机上，第二个是发送的路由键（交换机进行需要符合绑定的队列），第三个参数为发送的消息
        rabbitTemplate.convertAndSend(CommonUtils.dirExchange,CommonUtils.routingKey,msg);
        System.out.println("消息发送成功");
        return "发送成功;发送内容为:"+msg;
    }

    /**
     * 发送消息
     * 结果：{"name":"张三、李四","age":22}
     */
    @GetMapping("/sendMessageUserTest")
    public void sendMessageUserTest(){
        User user = new User();
        user.setAge(22);
        user.setName("张三、李四");
        // 第一个参数为发送消息到那个交换机上，第二个是发送的路由键（交换机进行需要符合绑定的队列），第三个参数为发送的消息
        rabbitTemplate.convertAndSend(CommonUtils.dirExchange,CommonUtils.routingKey,user);
        System.out.println("消息发送成功");
    }

}