RabbitMQ支持消息的模式（SpringBoot）

发布于：2023-02-13 ⋅ 阅读:(619) ⋅ 点赞:(0)

使用springboot完成rabbitmq的消费模式

整体核心

在这里插入图片描述

准备工作

1：创建生产者工程：sspringboot-rabbitmq-fanout-producer

2：创建消费者工程：springboot-rabbitmq-fanout-consumer

3：引入spring-boot-rabbitmq的依赖

<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>

4：在在application.yml进行配置，生产者8080，消费者8081

# 服务端口
server:
  port: 8080
# 配置rabbitmq服务
spring:
  rabbitmq:
    username: admin
    password: admin
    virtual-host: /
    # 需要开启mq的服务节点
    host: 47.104.141.27
    port: 5672

接下来使用springboot实现消费模式-Fanout-direct-topic

5.定义订单的生产者

生产者负责生产订单，并调用消息队列调用其他服务

package com.chif.rabbitmq.springbootorderrabbitmqproducer.service;

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.UUID;

@Service
public class OrderService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void makeOrder(String userId, String product, int num) {
        //1.根据商品id查询库存是否重组
        //2.保存订单
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生产成功"+orderId);
        //3.通过MQ来完成消息的分发
        /*
         *  参数1：交换机
         *  参数2：路由key/queue队列名称
         */
        String exchangeName = "fanout_order_exchange";
        String routingKey = "";
        rabbitTemplate.convertAndSend(exchangeName, routingKey, orderId);
        System.out.println("producer发送订单信息成功");
    }


    public void makeOrderDirect(String userId, String productId, int num) {
        //1.根据商品id查询库存是否重组
        //2.保存订单
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生产成功"+orderId);
        //3.通过MQ来完成消息的分发
        /*
         *  参数1：交换机
         *  参数2：路由key/queue队列名称
         */
        String exchangeName = "direct_order_exchange";
        rabbitTemplate.convertAndSend(exchangeName, "email", orderId);
        rabbitTemplate.convertAndSend(exchangeName, "duanxin", orderId);
    }


    public void makeOrderTopic(String userId, String productId, int num) {
        //1.根据商品id查询库存是否重组
        //2.保存订单
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生产成功"+orderId);
        //3.通过MQ来完成消息的分发
        /*
         *  参数1：交换机
         *  参数2：路由key/queue队列名称
         */
        String exchangeName = "topic_order_exchange";
        String routingKey="com.chif";
        rabbitTemplate.convertAndSend(exchangeName, routingKey, orderId);
    }


    /**
     * 模拟用户下单
     * @param userId
     * @param productId
     * @param num
     */
    public void makeOrderTTL(String userId, String productId, int num) {
        //1.根据商品id查询库存是否重组
        //2.保存订单
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生产成功"+orderId);
        //3.通过MQ来完成消息的分发
        /*
         *  参数1：交换机
         *  参数2：路由key/queue队列名称
         */
        String exchangeName = "ttl_order_exchange";
        String routingKey="ttl";
        rabbitTemplate.convertAndSend(exchangeName, routingKey, orderId);
    }

    /**
     * 模拟用户下单
     * @param userId
     * @param productId
     * @param num
     */
    public void makeOrderTTLMessage(String userId, String productId, int num) {
        //1.根据商品id查询库存是否重组
        //2.保存订单
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生产成功"+orderId);
        //3.通过MQ来完成消息的分发
        /*
         *  参数1：交换机
         *  参数2：路由key/queue队列名称
         */
        String exchangeName = "ttl_order_exchange";
        String routingKey="TTLMsg";

        //给消息设置过期时间
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //这里就是字符串
                message.getMessageProperties().setExpiration("5000");
                message.getMessageProperties().setContentEncoding("UTF-8");
                return message;
            }
        };
        rabbitTemplate.convertAndSend(exchangeName, routingKey, orderId,messagePostProcessor);
    }

}

6.声明交换机和队列并绑定关系

声明交换机和队列并绑定关系建议在消费者代码中声明，因为消费者代码启动的时候去监听队列，如果队列没有声明会报错。

fanout模式

package com.chif.rabbitmq.springbootorderrabbitmqproducer.config;

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.context.annotation.Configuration;

@Configuration
public class RabbitMqConfiguration {

    //1.声明注册fanout模式的交换机
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanout_order_exchange", true, false);
    }

    //2.声明队列 sms.direct.queue,email.direct.queue,duanxin.direct.queue
    @Bean
    public Queue smsQueue() {
        return new Queue("sms.fanout.queue", true);
    }

    @Bean
    public Queue emailQueue() {
        return new Queue("email.fanout.queue", true);
    }

    @Bean
    public Queue duanxinQueue() {
        return new Queue("duanxin.fanout.queue", true);
    }

    //3.完成绑定关系
    @Bean
    public Binding smsBinding() {
        return BindingBuilder.bind(smsQueue()).to(fanoutExchange());
    }

    @Bean
    public Binding emailBinding() {
        return BindingBuilder.bind(emailQueue()).to(fanoutExchange());
    }

    @Bean
    public Binding duanxinBinding() {
        return BindingBuilder.bind(duanxinQueue()).to(fanoutExchange());
    }

}

direct模式

package com.chif.rabbitmq.springbootorderrabbitmqproducer.config;

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

@Configuration
public class DirectRabbitMqConfiguration {

