springboot+rabbitmq路由模式

路由模式是把队列通过rout绑定到交换机上

org.springframework.bootspring-boot-starter-webcom.alibabafastjson1.2.75org.springframework.bootspring-boot-starter-amqp
生产者

配置文件

server:port: 7001
spring:rabbitmq:host: localhostport: 5672username: xxpassword: xxvirtual-host: /


常量类

package com.example.rabbitmq.constant;public class Constant {//队列名称public static final String DIRECT_QUEUE_NAME02 &＃61; "testDirectQueue02";//队列名称public static final String DIRECT_QUEUE_NAME &＃61; "testDirectQueue";//交换机名称public static final String DIRECT_EXCHANGE_NAME &＃61; "testDirectExchange";//交换机绑定routpublic static final String DIRECT_ROUTING &＃61; "testDirectRouting";//交换机绑定routpublic static final String DIRECT_ROUTING02 &＃61; "testDirectRouting02";
}


rabbitmq配置类

package com.example.rabbitmq.config;import com.example.rabbitmq.constant.Constant;
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;/*** rabbitmq配置*/
&＃64;Configuration
public class RabbitConfig {// durable:是否持久化,默认是false,持久化队列&＃xff1a;会被存储在磁盘上&＃xff0c;当消息代理重启时仍然存在&＃xff0c;暂存队列&＃xff1a;当前连接有效// exclusive:默认也是false&＃xff0c;只能被当前创建的连接使用&＃xff0c;而且当连接关闭后队列即被删除。此参考优先级高于durable// autoDelete:是否自动删除&＃xff0c;当没有生产者或者消费者使用此队列&＃xff0c;该队列会自动删除。// return new Queue("TestDirectQueue",true,true,false);//一般设置一下队列的持久化就好,其余两个就是默认false&＃64;Beanpublic Queue testDirectQueue() {//第一个队列return new Queue(Constant.DIRECT_QUEUE_NAME, true);}// 路由模式 start// 生产者通过路由方式 把队列绑定到交换机上// 消息对于多个消费者来说 消息只能被消费一次// 生产消息的时候指定交换机和路由&＃xff0c;就会发送到绑定的队列上面// 消费者只要消费指定的队列就可以了 不会消费同一个交换机上的其他队列// 也就是说一个交换机可以绑定多个队列//Direct交换机 起名&＃xff1a;testDirectExchange&＃64;Beanpublic DirectExchange testDirectExchange() {return new DirectExchange(Constant.DIRECT_EXCHANGE_NAME, true, false);}//把第一个队列和交换机绑定&＃64;Beanpublic Binding bindingDirect() {return BindingBuilder.bind(testDirectQueue()).to(testDirectExchange()).with(Constant.DIRECT_ROUTING);}// 路由模式 end&＃64;Beanpublic Queue testDirectQueue02(){//第二个队列return new Queue(Constant.DIRECT_QUEUE_NAME02,true);}//第二个队列和交换机绑定&＃64;Beanpublic Binding bindingDirect02() {return BindingBuilder.bind(testDirectQueue02()).to(testDirectExchange()).with(Constant.DIRECT_ROUTING02);}
}


controller  这里总共调用了10次接口&＃xff0c;发现消费者总共消费了10次&＃xff0c;因为虽然是同一个交换机  但是绑定的队列不一样&＃xff0c;消费者消费的是队列 不是交换机

package com.example.rabbitmq.controller;import com.alibaba.fastjson.JSON;
import com.example.rabbitmq.constant.Constant;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.util.HashMap;
import java.util.Map;
import java.util.UUID;&＃64;RestController
public class RabbitmqController {&＃64;Autowiredprivate RabbitTemplate rabbitTemplate;&＃64;RequestMapping("/directMq")public void directMq() {Map map &＃61; new HashMap<>();map.put("messageId", UUID.randomUUID());map.put("messageData", "hello testDirect");map.put("createTime", System.currentTimeMillis());//Constant.DIRECT_ROUTING 对应的队列是testDirectQueuerabbitTemplate.convertAndSend(Constant.DIRECT_EXCHANGE_NAME, Constant.DIRECT_ROUTING, JSON.toJSONString(map));//Constant.DIRECT_ROUTING02 对应的队列是testDirectQueue02rabbitTemplate.convertAndSend(Constant.DIRECT_EXCHANGE_NAME, Constant.DIRECT_ROUTING02, JSON.toJSONString(map));//rabbitTemplate.convertAndSend(Constant.DIRECT_QUEUE_NAME,JSON.toJSONString(map));}
}

消费者

配置文件

server:port: 7002
spring:rabbitmq:host: localhostport: 5672username: xxpassword: xxvirtual-host: /listener:simple:acknowledge-mode: manualretry:enabled: trueprefetch: 1


消费者1代码

package com.example.rabbitmq.mq;import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;import java.io.IOException;
import java.nio.charset.StandardCharsets;&＃64;Component
public class RabbitmqListener {&＃64;RabbitListener(queues &＃61; "testDirectQueue")public void listener(Message message, &＃64;Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) {String msg &＃61; new String(message.getBody(), StandardCharsets.UTF_8);System.out.println("01&＃61;&＃61;" &＃43; msg);try {channel.basicAck(deliveryTag,false);} catch (IOException e) {e.printStackTrace();}}
}


消费者2代码

package com.example.rabbitmq.mq;import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;import java.io.IOException;
import java.nio.charset.StandardCharsets;&＃64;Component
public class RabbitmqListener {&＃64;RabbitListener(queues &＃61; "testDirectQueue")public void listener(Message message, &＃64;Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) {String msg &＃61; new String(message.getBody(), StandardCharsets.UTF_8);System.out.println("02&＃61;&＃61;" &＃43; msg);try {channel.basicAck(deliveryTag,false);} catch (IOException e) {e.printStackTrace();}}
}


发送10条消息查看消费情况

消费者1

消费者2

这里修改下代码&＃xff0c;

        消费者1消费队列testDirectQueue

        消费者2消费队列testDirectQueue02

只修改消费者2的代码即可

package com.example.rabbitmq.mq;import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;import java.io.IOException;
import java.nio.charset.StandardCharsets;&＃64;Component
public class RabbitmqListener {&＃64;RabbitListener(queues &＃61; "testDirectQueue02")public void listener(Message message, &＃64;Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) {String msg &＃61; new String(message.getBody(), StandardCharsets.UTF_8);System.out.println("02&＃61;&＃61;" &＃43; msg);try {channel.basicAck(deliveryTag,false);} catch (IOException e) {e.printStackTrace();}}
}


再次请求生产者接口10次 发现打印结果是

消费者1

消费者2

再次证明消费者消费的是队列不是交换机