AI 助理
备案控制台
开发者社区 云计算 文章 正文

AMQP-RabbitMQ/5/主题模式

2019-06-08 929
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 5. 主题模式 Topics根据主题接收消息routing_key - 它必须是由英文点.分隔的单词列表*可以替代一个单词。＃可以替换零个或多个单词。图示个人理解与路由模式基本相同，不同的点只是路由模式需要Queue订阅的路由键与消息的路由键完全匹配，Exchange才会将消息发送给Queue。

5. 主题模式 Topics

根据主题接收消息

routing_key - 它必须是由英文点.分隔的单词列表

  • *可以替代一个单词。
  • ＃可以替换零个或多个单词。
  • 图示

主题模式

个人理解

  • 与路由模式基本相同，不同的点只是路由模式需要Queue订阅的路由键与消息的路由键完全匹配，Exchange才会将消息发送给Queue。
  • 而主题模式(Exchange类型为topic)，消费者在将Exchange与Queue绑定时设置的路由键可以有通配。
  • 生产者
package com.futao.springmvcdemo.mq.rabbit.topic;

import com.futao.springmvcdemo.mq.rabbit.ExchangeTypeEnum;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import lombok.Cleanup;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * Topic模式-生产者
 *
 * @author futao
 * Created on 2019-04-22.
 */
@Slf4j
public class Producer {
    @SneakyThrows
    public static void main(String[] args) {
        @Cleanup
        Connection connection = RabbitMqConnectionTools.getConnection();
        @Cleanup
        Channel channel = connection.createChannel();
        //定义交换器类型
        channel.exchangeDeclare(ExchangeTypeEnum.TOPIC.getExchangeName(), BuiltinExchangeType.TOPIC);
        String msg = "Hello RabbitMq!";
        for (int i = 0; i < 10; i++) {
            channel.basicPublish(ExchangeTypeEnum.TOPIC.getExchangeName(), "log.info", null, (msg + i + "log.info").getBytes());
            log.info("Send msg:[{}] ,routingKey:[{}] success", (msg + i + "log.info"), "log.info");
        }
        for (int i = 0; i < 10; i++) {
            channel.basicPublish(ExchangeTypeEnum.TOPIC.getExchangeName(), "log.error", null, (msg + i + "log.error").getBytes());
            log.info("Send msg:[{}] ,routingKey:[{}] success", (msg + i + "log.error"), "log.error");
        }
        for (int i = 0; i < 10; i++) {
            channel.basicPublish(ExchangeTypeEnum.TOPIC.getExchangeName(), "log.error.rabbitmq", null, (msg + i + "log.error.rabbitmq").getBytes());
            log.info("Send msg:[{}] ,routingKey:[{}] success", (msg + i + "log.error.rabbitmq"), "log.error.rabbitmq");
        }
    }
}
  • 消费者1 - 订阅log.error and log.info
package com.futao.springmvcdemo.mq.rabbit.topic;

import com.futao.springmvcdemo.mq.rabbit.ExchangeTypeEnum;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * Topic模式-消费者1
 *
 * @author futao
 * Created on 2019-04-22.
 */
