【SpringBoot整合系列】SpringBoot整合RabbitMQ-基本使用

目录

  • SpringtBoot整合RabbitMQ
    • 1.依赖
    • 2.配置
    • RabbitMQ的7种模式
      • 1.简单模式（Hello World）
        • 应用场景
        • 代码示例
      • 2.工作队列模式（Work queues）
        • 应用场景
        • 代码示例
        • 手动 ack
          • 代码示例
      • 3.订阅模式（Publish/Subscribe）
        • 应用场景
        • 代码示例
      • 4.路由模式（Routing）
        • 应用场景
        • 代码示例
      • 5.主题模式（Topics）
        • 应用场景
        • 代码示例
      • 6.远程过程调用（RPC）
        • 应用场景
        • 代码示例
          • 消息生产者开发
          • 消息发送者开发
      • 7.发布者确认（Publisher Confirms）
        • 应用场景
    • RabbitMQ的四种交换机
      • 1.直连交换机（Direct exchange）
        • 代码示例
      • 2.扇形交换机（Fanout exchange）
        • 代码示例
      • 3.主题交换机（Topic exchange）
        • 代码示例
      • 4.首部交换机（Headers exchange）
        • 代码示例

SpringtBoot整合RabbitMQ

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><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency>

2.配置

server:port: 9090
spring:rabbitmq:host: 192.168.29.200port: 5672username: adminpassword: adminvirtual-host: /

RabbitMQ的7种模式

1.简单模式（Hello World）

在这里插入图片描述

  • 做最简单的事情，一个生产者对应一个消费者，RabbitMQ相当于一个消息代理，负责将A的消息转发给B
应用场景
  • 将发送的电子邮件放到消息队列，然后邮件服务在队列中获取邮件并发送给收件人
代码示例
  1. 配置类
    @Configuration
public class HelloWorldConfig {public static final String HELLO_WORLD_QUEUE_NAME = "hello_world_queue";@Beanpublic Queue queue1() {return new Queue(HELLO_WORLD_QUEUE_NAME);}
}
  2. 监听并消费消息
    @Component
public class HelloWorldConsumer {@RabbitListener(queues = HelloWorldConfig.HELLO_WORLD_QUEUE_NAME)public void receive(String msg) {System.out.println("msg = " + msg);}
}
  3. 生产消息并发送
    @SpringBootTest
class MyMqBootApplicationTests {@ResourceRabbitTemplate rabbitTemplate;@Testvoid helloworld() {rabbitTemplate.convertAndSend(HelloWorldConfig.HELLO_WORLD_QUEUE_NAME, "hello world!!!");}
}

2.工作队列模式（Work queues）

在这里插入图片描述

  • 在多个消费者之间分配任务（竞争的消费者模式），一个生产者对应多个消费者，一般适用于执行资源密集型任务，单个消费者处理不过来，需要多个消费者进行处理
  • 一个队列对应了多个消费者，默认情况下，由队列对消息进行平均分配，消息会被分到不同的消费者手中。消费者可以配置各自的并发能力，进而提高消息的消费能力，也可以配置手动 ack，来决定是否要消费某一条消息。
应用场景
  • 一个订单的处理需要10s，有多个订单可以同时放到消息队列，然后让多个消费者同时处理，这样就是并行了，而不是单个消费者的串行情况
代码示例

  1. 监听并消费消息

    @Component
public class HelloWorldConsumer {@RabbitListener(queues = HelloWorldConfig.HELLO_WORLD_QUEUE_NAME)public void receive(String msg) {System.out.println("receive = " + msg);}// concurrency 为 10，此时，receive2将会同时存在 10 个子线程去消费消息@RabbitListener(queues = HelloWorldConfig.HELLO_WORLD_QUEUE_NAME,concurrency = "10")public void receive2(String msg) {System.out.println("receive2 = " + msg+"------->"+Thread.currentThread().getName());}
}

  2. 生产消息并发送

        @Testvoid work() {for (int i = 0; i < 10; i++) {rabbitTemplate.convertAndSend(HelloWorldConfig.HELLO_WORLD_QUEUE_NAME, "hello");}}

  3. 运行结果：运行结果每次不一定一样

    receive2 = hello------->org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-9
