0


SpringAMQP使用管理RabbitMQ的五种消息模型

RabbitMQ概念: RabbitMQ是一个开源的消息代理(Message Broker)系统,实现了高级消息队列协议(AMQP,Advanced Message Queuing Protocol)版本0-9-1。它提供了可靠的消息传递、路由、持久化存储、集群支持等功能,可以用于分布式系统之间进行异步解耦通信,确保系统的松耦合和扩展性。

SpringAMQP概念: SpringAMQP是Spring对AMQP的实现定义了一套API规范,AMQP是一个消息发送和接受的协议

使用SpringAMQ实现五种消息队列模型

1.普通队列

2.工作队列(WorkQueue)

发布订阅=>根据交换机的不同分为三种

3.订阅模型之Fanout(广播)

4.订阅模型之Direct(路由)

5.订阅模型之Topic(话题)

使用前导:

1.在生产者和消费者项目上分别导入RabbitMQ依赖

  <!--AMQP依赖,包含RabbitMQ-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

2.同样在yaml文件上分别配置MQ地址

一、普通队列

特点:生产者直接将消息发送到一个特定的队列,消费者从该队列接收并处理消息。每个消息都会被一个且只有一个消费者消费

实现思路:

1.定义生产者

RabbitTemplate

是一个核心类,它提供了发送和接收消息到RabbitMQ的基本操作

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    //注入rabbitMQ模板
    @Autowired
    RabbitTemplate rabbitTemplate;
    @Test   //普通队列
    public void sendMessage1(){
        //声明队列名称
        String queueName="simple.queue";
        //设置发送的消息内容
        String sendMessage="明天会更好";
        //发送消息
        rabbitTemplate.convertAndSend(queueName,sendMessage);

    }
}

2.定义消费者

@RabbitListener(queues = "simple.queue")

是Spring AMQP框架中的注解,用于声明一个方法作为消息处理器,监听名为"simple.queue"的RabbitMQ队列。

当在类的方法上使用此注解时,Spring AMQP会自动配置并启动一个消息监听容器来监听指定队列。当"simple.queue"中有新的消息到达时,该注解标记的方法会被调用,并将消息体作为参数传递给该方法进行处理。

@Component
public class SpringListener {
    //普通队列
    @RabbitListener(queues = "simple.queue")
    public void ListenerMessage(String MSG){
        System.out.println(MSG);
    }
}

二、工作队列(WorkQueue)

特点:多个消费者可以从同一个队列中获取消息,但只能被一个消费者消费一旦消费完就会从队列中删除

WorkQueue解决消息预取分配不均:

(例如有一百条消息 两个消费者平分消息 都获得50条消息 但没有考虑到消费者的处理消息能力,消费者1可以一秒处理50条消息 而消费者2一秒只能处理一条消息 导致消费者1处理完了没事干而消费者2迟迟无法完成任务)

解决办法:在yaml文件中设置预取限额为1 处理完后才能继续获取消息 实现能者多劳加快消息的处理效率

实现思路:

1.定义生产者

  @Test   //WorkQueue模型
    public void sendMessage2() throws InterruptedException {
        //WorkQueue模型 只能被一个消费者消费
        String queueName="simple.queue";
        String sendMessage="明天会更好__";
        for (int i =0 ;i<50;i++){
            rabbitTemplate.convertAndSend(queueName,sendMessage+1);
            Thread.sleep(20);
        }

    }

2.定义消费者

   @RabbitListener(queues = "simple.queue")
    public void WorkQueues1(String MSG) throws InterruptedException {
        System.out.println("消费者1接收到的消息---["+MSG+"]"+ LocalDateTime.now());
         Thread.sleep(20);
    }
    @RabbitListener(queues = "simple.queue")
    public void WorkQueues2(String MSG) throws InterruptedException {
        System.err.println("消费者2接收到的消息---["+MSG+"]"+ LocalDateTime.now());
        Thread.sleep(200);
    }

三、订阅模型之Fanout(广播)

特点:生产者将消息发送到交换机(Exchange),而不是直接发送到队列。交换机会根据绑定规则将消息路由与之绑定的队列。

实现思路(使用配置类声明队列和交换机实现):

1.在消费者服务中在配置类声明队列、交换机,将两者绑定