@Slf4j
public class ConsumerOne {
    @SneakyThrows
    public static void main(String[] args) {
        Channel channel = RabbitMqConnectionTools.getChannel();
        //开启持久化队列
        boolean durable = true;
        channel.queueDeclare(RabbitMqQueueEnum.EXCHANGE_QUEUE_TOPIC_ONE.getQueueName(), durable, false, false, null);
        //定义交换器类型
        channel.exchangeDeclare(ExchangeTypeEnum.DIRECT.getExchangeName(), BuiltinExchangeType.TOPIC);
        //将消息队列与Exchange交换器与路由键绑定（同时订阅路由key为"log.info"和"log.error"的消息）
        channel.queueBind(RabbitMqQueueEnum.EXCHANGE_QUEUE_TOPIC_ONE.getQueueName(), ExchangeTypeEnum.TOPIC.getExchangeName(), "log.error");
        channel.queueBind(RabbitMqQueueEnum.EXCHANGE_QUEUE_TOPIC_ONE.getQueueName(), ExchangeTypeEnum.TOPIC.getExchangeName(), "log.info");
        //告诉rabbitmq一次只发送一条消息，并且在前一个消息未被处理或者消费之前，不继续发送下一个消息
        channel.basicQos(1);
        log.info("Waiting for message...");
        DeliverCallback deliverCallback = ((consumerTag, message) -> {
            log.info("收到消息:[{}],routingKey：[{}],tag:[{}]", new String(message.getBody()), message.getEnvelope().getRoutingKey(), consumerTag);
            //acknowledgment应答
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            try {
                Thread.sleep(1000);
            } catch (Exception e) {

            }
        });
        //关闭自动应答
        boolean autoAck = false;
        channel.basicConsume(RabbitMqQueueEnum.EXCHANGE_QUEUE_TOPIC_ONE.getQueueName(), autoAck, deliverCallback, consumerTag -> {
        });
    }
}
  • 消费者2 - 订阅log.error
package com.futao.springmvcdemo.mq.rabbit.topic;

import com.futao.springmvcdemo.mq.rabbit.ExchangeTypeEnum;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * Topic模式-消费者2
 *
 * @author futao
 * Created on 2019-04-22.
 */
@Slf4j
public class ConsumerTwo {
    @SneakyThrows
    public static void main(String[] args) {
        Channel channel = RabbitMqConnectionTools.getChannel();
        //开启持久化队列
        boolean durable = true;
        channel.queueDeclare(RabbitMqQueueEnum.EXCHANGE_QUEUE_TOPIC_TWO.getQueueName(), durable, false, false, null);
        //定义交换器类型
        channel.exchangeDeclare(ExchangeTypeEnum.TOPIC.getExchangeName(), BuiltinExchangeType.TOPIC);
        //将消息队列与Exchange交换器进行绑定(只订阅路由key为"log.error"的消息)
        channel.queueBind(RabbitMqQueueEnum.EXCHANGE_QUEUE_TOPIC_TWO.getQueueName(), ExchangeTypeEnum.TOPIC.getExchangeName(), "log.error");
        //告诉rabbitmq一次只发送一条消息，并且在前一个消息未被处理或者消费之前，不继续发送下一个消息
        channel.basicQos(1);
        log.info("Waiting for message...");
        DeliverCallback deliverCallback = ((consumerTag, message) -> {
            log.info("收到消息:[{}],routingKey：[{}],tag:[{}]", new String(message.getBody()), message.getEnvelope().getRoutingKey(), consumerTag);
            //acknowledgment应答
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            try {
                Thread.sleep(2000);
            } catch (Exception e) {

            }
        });
        //关闭自动应答
        boolean autoAck = false;
        channel.basicConsume(RabbitMqQueueEnum.EXCHANGE_QUEUE_TOPIC_TWO.getQueueName(), autoAck, deliverCallback, consumerTag -> {
        });
    }
}
  • 消费者3 - 订阅log.*
package com.futao.springmvcdemo.mq.rabbit.topic;

import com.futao.springmvcdemo.mq.rabbit.ExchangeTypeEnum;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * Topic模式-消费者3
 *
 * @author futao
 * Created on 2019-04-22.
 */
