RabbitMQ (四) --------- SpringBoot 集成 RabbitMQ

一、创建消息生产者工程
二、创建消息接收者工程
三、Direct 模式消息发送和接收
- 1. 编写 Direct 模式的消息发送
- 2. 编写 Direct 模式的消息接收
四、Fanout 模式消息发送和接收
- 1. 编写 Fanout 模式的消息发送
- 2. 编写 Fanout 模式的消息接收
五、Topic 模式消息发送和接收
- 1. 编写 Topic 模式消息发送
- 2. 编写 Topic 模式消息接收

一、创建消息生产者工程

创建模块 rabbitmq-springboot-send

配置模块 rabbitmq-springboot-send 的 application.properties 文件添加对 RabbitMQ 的集成

#配置RabbitMQ链接信息
#配置RabbitMQ服务器的IP地址
spring.rabbitmq.host=192.168.160.133
#配置RabbitMQ服务器的端口
spring.rabbitmq.port=5672
#配置RabbitMQ服务器的访问账号
spring.rabbitmq.username=root
#配置RabbitMQ服务器的访问密码
spring.rabbitmq.password=aszhuo

二、创建消息接收者工程

创建模块 rabbitmq-springboot-receive

配置模块 rabbitmq-springboot-receive 的 application.properties 文件添加对 RabbitMQ 的集成

三、Direct 模式消息发送和接收 1. 编写 Direct 模式的消息发送

在 rabbitmq-springboot-send 模块中创建类，com.fancy.rabbitmq.direct.Send

package com.fancy.rabbitmqspringbootsend.direct;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Service
public class Send {

    // 自动注入 Amqp 的模板对象

    
    @Resource
    private AmqpTemplate template;

    public void send() {
        //发送消息到队列
        //参数 1 为消息存放的交换机名称 （需要事前创建）
        //参数 2 为RoutingKey，接收者需要根据这个key精准接收消息
        //参数 3 为具体存入队列中的消息数据
        template.convertAndSend("BootDirectExchange", "BootRouting", "SpringBootDirect");
    }
}

创建 Amqp 配置类 com.fancy.rabbitmq.config.AmqpConfig

package com.fancy.rabbitmqspringbootsend.config;

import org.springframework.amqp.core.DirectExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class AmqpConfig {
    // @Bean 用于模拟 Spring 配置文件中的  标签 , 用于创建 名字为
    // BootDirectExchange 的交换机
    @Bean
    public DirectExchange myChange() {
        return new DirectExchange("BootDirectExchange");
    }
}

配置主启动类，测试运行 Direct 消息发送

package com.fancy.rabbitmqspringbootsend;

import com.fancy.rabbitmqspringbootsend.direct.Send;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationContext;

@SpringBootApplication
public class RabbitmqSpringbootSendApplication {

    public static void main(String[] args) {
        ApplicationContext ac = SpringApplication.run(RabbitmqSpringbootSendApplication.class, args);
        Send send = (Send)ac.getBean("send");
        send.send();
    }

}

2. 编写 Direct 模式的消息接收

在 rabbitmq-springboot-receive 模块中创建类，com.fancy.rabbitmq.direct.Receive

package com.fancy.rabbitmqspringbootsend.direct;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Service
public class Send {

    // 自动注入 Amqp 的模板对象

    
    @Resource
    private AmqpTemplate template;

    public void send() {
        //发送消息到队列
        //参数 1 为消息存放的交换机名称 （需要事前创建）
        //参数 2 为RoutingKey，接收者需要根据这个key精准接收消息
        //参数 3 为具体存入队列中的消息数据
        template.convertAndSend("BootDirectExchange", "BootRouting", "SpringBootDirect");
    }
}

创建 Amqp 配置类 com.fancy.rabbitmq.config.AmqpConfig

package com.fancy.rabbitmqspringbootreceive.config;


import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;



@Configuration
public class AmqpConfig {

    //创建一个名字为  myQueueDirect 的队列
    @Bean
    public Queue queue() {
        return new Queue("myQueueDirect");
    }

    //创建一个名字为 BootDirectExchange 的交换机
    @Bean
    public Exchange myChange() {
        return new DirectExchange("BootDirectExchange");
    }

    // 将队列绑定到交换机
    @Bean("binding")
    // 参数1 为 自定义队列对象, 参数名为 queue 为 自定义队列对象 bean 的 id
    // 参数2 为 自定义的交换机, 参数名为 myChange 为自定义交换机 bean 的 id
    public Binding binding(Queue queue, Exchange exchange) {
        // 将队列绑定到交换机, 参数 BootRouting 为 RoutingKey
        return BindingBuilder.bind(queue).to(exchange).with("BootRouting").noargs();

    }
    
}

