关于rabbitmq消息推送的小demo

小鱼儿 2023-10-07 22:14 12阅读 0赞

**目录**

一.前言

1.1场景

1.2消息交换机三种形式

二.建设demo工程

2.1 依赖

2.2yml文件指定rabbitmq连接信息

2.3直连型消息链接

--------------------

## 一.前言 ##

### 1.1场景 ###

在我们实际开发中到一个特定的时候是比如工作流到某个状态时, 我们会向某某单位发送消息, 这时就会用到我们的消息推送---rabbitmq

**简单画一下:**

![1f3872f7f7b4425f9df6aa08adcf66ae.png][]

### 1.2消息交换机三种形式 ###

首先我们了解下消息队列是由交换机exchange和队列组合构成的，有三种形式

1.  直连型：一个交换机关联一个队列，指定一个路由key，消息通过交换机名称和路由key发送到指定队列，发送一个，队列里面就多一个消息。
2.  扇型：一个交换机关联多个队列。消息通过交换机名称发送，所有关联了这个交换机的队列都将收到消息，发送一个消息再N个消息队列产生N个一模一样的消息数据。
3.  主题型：一个交换机根据规则关联多个队列。这种类型与扇型的很像，但是主题型会根据动态路由key决定消息的投递到哪个队列。这里的路由规则很像正则表达式。会根据事先设定的路由规则动态将消息投递到队列，可能投递到一个队列也可能投递到多个队列。

## 二.建设demo工程 ##

### 2.1 依赖 ###

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

### 2.2yml文件指定rabbitmq连接信息 ###

server:
      port: 8021
    spring:
      #给项目来个名字
      application:
        name: rabbitmq-provider
      #配置rabbitMq 服务器
      rabbitmq:
        host: localhost
        port: 5672
        #确认消息已发送到交换机(Exchange)
        publisher-confirm-type: correlated
        #确认消息已发送到队列(Queue)
        publisher-returns: true

**注意：**我们需要创建两个工程，一个生产者`producer`、一个消费者`comsumer`，生产者用来生产消息，消费者用来消费生产者将消息投递到rabbitmq中的消息。

两个工程中的pom依赖一样，yml也一样，只需要将`server.port`设置成不同的端口即可。这里我们将生产者设置为`8021`端口，消费者设置为`8022`端口。

### **2.3直连型消息链接** ###

从上面的讲解中我们知道，有交换机`exchange`，有队列`queue`，有路由`routing`，因此我们需要在生产者端将三者关联起来，然后发送消息，这样消费端才能收到消息。

**创建config工作类 绑定关联**

@Configuration
    public class Config {
        public static String directRouting = "directRouting";
        public static String directQueue = "directQueue";
        public static String directExchange = "directExchange";
        @Bean
        public Queue DirectQueue() {
            return new Queue(Config.directQueue,true);  //true 是否持久
        }
        @Bean
        DirectExchange DirectExchange() {
            return new DirectExchange(Config.directExchange);
        }
        @Bean
        Binding bindingDirect() {
    //        BindingBuilder.bind(队列A).to(交换机B).with(路由) 将队列A绑定到交换机B，使用路由C传递消息
            return BindingBuilder.bind(DirectQueue()).to(DirectExchange()).with(directRouting);
        }

**发送消息**

@Autowired
    private RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,这提供了接收/发送等等方法
    
    @GetMapping("/sendDirectMsg")
    public String sendMsg() {
        Map<String,Object> map=new HashMap<String,Object>();
        map.put("id",UUID.randomUUID().toString());
        map.put("data","hello,i am direct msg!");
        map.put("datetime",System.currentTimeMillis());
        //交换机 路由 消息（发送消息的时候不需要管队列，因为队列已经在DirectRabbitConfig中配置了，队列应该是消费者关心的事情）
        rabbitTemplate.convertAndSend(DirectRabbitConfig.directExchange, DirectRabbitConfig.directRouting, map);
        return "ok";
    }

**第二个工程中: 接收消息**

@Component
    @RabbitListener(queues = "directQueue")//监听的队列名称 directQueue，不需要管路由和交换机，因为这些是生产者管理的事情。消费者只需要关心队列即可
    public class DirectReceiver {
        @RabbitHandler
        public void handler(Map testMessage) {
            System.out.println("directReceiver消费者收到消息  : " + testMessage.toString());
        }
    }

[1f3872f7f7b4425f9df6aa08adcf66ae.png]: https://img-blog.csdnimg.cn/1f3872f7f7b4425f9df6aa08adcf66ae.png