@Configuration
public class FanoutConfig {
    // itcast.fanout交换机
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("itcast.fanout");
    }

    // fanout.queue1
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }

    // 绑定队列1到交换机
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
        return BindingBuilder
                .bind(fanoutQueue1)
                .to(fanoutExchange);
    }
    // fanout.queue2
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }

    // 绑定队列2到交换机
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
        return BindingBuilder
                .bind(fanoutQueue2)
                .to(fanoutExchange);
    }
}

2.在消费者服务中编写消费者方法

    @Component
    public class SpringListener {
    //订阅模型之Fanout(广播)  通过配置类配置交换机和队列实现
    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg) {
        System.out.println("消费者接收到fanout.queue1的消息:【" + msg + "】");
    }
    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String msg) {
        System.out.println("消费者接收到fanout.queue2的消息:【" + msg + "】");
    }
}

3.在生产者中编写测试方法发送消息

   @Test  //订阅模型之Fanout(广播)
    public void sendMessage3(){
        //交换机名称
        String ExchangeName="itcast.fanout";
        //发送的消息
        String sendMessage="hello everyone!!!";

        rabbitTemplate.convertAndSend(ExchangeName,"",sendMessage);

    }

四、订阅模型之Direct(路由)

特点:交换机会根据绑定规则将消息路由到与之绑定的队列。所有绑定到同一交换机并且匹配路由键的队列都将收到消息,多个消费者可以同时订阅并处理来自同一主题的消息

详解:

每一个Queue都会与Exchange设置一个BingdingKey

生产者发布消息时需要指定消息的RoutingKey

Exchange将消息路由到BingKey与RoutingKey一致的队列

当Queue设置了一样的key那么它的效果就会广播一样了

实现思路(使用@RabbitListener注解实现交换机和队列的声明 因为通过Config配置Bean太繁琐):

1.利用@RabbitListener声明Exchange、Queue、Key(声明后运行代码即会自动创建对应的交换机和队列)

并编写消费者方法listenDirectQueue1和listenDirectQueue2分别监听direct.queue1和direct.queue2队列

    //订阅模型之Direct(路由) 通过@RabbitListener注解配置交换机和队列直接实现
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
            key = {"red", "blue"}
    ))
    public void listenDirectQueue1(String msg){
        System.out.println("消费者接收到direct.queue1的消息:【" + msg + "】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2"),
            exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
            key = {"red", "yellow"}
    ))
    public void listenDirectQueue2(String msg){
        System.out.println("消费者接收到direct.queue2的消息:【" + msg + "】");
    }

2.在生产者中编写测试方法发送消息


    @Test  //订阅模型之Direct(路由)
    public void sendMessage4(){
        //交换机名称
        String ExchangeName="itcast.direct";
        //发送的消息
        String sendMessage="hello blue!!!";

        rabbitTemplate.convertAndSend(ExchangeName,"blue",sendMessage);

    }

五、订阅模型之Topic(话题)

**特点:订阅模型之Topic(话题) 与Direct(路由)类似 但是Direct的key只能是多个单词的列表 **

而topic可以使用通配符且用.划分 #表示0个或多个单词 如图所示

实现思路:

1.利用@RabbitListener声明Exchange、Queue、key

在消费者服务中编写两个消费者方法分别监听定义的两个队列

 
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
            key = "china.#"   //消费所有跟中国有关的消息
    ))
    public void listenTopicQueue1(String msg){
        System.out.println("消费者接收到topic.queue1的消息:【" + msg + "】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue2"),
            exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
            key = "#.news"    //消费所有新闻有关的消息
    ))
    public void listenTopicQueue2(String msg){
        System.out.println("消费者接收到topic.queue2的消息:【" + msg + "】");
    }
}

2.在生产者中编写发送消息的测试方法

 @Test  //订阅模型之Topic(话题)
    public void sendMessage5(){
        //交换机名称
        String ExchangeName="itcast.topic";
        //发送的消息
        String sendMessage="hello word!!!";

        rabbitTemplate.convertAndSend(ExchangeName,"china.new",sendMessage);

    }

本文转载自: https://blog.csdn.net/m0_64595627/article/details/136562430
版权归原作者 慕名心 所有, 如有侵权,请联系我们删除。

“SpringAMQP使用管理RabbitMQ的五种消息模型”的评论:

还没有评论