receive2 = hello------->org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-10
receive2 = hello------->org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-1
receive2 = hello------->org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-3
receive2 = hello------->org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-7
receive2 = hello------->org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-5
receive = hello
receive2 = hello------->org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-4
receive2 = hello------->org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-6
receive2 = hello------->org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-8
手动 ack

手动 ack可以自行决定是否消费 RabbitMQ 发来的消息

代码示例
  1. 配置文件：配置手动ack
server:port: 9090
spring:rabbitmq:host: 192.168.29.200port: 5672username: adminpassword: adminvirtual-host: /listener:simple:acknowledge-mode: manual # 配置手动ack
  1. 消费代码：receive2 拒绝了所有消息，第一个消费者消费了所有消息
    @Component
public class HelloWorldConsumer {@RabbitListener(queues = HelloWorldConfig.HELLO_WORLD_QUEUE_NAME)public void receive(Message message, Channel channel) throws IOException {System.out.println("receive="+message.getPayload());//手动确认channel.basicAck(((Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG)),true);}@RabbitListener(queues = HelloWorldConfig.HELLO_WORLD_QUEUE_NAME, concurrency = "10")public void receive2(Message message, Channel channel) throws IOException {System.out.println("receive2 = " + message.getPayload() + "------->" + Thread.currentThread().getName());//手动拒绝channel.basicReject(((Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG)), true);}
}
  2. 测试结果(生产代码不变)
  • 此时receive2 拒绝了所有消息，receive消费了所有消息（如果receive2没有拒绝，receive断然不会消费10次）

3.订阅模式（Publish/Subscribe）

在这里插入图片描述

  • 一次向许多消费者发送消息，一个生产者发送的消息会被多个消费者获取，也就是将消息将广播到所有的消费者中。
  • 一个生产者，多个消费者，每一个消费者都有自己的一个队列，生产者没有将消息直接发送到队列，而是发送到了交换机，每个队列绑定交换机，生产者发送的消息经过交换机，到达队列，实现一个消息被多个消费者获取的目的。
  • 需要注意的是，如果将消息发送到一个没有队列绑定的 Exchange上面，那么该消息将会丢失，这是因为在 RabbitMQ 中 Exchange 不具备存储消息的能力，只有队列具备存储消息的能力
应用场景
  • 更新商品库存后需要通知多个缓存和多个数据库，这里的结构应该是：
    • 一个fanout类型交换机扇出两个个消息队列，分别为缓存消息队列、数据库消息队列
    • 一个缓存消息队列对应着多个缓存消费者
    • 一个数据库消息队列对应着多个数据库消费者
代码示例

具体内容看本文单独的目录 RabbitMQ的四种交换机 ， 我这里单独拿出来解释了

4.路由模式（Routing）

在这里插入图片描述

  • 有选择地（Routing key）接收消息，发送消息到交换机并且要指定路由key ，消费者将队列绑定到交换机时需要指定路由key，仅消费指定路由key的消息
  • 一个生产者，一个交换机，两个队列，两个消费者，生产者在创建 Exchange 后，根据 RoutingKey 去绑定相应的队列，并且在发送消息时，指定消息的具体 RoutingKey 即可
应用场景
  • 如在商品库存中增加了1台iphone12，iphone12促销活动消费者指定routing key为iphone12，只有此促销活动会接收到消息，其它促销活动不关心也不会消费此routing key的消息
代码示例

参考本文单独的目录 RabbitMQ的四种交换机-1

5.主题模式（Topics）

在这里插入图片描述

  • 根据主题（Topics）来接收消息，将路由key和某模式进行匹配，此时队列需要绑定在一个模式上，#匹配一个词或多个词，*只匹配一个词。
  • 一个生产者，一个交换机，两个队列，两个消费者，生产者创建 Topic 的 Exchange 并且绑定到队列中，这次绑定可以通过 * 和 # 关键字，对指定 RoutingKey 内容，编写时注意格式 xxx.xxx.xxx 去编写
应用场景

同上，iphone促销活动可以接收主题为iphone的消息，如iphone12、iphone13，iphone…等

代码示例

参考本文单独的目录 RabbitMQ的四种交换机-3

6.远程过程调用（RPC）

在这里插入图片描述
如果我们需要在远程计算机上运行功能并等待结果就可以使用RPC，具体流程可以看图。

