跳至主要內容

15. SpringBoot集成RabbitMQ

PangHu大约 2 分钟

本文来讲解SpringBoot集成RabbitMQ，代码已上传到Github，有兴趣的同学可以下载来看看：https://github.com/ylw-github/SpringBoot-RabbitMQ-Demo.git

1. 生产者

1.新建生产者maven项目RabbitMQ-Producer：

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.0.0.RELEASE</version>
</parent>
<dependencies>

    <!-- springboot-web组件 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!-- 添加springboot对amqp的支持 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-lang3</artifactId>
    </dependency>
    <!--fastjson -->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.49</version>
    </dependency>
</dependencies>

2.application.yml:

server:
  port: 8081
spring:
  rabbitmq:
    ####连接地址
    host: 127.0.0.1
    ####端口号
    port: 5672
    ####账号
    username: guest
    ####密码
    password: guest
    ### 地址
    virtual-host: /

3.交换机绑定队列

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

@Component
public class FanoutConfig {

    // 邮件队列
    private String FANOUTEMAIL_QUEUE = “fanout_eamil_queue”;

    // 短信队列
    private String FANOUT_SMS_QUEUE = “fanout_sms_queue”;
    // 短信队列
    private String EXCHANGE_NAME = “fanoutExchange”;

    // 1.定义队列邮件
    @Bean
    public Queue fanOutEamilQueue() {
        return new Queue(FANOUT_EMAIL_QUEUE);
    }

    @Bean
    public Queue fanOutSmsQueue() {
        return new Queue(FANOUT_SMS_QUEUE);
    }

    // 2.定义交换机
    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange(EXCHANGE_NAME);
    }

    // 3.队列与交换机绑定邮件队列
    @Bean
    Binding bindingExchangeEamil(Queue fanOutEamilQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(fanOutEamilQueue).to(fanoutExchange);
    }

    // 4.队列与交换机绑定短信队列
    @Bean
    Binding bindingExchangeSms(Queue fanOutSmsQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(fanOutSmsQueue).to(fanoutExchange);
    }
}

4.生产者

@Component
public class FanoutProducer {
    @Autowired
    private AmqpTemplate amqpTemplate;

    public void send(String queueName) {
        String msg = “my_fanout_msg:” + new Date();
        System.out.println(msg + “:” + msg);
        amqpTemplate.convertAndSend(queueName, msg);
    }
}

5.控制层调用代码

@RestController
public class ProducerController {
    @Autowired
    private FanoutProducer fanoutProducer;

    @RequestMapping(“/sendFanout”)
    public String sendFanout(String queueName) {
        fanoutProducer.send(queueName);
        return “success”;
    }
}

2. 消费者

1.添加maven依赖

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.0.0.RELEASE</version>
</parent>
<dependencies>

<!— springboot-web组件 —>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!— 添加springboot对amqp的支持 —>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-lang3</artifactId>
</dependency>
<!—fastjson —>
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.2.49</version>
</dependency>
</dependencies>

2.application.yml

server:
  port: 8082
spring:
  rabbitmq:
    ####连接地址
    host: 127.0.0.1
    ####端口号
    port: 5672
    ####账号
    username: guest
    ####密码
    password: guest
    ### 地址
    virtual-host: /

3.邮件消费者

@Component
@RabbitListener(queues = “fanout_eamil_queue”)
public class FanoutEamilConsumer {
    @RabbitHandler
    public void process(String msg) throws Exception {
        System.out.println(“邮件消费者获取生产者消息msg:” + msg);
    }
}

4.短信消费者

@Component
@RabbitListener(queues = “fanout_sms_queue”)
public class FanoutSmsConsumer {

    @RabbitHandler
    public void process(String msg) {
        System.out.println(“短信消费者获取生产者消息msg:” + msg);
    }
}

3. 测试

1.启动生产者项目，浏览器访问： http://localhost:8081/sendFanout

img
img

可以在交换机列表看到：

消息中间件系列教程（13） -RabbitMQ-SpringBoot集成RabbitMQ_# 消息中间件_02
消息中间件系列教程（13） -RabbitMQ-SpringBoot集成RabbitMQ_# 消息中间件_02

2.启动消费者项目，可以两个队列已经注册列表：

消息中间件系列教程（13） -RabbitMQ-SpringBoot集成RabbitMQ_# 消息中间件_03
消息中间件系列教程（13） -RabbitMQ-SpringBoot集成RabbitMQ_# 消息中间件_03

3.生产者发送消息，浏览器访问： http://localhost:8081/sendFanout?queueName=fanout_eamil_queue，可以看到邮件消费者消费信息：

img
img

浏览器再次访问： http://localhost:8081/sendFanout?queueName=fanout_sms_queue，可以看到短信消费者消费信息。

img
img