@Slf4j
public class ConsumerThree {
    @SneakyThrows
    public static void main(String[] args) {
        Channel channel = RabbitMqConnectionTools.getChannel();
        //开启持久化队列
        boolean durable = true;
        channel.queueDeclare(RabbitMqQueueEnum.EXCHANGE_QUEUE_TOPIC_THREE.getQueueName(), durable, false, false, null);
        //定义交换器类型
        channel.exchangeDeclare(ExchangeTypeEnum.DIRECT.getExchangeName(), BuiltinExchangeType.TOPIC);
        //将消息队列与Exchange交换器与路由键绑定（同时订阅路由key为"log.info"和"log.error"的消息）
        channel.queueBind(RabbitMqQueueEnum.EXCHANGE_QUEUE_TOPIC_THREE.getQueueName(), ExchangeTypeEnum.TOPIC.getExchangeName(), "log.*");
        //告诉rabbitmq一次只发送一条消息，并且在前一个消息未被处理或者消费之前，不继续发送下一个消息
        channel.basicQos(1);
        log.info("Waiting for message...");
        DeliverCallback deliverCallback = ((consumerTag, message) -> {
            log.info("收到消息:[{}],routingKey：[{}],tag:[{}]", new String(message.getBody()), message.getEnvelope().getRoutingKey(), consumerTag);
            //acknowledgment应答
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            try {
                Thread.sleep(1000);
            } catch (Exception e) {

            }
        });
        //关闭自动应答
        boolean autoAck = false;
        channel.basicConsume(RabbitMqQueueEnum.EXCHANGE_QUEUE_TOPIC_THREE.getQueueName(), autoAck, deliverCallback, consumerTag -> {
        });
    }
}
  • 消费者4 - 订阅log.#
package com.futao.springmvcdemo.mq.rabbit.topic;

import com.futao.springmvcdemo.mq.rabbit.ExchangeTypeEnum;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * Topic模式-消费者4
 *
 * @author futao
 * Created on 2019-04-22.
 */
@Slf4j
public class ConsumerFour {
    @SneakyThrows
    public static void main(String[] args) {
        Channel channel = RabbitMqConnectionTools.getChannel();
        //开启持久化队列
        boolean durable = true;
        channel.queueDeclare(RabbitMqQueueEnum.EXCHANGE_QUEUE_TOPIC_FOUR.getQueueName(), durable, false, false, null);
        //定义交换器类型
        channel.exchangeDeclare(ExchangeTypeEnum.DIRECT.getExchangeName(), BuiltinExchangeType.TOPIC);
        //将消息队列与Exchange交换器与路由键绑定（同时订阅路由key为"log.info"和"log.error"的消息）
        channel.queueBind(RabbitMqQueueEnum.EXCHANGE_QUEUE_TOPIC_FOUR.getQueueName(), ExchangeTypeEnum.TOPIC.getExchangeName(), "log.#");
        //告诉rabbitmq一次只发送一条消息，并且在前一个消息未被处理或者消费之前，不继续发送下一个消息
        channel.basicQos(1);
        log.info("Waiting for message...");
        DeliverCallback deliverCallback = ((consumerTag, message) -> {
            log.info("收到消息:[{}],routingKey：[{}],tag:[{}]", new String(message.getBody()), message.getEnvelope().getRoutingKey(), consumerTag);
            //acknowledgment应答
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            try {
                Thread.sleep(1000);
            } catch (Exception e) {

            }
        });
        //关闭自动应答
        boolean autoAck = false;
        channel.basicConsume(RabbitMqQueueEnum.EXCHANGE_QUEUE_TOPIC_FOUR.getQueueName(), autoAck, deliverCallback, consumerTag -> {
        });
    }
}
  • 测试
>>> 生产者-生产主题分别为：`log.info` and `log.error` and `log.error.rabbitmq`的消息
略