    //1.声明注册fanout模式的交换机
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange("direct_order_exchange", true, false);
    }

    //2.声明队列 sms.direct.queue,email.direct.queue,duanxin.direct.queue
    @Bean
    public Queue smsDirectQueue() {
        return new Queue("sms.direct.queue", true);
    }

    @Bean
    public Queue emailDirectQueue() {
        return new Queue("email.direct.queue", true);
    }

    @Bean
    public Queue duanxinDirectQueue() {
        return new Queue("duanxin.direct.queue", true);
    }

    //3.完成绑定关系
    @Bean
    public Binding smsDirectBinding() {
        return BindingBuilder.bind(smsDirectQueue()).to(directExchange()).with("sms");
    }

    @Bean
    public Binding emailDirectBinding() {
        return BindingBuilder.bind(emailDirectQueue()).to(directExchange()).with("email");
    }

    @Bean
    public Binding duanxinDirectBinding() {
        return BindingBuilder.bind(duanxinDirectQueue()).to(directExchange()).with("duanxin");
    }

}

7.进行消息的分发和测试

package com.chif.rabbitmq.springbootorderrabbitmqproducer;

import com.chif.rabbitmq.springbootorderrabbitmqproducer.service.OrderService;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
class SpringbootOrderRabbitmqProducerApplicationTests {

    @Autowired
    private OrderService orderService;

    @Test
    void contextLoads() {
        orderService.makeOrder("1","1",12);
    }

    @Test
    void DirectContextLoads() {
        orderService.makeOrderDirect("1","1",12);
    }

    @Test
    void TopicContextLoads() {
        orderService.makeOrderTopic("1","1",12);
    }

    @Test
    void TTLContextLoads() {
        orderService.makeOrderTTL("1","1",12);
    }

    @Test
    void TTLMessageLoads() {
        orderService.makeOrderTTLMessage("1","1",12);
    }

}

8：查看和观察web控制台的状况

9：定义消费者

消费者负责调用短信、邮件、sms等服务

fanout模式(duanxin，sms，email)

package com.chif.rabbitmq.springbootorderrabbitmqproducer.service.fanout;

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

@Service
@RabbitListener(queues = {"sms.direct.queue"})
public class FanoutSMSConsumer {

    @RabbitHandler
    public void reviceMessage(String message){
        System.out.println("sms direct-->接收到了订单信息是-->"+message);
    }
}


package com.chif.rabbitmq.springbootorderrabbitmqproducer.service.fanout;

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

@Service
@RabbitListener(queues = {"email.direct.queue"})
public class FanoutEmailConsumer {

    @RabbitHandler
    public void reviceMessage(String message){
        System.out.println("email direct-->接收到了订单信息是-->"+message);
    }
}


package com.chif.rabbitmq.springbootorderrabbitmqproducer.service.fanout;

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

@Service
@RabbitListener(queues = {"duanxin.direct.queue"})
public class FanoutDuanxiConsumer {

    @RabbitHandler
    public void reviceMessage(String message){
        System.out.println("duanxin direct-->接收到了订单信息是-->"+message);
    }
}

direct模式

package com.chif.rabbitmq.springbootorderrabbitmqproducer.service.direct;

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

@Service
@RabbitListener(queues = {"duanxin.direct.queue"})
public class DirectDuanxiConsumer {

    @RabbitHandler
    public void reviceMessage(String message){
        System.out.println("duanxin direct-->接收到了订单信息是-->"+message);
    }
}


package com.chif.rabbitmq.springbootorderrabbitmqproducer.service.direct;

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

@Service
@RabbitListener(queues = {"email.direct.queue"})
public class DirectEmailConsumer {

    @RabbitHandler
    public void reviceMessage(String message){
        System.out.println("email direct-->接收到了订单信息是-->"+message);
    }
}


package com.chif.rabbitmq.springbootorderrabbitmqproducer.service.direct;

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

@Service
@RabbitListener(queues = {"sms.direct.queue"})
public class DirectSMSConsumer {

    @RabbitHandler
    public void reviceMessage(String message){
        System.out.println("sms direct-->接收到了订单信息是-->"+message);
    }
}

topic模式

package com.chif.rabbitmq.springbootorderrabbitmqproducer.service.topic;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Service;

@Service
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "duanxin.topic.queue",durable = "true",autoDelete = "false"),
        exchange = @Exchange(value = "topic_order_exchange",type = ExchangeTypes.TOPIC),
        key = "#.duanxin.#"
))
public class TopicDuanxiConsumer {

    @RabbitHandler
    public void reviceMessage(String message){
        System.out.println("duanxin topic-->接收到了订单信息是-->"+message);
    }
}


package com.chif.rabbitmq.springbootorderrabbitmqproducer.service.topic;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Service;

@Service
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "email.topic.queue",durable = "true",autoDelete = "false"),
        exchange = @Exchange(value = "topic_order_exchange",type = ExchangeTypes.TOPIC),
        key = "*.email.#"
))
public class TopicEmailConsumer {

    @RabbitHandler
    public void reviceMessage(String message){
        System.out.println("email topic-->接收到了订单信息是-->"+message);
    }
}


package com.chif.rabbitmq.springbootorderrabbitmqproducer.service.topic;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Service;

@Service
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "sms.topic.queue",durable = "true",autoDelete = "false"),
        exchange = @Exchange(value = "topic_order_exchange",type = ExchangeTypes.TOPIC),
        key = "com.#"
))
public class TopicSMSConsumer {

    @RabbitHandler
    public void reviceMessage(String message){
        System.out.println("sms topic-->接收到了订单信息是-->"+message);
    }
}
本文含有隐藏内容，请 开通VIP 后查看

网站公告

今日签到

点亮在社区的每一天
去签到

热门文章

最新发布