  • 首先 Client 发送一条消息，和普通的消息相比，这条消息多了两个关键内容：一个是 correlation_id，这个表示这条消息的唯一 id，还有一个内容是 reply_to，这个表示消息回复队列的名字。
  • Server 从消息发送队列获取消息并处理相应的业务逻辑，处理完成后，将处理结果发送到 reply_to 指定的回调队列中。
  • Client 从回调队列中读取消息，就可以知道消息的执行情况是什么样子了。

这种情况其实非常适合处理异步调用。

应用场景
  • 需要等待接口返回数据，如订单支付
代码示例
消息生产者开发

  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><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency>

  2. 配置

    server:port: 7777
spring:rabbitmq:host: 192.168.29.200port: 5672username: adminpassword: adminvirtual-host: /publisher-confirm-type: correlated # 配置消息确认方式，我们通过 correlated 来确认，只有开启了这个配置，将来的消息中才会带 correlation_id，只有通过 correlation_id 我们才能将发送的消息和返回值之间关联起来。publisher-returns: true #开启发送失败退回。

  3. 配置类

  4. /*** @author: zjl* @datetime: 2024/5/9* @desc: * 这个配置类中我们分别配置了消息发送队列 msgQueue 和消息返回队列 replyQueue，* 然后将这两个队列和消息交换机进行绑定。常规操作。* 在 Spring Boot 中我们负责消息发送的工具是 RabbitTemplate，* 默认情况下，系统自动提供了该工具，但是这里我们需要对该工具重新进行定制，* 主要是添加消息发送的返回队列，最后我们还需要给返回队列设置一个监听器*/
@Configuration
public class RabbitConfig {public static final String RPC_QUEUE1 = "queue_1";public static final String RPC_QUEUE2 = "queue_2";public static final String RPC_EXCHANGE = "rpc_exchange";/*** 设置消息发送RPC队列*/@Beanpublic Queue msgQueue() {return new Queue(RPC_QUEUE1);}/*** 设置返回队列*/@Beanpublic Queue replyQueue() {return new Queue(RPC_QUEUE2);}/*** 设置交换机*/@Beanpublic TopicExchange exchange() {return new TopicExchange(RPC_EXCHANGE);}/*** 请求队列和交换器绑定*/@Beanpublic Binding msgBinding() {return BindingBuilder.bind(msgQueue()).to(exchange()).with(RPC_QUEUE1);}/*** 返回队列和交换器绑定*/@Beanpublic Binding replyBinding() {return BindingBuilder.bind(replyQueue()).to(exchange()).with(RPC_QUEUE2);}/*** 使用 RabbitTemplate发送和接收消息* 并设置回调队列地址*/@Beanpublic RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {RabbitTemplate template = new RabbitTemplate(connectionFactory);template.setReplyAddress(RPC_QUEUE2);template.setReplyTimeout(6000);return template;}/*** 给返回队列设置监听器*/@Beanpublic SimpleMessageListenerContainer replyContainer(ConnectionFactory connectionFactory) {SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();container.setConnectionFactory(connectionFactory);container.setQueueNames(RPC_QUEUE2);container.setMessageListener(rabbitTemplate(connectionFactory));return container;}
}

  5. 消息发送