>>> 消费者1 - 订阅`log.error` and `log.info`
19:42:31.130 [main] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - Waiting for message...
19:42:48.506 [pool-1-thread-4] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!0log.info],routingKey：[log.info],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:42:49.509 [pool-1-thread-5] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!1log.info],routingKey：[log.info],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:42:50.513 [pool-1-thread-6] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!2log.info],routingKey：[log.info],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:42:51.517 [pool-1-thread-7] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!3log.info],routingKey：[log.info],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:42:52.520 [pool-1-thread-8] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!4log.info],routingKey：[log.info],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:42:53.524 [pool-1-thread-9] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!5log.info],routingKey：[log.info],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:42:54.529 [pool-1-thread-10] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!6log.info],routingKey：[log.info],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:42:55.535 [pool-1-thread-11] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!7log.info],routingKey：[log.info],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:42:56.539 [pool-1-thread-12] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!8log.info],routingKey：[log.info],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:42:57.540 [pool-1-thread-13] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!9log.info],routingKey：[log.info],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:42:58.545 [pool-1-thread-14] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!0log.error],routingKey：[log.error],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:42:59.551 [pool-1-thread-15] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!1log.error],routingKey：[log.error],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:43:00.556 [pool-1-thread-16] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!2log.error],routingKey：[log.error],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:43:01.560 [pool-1-thread-17] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!3log.error],routingKey：[log.error],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:43:02.561 [pool-1-thread-18] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!4log.error],routingKey：[log.error],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:43:03.566 [pool-1-thread-19] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!5log.error],routingKey：[log.error],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:43:04.569 [pool-1-thread-20] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!6log.error],routingKey：[log.error],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:43:05.572 [pool-1-thread-21] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!7log.error],routingKey：[log.error],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:43:06.574 [pool-1-thread-22] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!8log.error],routingKey：[log.error],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]
19:43:07.580 [pool-1-thread-23] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerOne - 收到消息:[Hello RabbitMq!9log.error],routingKey：[log.error],tag:[amq.ctag-YQRXOwb2xsoODmB2fpJ1Qg]

>>> 消费者2 - 订阅`log.error` 
19:42:36.272 [main] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerTwo - Waiting for message...
19:42:48.516 [pool-1-thread-4] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerTwo - 收到消息:[Hello RabbitMq!0log.error],routingKey：[log.error],tag:[amq.ctag-Q3m55Q8-XWC-SlF1qX80cA]
19:42:50.522 [pool-1-thread-5] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerTwo - 收到消息:[Hello RabbitMq!1log.error],routingKey：[log.error],tag:[amq.ctag-Q3m55Q8-XWC-SlF1qX80cA]
19:42:52.523 [pool-1-thread-6] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerTwo - 收到消息:[Hello RabbitMq!2log.error],routingKey：[log.error],tag:[amq.ctag-Q3m55Q8-XWC-SlF1qX80cA]
19:42:54.527 [pool-1-thread-7] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerTwo - 收到消息:[Hello RabbitMq!3log.error],routingKey：[log.error],tag:[amq.ctag-Q3m55Q8-XWC-SlF1qX80cA]
19:42:56.529 [pool-1-thread-8] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerTwo - 收到消息:[Hello RabbitMq!4log.error],routingKey：[log.error],tag:[amq.ctag-Q3m55Q8-XWC-SlF1qX80cA]
19:42:58.533 [pool-1-thread-9] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerTwo - 收到消息:[Hello RabbitMq!5log.error],routingKey：[log.error],tag:[amq.ctag-Q3m55Q8-XWC-SlF1qX80cA]
19:43:00.534 [pool-1-thread-10] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerTwo - 收到消息:[Hello RabbitMq!6log.error],routingKey：[log.error],tag:[amq.ctag-Q3m55Q8-XWC-SlF1qX80cA]
19:43:02.539 [pool-1-thread-11] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerTwo - 收到消息:[Hello RabbitMq!7log.error],routingKey：[log.error],tag:[amq.ctag-Q3m55Q8-XWC-SlF1qX80cA]
19:43:04.541 [pool-1-thread-12] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerTwo - 收到消息:[Hello RabbitMq!8log.error],routingKey：[log.error],tag:[amq.ctag-Q3m55Q8-XWC-SlF1qX80cA]
19:43:06.543 [pool-1-thread-13] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerTwo - 收到消息:[Hello RabbitMq!9log.error],routingKey：[log.error],tag:[amq.ctag-Q3m55Q8-XWC-SlF1qX80cA]

