Rabbit MQ实战【topic交换器】

逃离我推掉我的手 2023-07-03 06:17 24阅读 0赞

spring boot版本：2.1.10.RELEASE  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9qei0yMDE3LmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70]

本文涉及两个项目 `rabbitmq-topic-provider` 和 `rabbitmq-topic-consumer`，相关依赖和配置相同。

#### 相关依赖 ####

<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>
    
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>

#### 相关配置 ####

#rabbitmq配置
    spring.rabbitmq.host=192.168.xxx.xxx
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=admin
    spring.rabbitmq.password=admin
    spring.rabbitmq.virtual-host=/
    
    #设置交换器(自定义变量)
    mq.config.exchange=log.topic
    
    mq.config.queue.info=log.info
    mq.config.queue.error=log.error
    mq.config.queue.logs=log.msg

### rabbitmq-topic-provider项目 ###

**（1）订单消息发送者**

package com.ebook.rabbitmq;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    
    /** * @author:JZ * @date:2020/2/2 */
    
    @Component
    public class OrderSender { 
    
        @Value("${mq.config.exchange}")
        private String exchange;
    
        @Autowired
        private AmqpTemplate rabbitTemplate;
    
        public void send() { 
            this.rabbitTemplate.convertAndSend(this.exchange, "order.log.debug", "order.log.debug.......");
            this.rabbitTemplate.convertAndSend(this.exchange, "order.log.info", "order.log.info.......");
            this.rabbitTemplate.convertAndSend(this.exchange, "order.log.warn", "order.log.warn.......");
            this.rabbitTemplate.convertAndSend(this.exchange, "order.log.error", "order.log.error.......");
        }
    
    }

**（2）产品消息发送者**

package com.ebook.rabbitmq;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    
    /** * @author:JZ * @date:2020/2/2 */
    
    @Component
    public class ProductSender { 
    
        @Value("${mq.config.exchange}")
        private String exchange;
    
        @Autowired
        private AmqpTemplate rabbitTemplate;
    
        public void send() { 
            this.rabbitTemplate.convertAndSend(this.exchange, "product.log.debug", "product.log.debug.......");
            this.rabbitTemplate.convertAndSend(this.exchange, "product.log.info", "product.log.info.......");
            this.rabbitTemplate.convertAndSend(this.exchange, "product.log.warn", "product.log.warn.......");
            this.rabbitTemplate.convertAndSend(this.exchange, "product.log.error", "product.log.error.......");
        }
    
    }

**（3）用户消息发送者**

package com.ebook.rabbitmq;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    
    import java.util.Date;
    
    /** * @author:JZ * @date:2020/2/2 */
    
    @Component
    public class UserSender { 
    
        @Value("${mq.config.exchange}")
        private String exchange;
    
        @Autowired
        private AmqpTemplate rabbitTemplate;
    
        public void send() { 
            this.rabbitTemplate.convertAndSend(this.exchange, "user.log.debug", "user.log.debug.......");
            this.rabbitTemplate.convertAndSend(this.exchange, "user.log.info", "user.log.info.......");
            this.rabbitTemplate.convertAndSend(this.exchange, "user.log.warn", "user.log.warn.......");
            this.rabbitTemplate.convertAndSend(this.exchange, "user.log.error", "user.log.error.......");
        }
    
    }

**（4）测试类**

import com.ebook.rabbitmq.OrderSender;
    import com.ebook.rabbitmq.ProductSender;
    import com.ebook.rabbitmq.RabbitmqTopicProvider;
    import com.ebook.rabbitmq.UserSender;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    @RunWith(SpringRunner.class)
    @SpringBootTest(classes = RabbitmqTopicProvider.class)
    public class RabbitMqTopicApplicationTests { 
    
    	@Autowired
    	private UserSender userSender;
    
    	@Autowired
    	private ProductSender productSender;
    
    	@Autowired
    	private OrderSender orderSender;
    
    	@Test
    	public void send() throws InterruptedException { 
    		this.userSender.send();
    		this.productSender.send();
    		this.orderSender.send();
    	}
    
    }

### rabbitmq-topic-consumer项目 ###

**（1）Info信息接收者**

package com.ebook.rabbitmq;
    
    import org.springframework.amqp.core.ExchangeTypes;
    import org.springframework.amqp.rabbit.annotation.*;
    import org.springframework.stereotype.Component;
    
    /** * @author:JZ * @date:2020/2/2 */
    
    @Component
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "${mq.config.queue.info}", autoDelete = "true"),//autoDelete表示队列为临时队列
            exchange = @Exchange(value = "${mq.config.exchange}", type = ExchangeTypes.TOPIC),
            key = "*.log.info"
        ))
    public class InfoReciver { 
    
        @RabbitHandler
        public void process(String msg) { 
            System.out.println("接收到INFO日志：" + msg);
        }
    
    }

**（2）Error信息接收者**

package com.ebook.rabbitmq;
    
    import org.springframework.amqp.core.ExchangeTypes;
    import org.springframework.amqp.rabbit.annotation.*;
    import org.springframework.stereotype.Component;
    
    /** * @author:JZ * @date:2020/2/2 */
    
    @Component
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "${mq.config.queue.error}", autoDelete = "true"),//autoDelete表示队列为临时队列
            exchange = @Exchange(value = "${mq.config.exchange}", type = ExchangeTypes.TOPIC),
            key = "*.log.error"
        ))
    public class ErrorReciver { 
    
        @RabbitHandler
        public void process(String msg) { 
            System.out.println("接收到ERROR日志：" + msg);
        }
    
    }

**（3）全部消息接收者**

package com.ebook.rabbitmq;
    
    import org.springframework.amqp.core.ExchangeTypes;
    import org.springframework.amqp.rabbit.annotation.*;
    import org.springframework.stereotype.Component;
    
    /** * @author:JZ * @date:2020/2/2 */
    
    @Component
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "${mq.config.queue.logs}", autoDelete = "true"),//autoDelete表示队列为临时队列
            exchange = @Exchange(value = "${mq.config.exchange}", type = ExchangeTypes.TOPIC),
            key = "*.log.*"
        ))
    public class LogsReciver { 
    
        @RabbitHandler
        public void process(String msg) { 
            System.out.println("接收到ALL日志：" + msg);
        }
    
    }

#### 运行结果 ####

运行 `rabbitmq-topic-consumer` 项目，并运行测试类。

结果：

接收到INFO日志：user.log.info.......
    接收到INFO日志：product.log.info.......
    接收到INFO日志：order.log.info.......
    接收到ERROR日志：user.log.error.......
    接收到ERROR日志：product.log.error.......
    接收到ERROR日志：order.log.error.......
    接收到ALL日志：user.log.debug.......
    接收到ALL日志：user.log.info.......
    接收到ALL日志：user.log.warn.......
    接收到ALL日志：user.log.error.......
    接收到ALL日志：product.log.debug.......
    接收到ALL日志：product.log.info.......
    接收到ALL日志：product.log.warn.......
    接收到ALL日志：product.log.error.......
    接收到ALL日志：order.log.debug.......
    接收到ALL日志：order.log.info.......
    接收到ALL日志：order.log.warn.......
    接收到ALL日志：order.log.error.......

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9qei0yMDE3LmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2020020313073361.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9qei0yMDE3LmJsb2cuY3Nkbi5uZXQ=,size_16,color_FFFFFF,t_70