    /*** @author: zjl* @datetime: 2024/5/9* @desc:*     消息发送调用 sendAndReceive 方法，该方法自带返回值，返回值就是服务端返回的消息。*     服务端返回的消息中，头信息中包含了 spring_returned_message_correlation 字段，*     这个就是消息发送时候的 correlation_id，通过消息发送时候的 correlation_id*     以及返回消息头中的 spring_returned_message_correlation 字段值，*     我们就可以将返回的消息内容和发送的消息绑定到一起，*     确认出这个返回的内容就是针对这个发送的消息的。*/
@RestController
@Slf4j
public class RpcClientController {@Resourceprivate RabbitTemplate rabbitTemplate;@GetMapping("/send")public String send(String message) {// 创建消息对象Message newMessage = MessageBuilder.withBody(message.getBytes()).build();log.info("client send：{}", newMessage);//客户端发送消息Message result = rabbitTemplate.sendAndReceive(RabbitConfig.RPC_EXCHANGE, RabbitConfig.RPC_QUEUE1, newMessage);String response = "";if (result != null) {// 获取已发送的消息的 correlationIdString correlationId = newMessage.getMessageProperties().getCorrelationId();log.info("correlationId:{}", correlationId);// 获取响应头信息HashMap<String, Object> headers = (HashMap<String, Object>) result.getMessageProperties().getHeaders();// 获取 server 返回的消息 idString msgId = (String) headers.get("spring_returned_message_correlation");if (msgId.equals(correlationId)) {response = new String(result.getBody());log.info("client receive：{}", response);}}return response;}
}

这就是整个消息生产者的开发，其实最最核心的就是 sendAndReceive 方法的调用。调用虽然简单，但是准备工作还是要做足够。例如如果我们没有在 application.properties 中配置 correlated，发送的消息中就没有 correlation_id，这样就无法将返回的消息内容和发送的消息内容关联起来

消息发送者开发

  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><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency>

  2. 配置

    server:port: 8888
spring:rabbitmq:host: 192.168.29.200port: 5672username: adminpassword: adminvirtual-host: /publisher-confirm-type: correlated # 配置消息确认方式，我们通过 correlated 来确认，只有开启了这个配置，将来的消息中才会带 correlation_id，只有通过 correlation_id 我们才能将发送的消息和返回值之间关联起来。publisher-returns: true #开启发送失败退回。

  3. 配置类

    @Configuration
public class RabbitConfig {public static final String RPC_QUEUE1 = "queue_1";public static final String RPC_QUEUE2 = "queue_2";public static final String RPC_EXCHANGE = "rpc_exchange";/*** 配置消息发送队列*/@BeanQueue msgQueue() {return new Queue(RPC_QUEUE1);}/*** 设置返回队列*/@BeanQueue replyQueue() {return new Queue(RPC_QUEUE2);}/*** 设置交换机*/@BeanTopicExchange exchange() {return new TopicExchange(RPC_EXCHANGE);}/*** 请求队列和交换器绑定*/@BeanBinding msgBinding() {return BindingBuilder.bind(msgQueue()).to(exchange()).with(RPC_QUEUE1);}/*** 返回队列和交换器绑定*/@BeanBinding replyBinding() {return BindingBuilder.bind(replyQueue()).to(exchange()).with(RPC_QUEUE2);}
}

  4. 消息消费

    @RestController
@Slf4j
public class RpcConsumerReceiver {/** 服务端首先收到消息并打印出来。* 服务端提取出原消息中的 correlation_id。* 服务端调用 sendAndReceive 方法，将消息发送给 RPC_QUEUE2 队列，同时带上 correlation_id 参数。*/@Resourceprivate RabbitTemplate rabbitTemplate;@RabbitListener(queues = RabbitConfig.RPC_QUEUE1)public void process(Message msg) {log.info("server receive : {}",msg.toString());Message response = MessageBuilder.withBody(("i'm receive:"+new String(msg.getBody())).getBytes()).build();CorrelationData correlationData = new CorrelationData(msg.getMessageProperties().getCorrelationId());rabbitTemplate.sendAndReceive(RabbitConfig.RPC_EXCHANGE, RabbitConfig.RPC_QUEUE2, response, correlationData);}
}

7.发布者确认（Publisher Confirms）

  • 与发布者进行可靠的发布确认，发布者确认是RabbitMQ扩展，可以实现可靠的发布。
  • 在通道上启用发布者确认后，RabbitMQ将异步确认发送者发布的消息，这意味着它们已在服务器端处理
应用场景
  • 对于消息可靠性要求较高，比如钱包扣款

RabbitMQ的四种交换机

1.直连交换机（Direct exchange）