>>> 消费者3 - 订阅`log.*`
19:42:39.746 [main] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - Waiting for message...
19:42:48.507 [pool-1-thread-4] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!0log.info],routingKey：[log.info],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:42:49.509 [pool-1-thread-5] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!1log.info],routingKey：[log.info],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:42:50.513 [pool-1-thread-6] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!2log.info],routingKey：[log.info],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:42:51.517 [pool-1-thread-7] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!3log.info],routingKey：[log.info],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:42:52.520 [pool-1-thread-8] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!4log.info],routingKey：[log.info],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:42:53.524 [pool-1-thread-9] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!5log.info],routingKey：[log.info],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:42:54.529 [pool-1-thread-10] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!6log.info],routingKey：[log.info],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:42:55.535 [pool-1-thread-11] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!7log.info],routingKey：[log.info],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:42:56.539 [pool-1-thread-12] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!8log.info],routingKey：[log.info],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:42:57.540 [pool-1-thread-13] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!9log.info],routingKey：[log.info],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:42:58.545 [pool-1-thread-14] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!0log.error],routingKey：[log.error],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:42:59.551 [pool-1-thread-15] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!1log.error],routingKey：[log.error],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:43:00.556 [pool-1-thread-16] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!2log.error],routingKey：[log.error],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:43:01.560 [pool-1-thread-17] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!3log.error],routingKey：[log.error],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:43:02.561 [pool-1-thread-18] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!4log.error],routingKey：[log.error],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:43:03.566 [pool-1-thread-19] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!5log.error],routingKey：[log.error],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:43:04.569 [pool-1-thread-20] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!6log.error],routingKey：[log.error],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:43:05.572 [pool-1-thread-21] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!7log.error],routingKey：[log.error],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:43:06.574 [pool-1-thread-22] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!8log.error],routingKey：[log.error],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]
19:43:07.580 [pool-1-thread-23] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerThree - 收到消息:[Hello RabbitMq!9log.error],routingKey：[log.error],tag:[amq.ctag-0-_1m0BhTsYpjikUjsWEQA]