运行测试 Receive 消息接收，编写 Application.java 类

package com.fancy.rabbitmqspringbootreceive;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class RabbitmqSpringbootReceiveApplication {

    public static void main(String[] args) {
        SpringApplication.run(RabbitmqSpringbootReceiveApplication.class, args);
    }

}

四、Fanout 模式消息发送和接收 1. 编写 Fanout 模式的消息发送

在 rabbitmq-springboot-send模块中创建类，com.fancy.rabbitmq.fanout.Send

package com.fancy.rabbitmqspringbootsend.fanout;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Service
public class Send {

    @Resource
    private AmqpTemplate template;

    public void fanoutSend() {
        template.convertAndSend("BootFanoutExchange", "", "SpringBootFanout");
    }
}

修改 Amqp 配置类， com.fancy.rabbitmq.config.AmqpConfig，增加以下内容

@Bean
public FanoutExchange fanoutExchange() {
    // 创建一个基于 Fanout 的交换机, 名字为 BootFanoutExchange 
    return new FanoutExchange("BootFanoutExchange");
}

运行测试 Direct 消息发送，编写 Application.java 类

package com.fancy.rabbitmqspringbootsend;

import com.fancy.rabbitmqspringbootsend.fanout.Send;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationContext;

@SpringBootApplication
public class RabbitmqSpringbootSendApplication {

    public static void main(String[] args) {
        ApplicationContext ac = SpringApplication.run(RabbitmqSpringbootSendApplication.class, args);
        Send send = (Send)ac.getBean("send");
        send.fanoutSend();
    }

}

2. 编写 Fanout 模式的消息接收

在 rabbitmq-springboot-send 模块中创建类，com.fancy.fanout.Send

package com.fancy.rabbitmqspringbootreceive.fanout;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Service
public class Receive {

    @RabbitListener(queues = "fanoutQueue")
    public void fanoutRecevie(String message) {
        System.out.println("Boot 的 Fanout 消息 ----" + message);
    }
    
}

修改 Amqp 配置类 com.fancy.rabbitmq.config.AmqpConfig，增加以下内容

@Bean 
public Queue fanoutQueue() {
    return new Queue("fanoutQueue");
}

@Bean 
public FanoutExchange fanoutExchange() {
    return new FanoutExchange("BootFanoutExchange");
}

@Bean 
public Binding fanoutBinding(Queue fanoutQueue, FanoutExchange fanoutExchange) {
    // 将队列绑定到指定的交换机上
    // 参数 1 为 指定的队列对象
    // 参数 2 为 指定的交换机对象
    return BindingBuilder.bind(fanoutQueue).to(fanoutExchange);
}

五、Topic 模式消息发送和接收 1. 编写 Topic 模式消息发送

在 rabbitmq-springboot-send 模块中创建类，com.fancy.rabbitmq.topic.Send

package com.fancy.rabbitmqspringbootsend.topic;

import org.springframework.amqp.core.AmqpTemplate;

import javax.annotation.Resource;

public class Send {
    
    @Resource
    private AmqpTemplate template;
    
    public void topicSend() {
        template.convertAndSend("BootTopicExchange", "Boot.text", "SpringBootTopic");
    }
}

修改 Amqp 配置类com.fancy.rabbitmq.config.AmqpConfig，增加以下内容

@Bean 
public TopicExchange topicExchange() {
    return new TopicExchange("BootTopicExchange");
}

2. 编写 Topic 模式消息接收

在 rabbitmq-springboot-receive 模块中创建类，com.fancy.rabbitmq.topic.Receive

package com.fancy.rabbitmqspringbootreceive.topic;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

@Service
public class Receive {

    @RabbitListener(queues = "topicQueue")
    public void topicReceive(String message) {
        System.out.println("Boot 的 topic 消息 1" + message);
    }

    @RabbitListener(queues = "topicQueue2")
    public void topicReceive2(String message) {
        System.out.println("Boot 的 topic 消息 2 " + message);
    }
}

修改 Amqp 配置类 com.fancy.rabbitmq.config.AmqpConfig，增加以下内容

@Bean
public TopicExchange TopicExchange() {
    return new TopicExchange("BootTopicExchange");
}

@Bean
public Queue topicQueue() {
    return new Queue("topicQueue");
}

@Bean
public Queue topicQueue2() {
    return new Queue("topicQueue2");
}

@Bean
public Binding topicBinding(Queue topicQueue, TopicExchange topicExchange) {
    // 为RoutingKey的匹配规则, #.test表示 可以接收以任意路径靠头的但是必须以 test 结尾的队列
    return BindingBuilder.bind(topicQueue).to(TopicExchange()).with("#.text");
}

RabbitMQ (四) --------- SpringBoot 集成 RabbitMQ

Java相关栏目本月热门文章