  • 具有路由功能的交换机，绑定到此交换机的时候需要指定一个routing_key，交换机发送消息的时候需要routing_key，会将消息发送道对应的队列
  • DirectExchange 的路由策略是将消息队列绑定到一个 DirectExchange 上，当一条消息到达 DirectExchange 时会被转发到与该条消息 routing key 相同的 Queue 上
  • 例如消息队列名为 “hello-queue”，则 routingkey 为 “hello-queue” 的消息会被该消息队列接收。
代码示例
  1. 配置类
    @Configuration
public class RabbitDirectConfig {//首先提供一个消息队列Queue，然后创建一个DirectExchange对象，三个参数分别是名字，重启后是否依然有效以及长期未用时是否删除。//创建一个Binding对象将Exchange和Queue绑定在一起。//DirectExchange和Binding两个Bean的配置可以省略掉，即如果使用DirectExchange，可以只配置一个Queue的实例即可。public final static String DIRECTNAME = "mq-direct";@Beanpublic Queue queue() {return new Queue("hello-queue");}@Beanpublic DirectExchange directExchange() {return new DirectExchange(DIRECTNAME, true, false);}@Beanpublic Binding binding() {return BindingBuilder.bind(queue()).to(directExchange()).with("direct");}
}
  2. 消费者
    通过 @RabbitListener 注解指定一个方法是一个消息消费方法，方法参数就是所接收到的消息
    @Component
public class DirectReceiver {@RabbitListener(queues = "hello-queue")public void handler1(String msg) {System.out.println("DirectReceiver:" + msg);}
}
  3. 生产者发送消息
    @RestController
public class SendController {@Resourceprivate RabbitTemplate rabbitTemplate;@RequestMapping("/send")public String send(){rabbitTemplate.convertAndSend("hello-queue", "hello direct!");return "success";}
}

2.扇形交换机（Fanout exchange）

  • 广播消息到所有队列，没有任何处理，速度最快
  • FanoutExchange 的数据交换策略是把所有到达 FanoutExchange 的消息转发给所有与它绑定的 Queue 上，在这种策略中，routingkey 将不起任何作用
代码示例

  1. 配置类

    @Configuration
public class RabbitFanoutConfig {//在这里首先创建 FanoutExchange，参数含义与创建 DirectExchange 参数含义一致，// 然后创建两个 Queue，再将这两个 Queue 都绑定到 FanoutExchange 上public final static String FANOUTNAME = "mq-fanout";@Beanpublic FanoutExchange fanoutExchange() {return new FanoutExchange(FANOUTNAME, true, false);}@Beanpublic Queue queueOne() {return new Queue("queue-one");}@Beanpublic Queue queueTwo() {return new Queue("queue-two");}@Beanpublic Binding bindingOne() {return BindingBuilder.bind(queueOne()).to(fanoutExchange());}@Beanpublic Binding bindingTwo() {return BindingBuilder.bind(queueTwo()).to(fanoutExchange());}
}

  2. 消费者

    @Component
public class FanoutReceiver {@RabbitListener(queues = "queue-one")public void receiver1(String message) {System.out.println("FanoutReceiver:receiver1:" + message);}@RabbitListener(queues = "queue-two")public void receiver2(String message) {System.out.println("FanoutReceiver:receiver2:" + message);}
}

  3. 生产者发送消息

    @RestController
public class SendController {@Resourceprivate RabbitTemplate rabbitTemplate;@RequestMapping("/send")public String send(){//注意这里发送消息时不需要 routingkey，指定 exchange 即可，routingkey 可以直接传一个 nullrabbitTemplate.convertAndSend(RabbitFanoutConfig.FANOUTNAME,null, "hello fanout!");return "success";}
}

3.主题交换机（Topic exchange）