>>> 消费者4 - 订阅`log.#`
19:42:42.624 [main] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - Waiting for message...
19:42:48.506 [pool-1-thread-4] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!0log.info],routingKey：[log.info],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:42:49.509 [pool-1-thread-5] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!1log.info],routingKey：[log.info],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:42:50.513 [pool-1-thread-6] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!2log.info],routingKey：[log.info],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:42:51.517 [pool-1-thread-7] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!3log.info],routingKey：[log.info],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:42:52.520 [pool-1-thread-8] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!4log.info],routingKey：[log.info],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:42:53.524 [pool-1-thread-9] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!5log.info],routingKey：[log.info],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:42:54.529 [pool-1-thread-10] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!6log.info],routingKey：[log.info],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:42:55.535 [pool-1-thread-11] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!7log.info],routingKey：[log.info],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:42:56.539 [pool-1-thread-12] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!8log.info],routingKey：[log.info],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:42:57.540 [pool-1-thread-13] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!9log.info],routingKey：[log.info],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:42:58.545 [pool-1-thread-14] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!0log.error],routingKey：[log.error],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:42:59.551 [pool-1-thread-15] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!1log.error],routingKey：[log.error],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:00.556 [pool-1-thread-16] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!2log.error],routingKey：[log.error],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:01.560 [pool-1-thread-17] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!3log.error],routingKey：[log.error],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:02.561 [pool-1-thread-18] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!4log.error],routingKey：[log.error],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:03.566 [pool-1-thread-19] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!5log.error],routingKey：[log.error],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:04.569 [pool-1-thread-20] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!6log.error],routingKey：[log.error],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:05.572 [pool-1-thread-21] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!7log.error],routingKey：[log.error],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:06.574 [pool-1-thread-22] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!8log.error],routingKey：[log.error],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:07.580 [pool-1-thread-23] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!9log.error],routingKey：[log.error],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:08.586 [pool-1-thread-24] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!0log.error.rabbitmq],routingKey：[log.error.rabbitmq],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:09.587 [pool-1-thread-25] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!1log.error.rabbitmq],routingKey：[log.error.rabbitmq],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:10.589 [pool-1-thread-26] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!2log.error.rabbitmq],routingKey：[log.error.rabbitmq],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:11.593 [pool-1-thread-26] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!3log.error.rabbitmq],routingKey：[log.error.rabbitmq],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:12.599 [pool-1-thread-26] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!4log.error.rabbitmq],routingKey：[log.error.rabbitmq],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:13.602 [pool-1-thread-26] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!5log.error.rabbitmq],routingKey：[log.error.rabbitmq],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:14.605 [pool-1-thread-26] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!6log.error.rabbitmq],routingKey：[log.error.rabbitmq],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:15.607 [pool-1-thread-26] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!7log.error.rabbitmq],routingKey：[log.error.rabbitmq],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:16.608 [pool-1-thread-26] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!8log.error.rabbitmq],routingKey：[log.error.rabbitmq],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
19:43:17.609 [pool-1-thread-26] INFO com.futao.springmvcdemo.mq.rabbit.topic.ConsumerFour - 收到消息:[Hello RabbitMq!9log.error.rabbitmq],routingKey：[log.error.rabbitmq],tag:[amq.ctag-Ml8Iiz8KSEh4RBn4yfA0xA]
文章标签：
云消息队列 MQ
关键词：
云消息队列 MQ模式
相关实践学习
消息队列RocketMQ版：基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能，涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
futaosmile
目录
相关文章
小王老师呀
|
2月前
|
消息中间件
RabbitMQ的 RPC 消息模式你会了吗？
【9月更文挑战第11天】RabbitMQ 的 RPC（远程过程调用）消息模式允许客户端向服务器发送请求并接收响应。其基本原理包括：1) 客户端发送请求，创建回调队列并设置关联标识符；2) 服务器接收请求并发送响应至回调队列；3) 客户端根据关联标识符接收并匹配响应。实现步骤涵盖客户端和服务器的连接、信道创建及请求处理。注意事项包括关联标识符唯一性、回调队列管理、错误处理及性能考虑。RPC 模式适用于构建可靠的分布式应用程序，但需根据需求调整优化。
小王老师呀
55 2
土木林森
|
3月前
|
消息中间件 开发者
【RabbitMQ深度解析】Topic交换器与模式匹配：掌握消息路由的艺术！
【8月更文挑战第24天】在消息队列（MQ）体系中，交换器作为核心组件之一负责消息路由。特别是`topic`类型的交换器，它通过模式匹配实现消息的精准分发，适用于发布-订阅模式。不同于直接交换器和扇形交换器，`topic`交换器支持更复杂的路由策略，通过带有通配符（如 * 和 #）的模式字符串来定义队列与交换器间的绑定关系。
土木林森
70 2
秋夜雨巷
|
3月前
|
消息中间件
RabbitMQ广播模式
RabbitMQ广播模式
秋夜雨巷
60 1
1288912195458132
|
3月前
|
消息中间件 应用服务中间件 网络安全
rabbitMQ镜像模式搭建
rabbitMQ镜像模式搭建
1288912195458132
119 0
三分钟热度的鱼
|
4月前
|
消息中间件 传感器 负载均衡
消息队列 MQ使用问题之如何配置一主一从的同步复制模式
消息队列（MQ）是一种用于异步通信和解耦的应用程序间消息传递的服务，广泛应用于分布式系统中。针对不同的MQ产品，如阿里云的RocketMQ、RabbitMQ等，它们在实现上述场景时可能会有不同的特性和优势，比如RocketMQ强调高吞吐量、低延迟和高可用性，适合大规模分布式系统；而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集，这些场景涵盖了多种行业和业务需求。
三分钟热度的鱼
43 0
消息队列 MQ使用问题之如何配置一主一从的同步复制模式
萝卜丝丸子
|
4月前
|
消息中间件 存储 Kafka
MetaQ/RocketMQ 原理问题之RocketMQ DLedger融合模式的问题如何解决
MetaQ/RocketMQ 原理问题之RocketMQ DLedger融合模式的问题如何解决
萝卜丝丸子
50 2
monGyrate
|
3月前
|
消息中间件 Java Maven
RabbitMQ通配符模式
RabbitMQ通配符模式
monGyrate
60 0
三分钟热度的鱼
|
4月前
|
消息中间件 Java Apache
消息队列 MQ使用问题之如何在内外网环境下使用单组节点单副本模式
消息队列（MQ）是一种用于异步通信和解耦的应用程序间消息传递的服务，广泛应用于分布式系统中。针对不同的MQ产品，如阿里云的RocketMQ、RabbitMQ等，它们在实现上述场景时可能会有不同的特性和优势，比如RocketMQ强调高吞吐量、低延迟和高可用性，适合大规模分布式系统；而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集，这些场景涵盖了多种行业和业务需求。
三分钟热度的鱼
89 0
萝卜丝丸子
|
4月前
|
消息中间件 负载均衡 RocketMQ
MetaQ/RocketMQ 原理问题之在广播模式下，RebalanceService工作的问题如何解决
MetaQ/RocketMQ 原理问题之在广播模式下，RebalanceService工作的问题如何解决
萝卜丝丸子
40 0
好奇的菜鸟
|
5月前
|
消息中间件
RabbitMQ配置单活模式队列
RabbitMQ配置单活模式队列
好奇的菜鸟
137 0

