当前位置：首页 > news >正文

Springboot结合RabbitMQ

news 2025/7/26 20:33:55

pom.xml

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

application.yaml

spring:rabbitmq:host: 127.0.0.1username: guestpassword: guestport: 5672virtual-host: test_vhostconnection-timeout: 600000# 关闭自动ack，设置成手动acklistener:simple:acknowledge-mode: manualprefetch: 10concurrency: 2max-concurrency: 5

配置文件，也可以使用@PostConstruct

@Configuration
public class RabbitMQConfig {/*** 绑定*/public static final String EXCHANGE_NAME="springboot_topic_exchange";public static final String QUEUE_NAME="springboot_queue";@Bean("exchange")public Exchange exchange(){return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();}@Bean("queue")public Queue queue(){return QueueBuilder.durable(QUEUE_NAME).build();}@Bean()public Binding bindQueueExchange(@Qualifier("exchange")Exchange exchange,@Qualifier("queue")Queue queue){return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();}/** Fanout 模式*/public static final String EXCHANGE_NAME2="my_topic_exchange";public static final String QUEUE_NAME2="my_queue";//创建交换机@Beanpublic FanoutExchange fannoutExchange(){return new FanoutExchange(EXCHANGE_NAME2,true,false);}//创建队列@Beanpublic Queue fannoutQueue(){return new Queue(QUEUE_NAME2,true,false,false);}//声明绑定关系@Beanpublic Binding bindQueue(@Qualifier("exchange")Exchange exchange,@Qualifier("queue")Queue queue){return BindingBuilder.bind(fannoutQueue()).to(fannoutExchange());}/** topic 模式*/@Beanpublic Queue topicQueue(){return new Queue(QUEUE_NAME,true,false,false);}//创建交换机@Beanpublic TopicExchange topicExchange(){return new TopicExchange(EXCHANGE_NAME,true,false);}//声明绑定关系@Beanpublic Binding bindTopicQueue(){return BindingBuilder.bind(topicQueue()).to(topicExchange()).with("yuyang.#");}

消费者代码

@Component
@Slf4j
public class MessageListener  {@RabbitListener(queues="springboot_queue")public void process(Message message) {log.info("helloWorld模式 received message : {}",new String(message.getBody()));}}

生产者代码

	@Autowiredprivate RabbitTemplate rabbitTemplate;// pub/sub 发布订阅模式   交换机类型 fanout@GetMapping("/fanoutSend")public String fanoutSend() throws AmqpException, UnsupportedEncodingException {//fanout模式只往exchange里发送消息。分发到exchange下的所有queuerabbitMQService.sendMessage("my_topic_exchange","*","fanoutSend");return "message sended : "+message;}//topic 工作模式   交换机类型 topic@GetMapping("/topicSend")public String topicSend() throws AmqpException, UnsupportedEncodingException {//fanout模式只往exchange里发送消息。分发到exchange下的所有queuerabbitTemplate.convertAndSend("spring_topic_exchange","yuyang.test","sadedf");return "ok";}

手动ack

	@RabbitListener(queues = "springboot_queue")public void listenerQueue(Message message, Channel channel) throws IOException {log.info(new String(message.getBody()));channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}