  • 在直连交换机基础上增加模式匹配，也就是对routing_key进行模式匹配，*代表一个单词，#代表多个单词
  • TopicExchange 是比较复杂但是也比较灵活的一种路由策略，在 TopicExchange 中，Queue 通过 routingkey 绑定到 TopicExchange 上，
  • 当消息到达 TopicExchange 后，TopicExchange 根据消息的 routingkey 将消息路由到一个或者多个 Queue 上
代码示例

  1. 配置类

    @Configuration
public class RabbitTopicConfig {/*** 首先创建 TopicExchange，参数和前面的一致。* 然后创建三个 Queue，第一个 Queue 用来存储和 “xiaomi” 有关的消息，* 第二个 Queue 用来存储和 “huawei” 有关的消息，* 第三个 Queue 用来存储和 “phone” 有关的消息。* * 将三个 Queue 分别绑定到 TopicExchange 上，* 第一个 Binding 中的 “xiaomi.#” 表示消息的 routingkey 凡是以 “xiaomi” 开头的，都将被路由到名称为 “xiaomi” 的 Queue 上，* 第二个 Binding 中的 “huawei.#” 表示消息的 routingkey 凡是以 “huawei” 开头的，都将被路由到名称为 “huawei” 的 Queue 上，* 第三个 Binding 中的 “#.phone.#” 则表示消息的 routingkey 中凡是包含 “phone” 的，都将被路由到名称为 “phone” 的 Queue 上。*/public final static String TOPICNAME = "mq-topic";@Beanpublic TopicExchange topicExchange() {return new TopicExchange(TOPICNAME, true, false);}@Beanpublic Queue xiaomi() {return new Queue("xiaomi");}@Beanpublic Queue huawei() {return new Queue("huawei");}@Beanpublic Queue phone() {return new Queue("phone");}@Beanpublic Binding xiaomiBinding() {return BindingBuilder.bind(xiaomi()).to(topicExchange()).with("xiaomi.#");}@Beanpublic Binding huaweiBinding() {return BindingBuilder.bind(huawei()).to(topicExchange()).with("huawei.#");}@Beanpublic Binding phoneBinding() {return BindingBuilder.bind(phone()).to(topicExchange()).with("#.phone.#");}
}

  2. 消费者

    @Component
public class TopicReceiver {@RabbitListener(queues = "phone")public void receiver1(String message) {System.out.println("PhoneReceiver:" + message);}@RabbitListener(queues = "xiaomi")public void receiver2(String message) {System.out.println("XiaoMiReceiver:"+message);}@RabbitListener(queues = "huawei")public void receiver3(String message) {System.out.println("HuaWeiReceiver:"+message);}
}

  3. 生产者发送消息

@RestController
public class SendController {@Resourceprivate RabbitTemplate rabbitTemplate;@RequestMapping("/send")public String send(){//根据 RabbitTopicConfig 中的配置，//第一条消息将被路由到名称为 “xiaomi” 的 Queue 上，//第二条消息将被路由到名为 “huawei” 的 Queue 上，//第三条消息将被路由到名为 “xiaomi” 以及名为 “phone” 的 Queue 上，//第四条消息将被路由到名为 “huawei” 以及名为 “phone” 的 Queue 上，//最后一条消息则将被路由到名为 “phone” 的 Queue 上rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,"xiaomi.news","小米新闻..");rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,"huawei.news","华为新闻..");rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,"xiaomi.phone","小米手机..");rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,"huawei.phone","华为手机..");rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,"phone.news","手机新闻..");return "success";}
}

4.首部交换机（Headers exchange）

  • 忽略routing_key，使用Headers信息（一个Hash的数据结构）进行匹配，优势在于可以有更多更灵活的匹配规则
  • HeadersExchange 是一种使用较少的路由策略，HeadersExchange 会根据消息的 Header 将消息路由到不同的 Queue 上，这种策略也和 routingkey无关
代码示例

  1. 配置类

