SpringBoot整合RabbitMQ

引入SpringBoot的父类

    <parent>
        <groupId>org.springframework.bootgroupId>
        <artifactId>spring-boot-starter-parentartifactId>
        <version>2.2.8.RELEASEversion>
        <relativePath/> 
    parent>
1
2
3
4
5
6

maven的导入

        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-amqpartifactId>
        dependency>
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-webartifactId>
        dependency>
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-testartifactId>
            <scope>testscope>
        dependency>
        <dependency>
            <groupId>org.springframework.amqpgroupId>
            <artifactId>spring-rabbit-testartifactId>
            <scope>testscope>
        dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

6.1 生产者

生产者的controller

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.MessagePropertiesBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.ResponseBody;
import java.nio.charset.StandardCharsets;

@Controller
public class SenderController {

    @Autowired
    private AmqpTemplate template;

    @GetMapping("/send/{msg}")
    @ResponseBody
    public String sendMsg(@PathVariable String msg) {
        MessageProperties properties = MessagePropertiesBuilder.newInstance()
                .setContentEncoding(StandardCharsets.UTF_8.name())
                .setHeader("key", "value").build();

        Message msgData = MessageBuilder.withBody(msg.getBytes(StandardCharsets.UTF_8))
                .andProperties(properties)
                .build();

        template.send("boot.ex", "boot.rk", msgData);

        return "successMsg";
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

生产者的配制

package com.nullnull.learn;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.ExchangeBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class RabbitConfig {

    @Bean
    public Queue queue() {
        return QueueBuilder.nonDurable("boot.queue").build();
    }

    @Bean
    public Exchange exchange() {
        Exchange exchange = ExchangeBuilder.directExchange("boot.ex").build();
        return exchange;
    }

    @Bean
    public Binding binding() {
        return new Binding("boot.queue", Binding.DestinationType.QUEUE, "boot.ex", "boot.rk", null);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

主启动类

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

@SpringBootApplication
public class RabbitApplication {
    public static void main(String[] args) {
        SpringApplication.run(RabbitApplication.class, args);
    }
}
1
2
3
4
5
6
7
8
9

启动SpringBoot工程，即运行RabbitApplication中的main函数

在浏览器中输入测试地址:http://127.0.0.1:8080/send/testmsg

页面将收到响应: successMsg

检查队列的情况

root@nullnull-os mellanox]# rabbitmqctl list_exchanges --formatter pretty_table
Listing exchanges for vhost / ...
┌────────────────────┬─────────┐
│ name               │ type    │
├────────────────────┼─────────┤
│ amq.fanout         │ fanout  │
├────────────────────┼─────────┤
│ ex.anno.fanout     │ fanout  │
├────────────────────┼─────────┤
│ ex.busi.topic      │ topic   │
├────────────────────┼─────────┤
│ amq.rabbitmq.trace │ topic   │
├────────────────────┼─────────┤
│ amq.headers        │ headers │
├────────────────────┼─────────┤
│ amq.topic          │ topic   │
├────────────────────┼─────────┤
│ amq.direct         │ direct  │
├────────────────────┼─────────┤
│ ex.direct          │ direct  │
├────────────────────┼─────────┤
│ boot.ex            │ direct  │
├────────────────────┼─────────┤
│                    │ direct  │
├────────────────────┼─────────┤
│ ex.routing         │ direct  │
├────────────────────┼─────────┤
│ amq.match          │ headers │
└────────────────────┴─────────┘
[root@nullnull-os mellanox]# rabbitmqctl list_queues --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌────────────┬──────────┐
│ name       │ messages │
├────────────┼──────────┤
│ queue.msg  │ 0        │
├────────────┼──────────┤
│ queue.anno │ 0        │
├────────────┼──────────┤
│ boot.queue │ 1        │
└────────────┴──────────┘
[root@nullnull-os mellanox]# rabbitmqctl list_bindings --formatter pretty_table
Listing bindings for vhost /...
┌────────────────┬─────────────┬──────────────────┬──────────────────┬──────────────┬───────────┐
│ source_name    │ source_kind │ destination_name │ destination_kind │ routing_key  │ arguments │
├────────────────┼─────────────┼──────────────────┼──────────────────┼──────────────┼───────────┤
│                │ exchange    │ queue.msg        │ queue            │ queue.msg    │           │
├────────────────┼─────────────┼──────────────────┼──────────────────┼──────────────┼───────────┤
│                │ exchange    │ queue.anno       │ queue            │ queue.anno   │           │
├────────────────┼─────────────┼──────────────────┼──────────────────┼──────────────┼───────────┤
│                │ exchange    │ boot.queue       │ queue            │ boot.queue   │           │
├────────────────┼─────────────┼──────────────────┼──────────────────┼──────────────┼───────────┤
│ boot.ex        │ exchange    │ boot.queue       │ queue            │ boot.rk      │           │
├────────────────┼─────────────┼──────────────────┼──────────────────┼──────────────┼───────────┤
│ ex.anno.fanout │ exchange    │ queue.anno       │ queue            │ routing.anno │           │
├────────────────┼─────────────┼──────────────────┼──────────────────┼──────────────┼───────────┤
│ ex.direct      │ exchange    │ queue.msg        │ queue            │ routing.q1   │           │
└────────────────┴─────────────┴──────────────────┴──────────────────┴──────────────┴───────────┘
[root@nullnull-os mellanox]# 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

观察发现，数据已经成功的发送至了RabbitMQ中了，至此生产者编写完成。

6.2 消费者

消费队列的配制文件

spring:
  application:
    name: springboot_rabbitmq

  rabbitmq:
    host: node1
    virtual-host: /
    username: root
    password: 123456
    port: 5672
1
2
3
4
5
6
7
8
9
10

队列配制信息

import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

    @Bean
    public Queue queue() {
        return QueueBuilder.nonDurable("boot.queue").build();
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

监听器的代码

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

@Component
public class MessageListener {


    //@RabbitListener(queues = "boot.queue")
    //public void getMsg(Message msg) {
    //    String headValue = msg.getMessageProperties().getHeader("key");
    //    String getValue = new String(msg.getBody(), StandardCharsets.UTF_8);
    //    System.out.println("接收到的消息头:" + headValue);
    //    System.out.println("接收到的消息:" + getValue);
    //}


    @RabbitListener(queues = "boot.queue")
    public void getMsgPay(@Payload String msg, @Header("key") String keyValue) {
        System.out.println("接收到的消息头:" + keyValue);
        System.out.println("接收到的消息:" + msg);
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

启动消费都后，观察控制台，便会有如下的输出:

2023-08-20 15:17:22.680  INFO 21720 --- [           main] o.s.amqp.rabbit.core.RabbitAdmin         : Auto-declaring a non-durable, auto-delete, or exclusive Queue (boot.queue) durable:false, auto-delete:false, exclusive:false. It will be redeclared if the broker stops and is restarted while the connection factory is alive, but all messages will be lost.
接收到的消息头value
接收到的消息:testmsg321
1
2
3

至此，SpringBoot消费都也已经成功的集成。