热门文章

最新文章

  • 1
    信用算力基于 RocketMQ 实现金融级数据服务的实践
  • 2
    RabbitMQ(从安装到使用)
  • 3
    MQTT协议
  • 4
    消息中间件的技术选型心得－RabbitMQ、ActiveMQ和ZeroMQ
  • 5
    【解决方案】MQTT 强劲代理服务 EMQX 安装报Unable to load emulator DLL错误
  • 6
    【Spring常见错误】Rocketmq 闪退
  • 7
    解决微信小程序MQTT真机连接问题与合法域名配置SSL问题
  • 8
    rabbitmq消息件在idc机房监控中的应用
  • 9
    RocketMQ 消息发送system busy、broker busy原因分析与解决方案
  • 10
    MQTT的心跳保活机制是通过`setKeepAliveInterval()`方法设置的
  • 1
    Spring Cloud 项目中实现推送消息到 RabbitMQ 消息中间件
    84
  • 2
    RocketMQ源码分析之事务消息实现原理下篇-消息服务器Broker提交回滚事务实现原理
    65
  • 3
    java面试刷题软件kafka和mq的区别面试
    68
  • 4
    MQTT协议与EMQ
    173
  • 5
    MQ产品使用合集之RocketMQ发消息失败了，proxy报connect to null failed如何解决
    788
  • 6
    MQ产品使用合集之设置rocketmq的timerMaxDelaySec时间出现报错如何解决
    326
  • 7
    MQ产品使用合集之在MQTT中，出现了 NO HEART 错误如何解决
    120
  • 8
    MQ产品使用合集之RocketMQ如果配置所有的ip，有些namesrv挂了的话，消息就发送失败了，消费也是失败的如何解决
    76
  • 9
    MQ产品使用合集之使用的RocketMQ5.1.3时，grpc客户端没有产生消息轨迹如何解决
    163
  • 10
    MQ产品使用合集之有RocketMQ arm架构的镜像吗
    322

    • 相关课程

    更多
  • 消息队列 RocketMQ 消息集成
  • 消息队列 MNS 入门课程
  • RocketMQ知识精讲与项目实战（第一阶段）
  • RocketMQ知识精讲与项目实战（第二阶段）
  • RocketMQ知识精讲与项目实战（第三阶段）
  • 微服务实战-RocketMQ Binder

    • 相关电子书

    更多
  • RocketMQ Client-GO 介绍
  • RocketMQ Prometheus Exporter 打造定制化 DevOps 平台
  • 基于 RocketMQ Prometheus Exporter 打造定制化 DevOps 平台

    • 相关实验场景

    更多
  • 通过轻量消息队列（原MNS）主题HTTP订阅+ARMS实现自定义数据多渠道告警
  • 快速体验云消息队列RocketMQ版
  • 消息队列RocketMQ版：基础消息收发功能体验
  • 快速学会使用云消息队列RabbitMQ版
  • 消息队列RocketMQ版：定时消息通知功能体验
  • 基于ROS的消息发布与订阅
    • 下一篇
    阿里云OSS设置跨域访问