    @Configuration
public class RabbitHeaderConfig {/*** 这里的配置大部分和前面介绍的一样，差别主要体现的 Binding 的配置上，* 第一个 bindingName 方法中，* whereAny 表示消息的 Header 中只要有一个 Header 匹配上 map 中的 key/value，* 就把该消息路由到名为 “name-queue” 的 Queue 上，* 这里也可以使用 whereAll 方法，* 表示消息的所有 Header 都要匹配。* whereAny 和 whereAll 实际上对应了一个名为 x-match 的属性。* bindingAge 中的配置则表示只要消息的 Header 中包含 age，不管 age 的值是多少，* 都将消息路由到名为 “age-queue” 的 Queue 上*/public final static String HEADERNAME = "mq-header";@Beanpublic HeadersExchange headersExchange() {return new HeadersExchange(HEADERNAME, true, false);}@Beanpublic Queue queueName() {return new Queue("name-queue");}@Beanpublic Queue queueAge() {return new Queue("age-queue");}@Beanpublic Binding bindingName() {Map<String, Object> map = new HashMap<>();map.put("name", "mq");return BindingBuilder.bind(queueName()).to(headersExchange()).whereAny(map).match();}@Beanpublic Binding bindingAge() {return BindingBuilder.bind(queueAge()).to(headersExchange()).where("age").exists();}
}

  2. 消费者

    @Component
public class HeaderReceiver {//注意这里的参数用 byte 数组接收@RabbitListener(queues = "name-queue")public void receiver1(byte[] msg) {System.out.println("HeaderReceiver:name:" + new String(msg, 0, msg.length));}@RabbitListener(queues = "age-queue")public void receiver2(byte[] msg) {System.out.println("HeaderReceiver:age:" + new String(msg, 0, msg.length));}
}

  3. 生产者发送消息

package cn.smbms.controller;import cn.smbms.config.RabbitFanoutConfig;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;/*** @author: zjl* @datetime: 2024/5/9* @desc: */
@RestController
public class SendController {@Resourceprivate RabbitTemplate rabbitTemplate;@RequestMapping("/send")public String send(){//这里创建两条消息，两条消息具有不同的 header，不同 header 的消息将被发到不同的 Queue 中去Message nameMsg = MessageBuilder.withBody("hello header! name-queue".getBytes()).setHeader("name", "sang").build();Message ageMsg = MessageBuilder.withBody("hello header! age-queue".getBytes()).setHeader("age", "99").build();rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME, null, ageMsg);rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME, null, nameMsg);return "success";}
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/681479.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

雪球期权是什么意思？你了解雪球期权吗？

雪球期权是什么意思？你了解雪球期权吗？

今天期权懂带你了解雪球期权是什么意思&#xff1f;你了解雪球期权吗&#xff1f;雪球期权属于场外期权的一种&#xff0c;交易的方式只能通过线下跟券商询价的方式进行&#xff0c;类似场外个股期权的交易方式。 雪球期权是什么意思&#xff1f; 雪球期权&#xff0c;顾名思义…
阅读更多...
按键精灵纯本地离线文字识别插件

按键精灵纯本地离线文字识别插件

目的 按键精灵是一款可以模拟鼠标和键盘操作的自动化工具。它可以帮助用户自动完成一些重复的、繁琐的任务&#xff0c;节省大量人工操作的时间。但按键精灵是不包含图色功能&#xff0c;无法识别屏幕上的图像&#xff0c;根据图像的变化自动执行相应的操作。本篇文章主要讲解下…
阅读更多...
数据分析——业务数据描述

数据分析——业务数据描述

业务数据描述 前言一、数据收集数据信息来源企业内部数据源市场调查数据源公共数据源和第三方数据源 二、公司内部数据客户资料数据销售明细数据营销活动数据 三、市场调查数据观察法提问法实验法 四、公共数据五、第三方数据六、数据预处理七、数据清洗丢弃部分数据补全缺失的…
阅读更多...
OpenBayes 一周速览｜Apple 开源大模型 OpenELM 上线；字节发布 COCONut 首个全景图像分割数据集，入选 CVPR2024

OpenBayes 一周速览｜Apple 开源大模型 OpenELM 上线；字节发布 COCONut 首个全景图像分割数据集，入选 CVPR2024

公共资源速递 This Weekly Snapshots &#xff01; 5 个数据集&#xff1a; * COCONut 大规模图像分割数据集 * THUCNews 新闻数据集 * DuConv 对话数据集 * 安徽电信知道问答数据集 * Sentiment Analysis 中文情感分析数据集 2 个模型&#xff1a; * OpenELM-3B-Inst…
阅读更多...
Linux与windows网络管理

Linux与windows网络管理

文章目录 一、TCP/IP1.1、TCP/IP概念TCP/IP是什么TCP/IP的作用TCP/IP的特点TCP/IP的工作原理 1.2、TCP/IP网络发展史1.3、OSI网络模型1.4、TCP/IP网络模型1.5、linux中配置网络网络配置文件位置DNS配置文件主机名配置文件常用网络查看命令 1.6、windows中配置网络CMD中网络常用…
阅读更多...
ifconfig命令找不到 command not found

ifconfig命令找不到 command not found

问题 今天解决虚拟机的网络问题后&#xff0c;使用ifconfig发现报错命令未找到 解决方案 输入yum install ifconfi的程序安装包 yum install ifconfig 如果显示没有可用软件包 ifconfig&#xff0c;错误&#xff1a;。 就输入yum search ifconfig匹配安装包程序 yum searc…
阅读更多...
[C++核心编程-01]----C++内存四区详细解析

[C++核心编程-01]----C++内存四区详细解析

目录 前言 正文 01-内存区域简介 02-全局区 03-栈区 04-堆区 05-new操作符 总结 前言 当程序运行时&#xff0c;操作系统会为程序分配一块内存空间&#xff0c;这块内存空间被划分为不同的区域&#xff0c;每个区域有其独特的作用…
阅读更多...
SRM系统供应链库存协同提升企业服务水平

SRM系统供应链库存协同提升企业服务水平

SRM系统供应链库存协同是一种以提高供应链整体效率和竞争力为目标的管理方法。它涉及到企业与供应商之间的紧密合作&#xff0c;以实现库存优化、成本降低、风险分担和灵活响应市场变化等目标。 一、SRM供应链库存协同的概念和特点 SRM供应链库存协同是指企业与供应商之间通过…
阅读更多...
Core_Air724UG学习

Core_Air724UG学习

产品描述 Core_Air724UG核心板是基于Air724UG cat1模板制作的开发实验板。 该模块支持Lua二次开发或AT指令&#xff0c;方便开发者根据自己的需求灵活选择。 Core_Air724UG核心板专注于小型化&#xff0c;PCB尺寸4246mm&#xff0c;有12x22哥标准2.54mm排针管脚&#xff0c;其…
阅读更多...
一对一WebRTC视频通话系列（五）——综合调试和功能完善

一对一WebRTC视频通话系列（五）——综合调试和功能完善

本系列博客主要记录一对一WebRTC视频通话实现过程中的一些重点&#xff0c;代码全部进行了注释&#xff0c;便于理解WebRTC整体实现。 本专栏知识点是通过<零声教育>的音视频流媒体高级开发课程进行系统学习&#xff0c;梳理总结后写下文章&#xff0c;对音视频相关内容感…
阅读更多...
Unity VR在编辑器下开启Quest3透视（PassThrough）功能

Unity VR在编辑器下开启Quest3透视（PassThrough）功能

现在有个需求是PC端串流在某些特定时候需要开启透视。我研究了两天发现一些坑，记录一下方便查阅，也给没踩坑的朋友一些思路方案。 先说结论，如果要打PC端或者在Unity编辑器中开启，那么OpenXR当前是不行的可能还需要一个长期的过程，必须需要切换到Oculus。当然Unity官方指…
阅读更多...
VisualGDB：Linux静态库项目创建、编译及库的使用

VisualGDB：Linux静态库项目创建、编译及库的使用

接上篇《VisualGDB&#xff1a;Linux动态库项目创建、编译及库的使用》&#xff0c;静态库的创建和使用与动态库基本无差别&#xff0c;唯一需要做的就是指定项目生成静态库。 一、指定项目生成静态库 二、重新构建和编译项目 这里注意&#xff0c;同样要copy一个libxxx.so格式…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023