Springboot+RabbitMQ实战demo

 

rabbitmq依赖包

<!-- rabbitmq依赖 begin -->
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!-- rabbitmq依赖 end -->

 

 

1、application.yml配置文件

spring:
  # rabbitmq配置
  rabbitmq:
    host: amqp.cn-shenzhen-429403-a.aliyuncs.com
    port: 5672
    username: lynch
    password: lynch
    virtual-host: order-vhost
    
    ##RabbitMQ生产端核心配置
    connection-timeout: 5000
    publisher-confirms: true
    publisher-returns: true
    
    ##RabbitMQ消费端核心配置
    listener:
       simple:
          #设置消费端手动 ack
          acknowledge-mode: manual
          #消费者最小数量
          concurrency: 5
          #消费之最大数量
          max-concurrency: 10
          
    template:
      #ReturnCallback的时候代表消息不可达，设置 broker不自动删除该消息，而是返回到生产端，让我们进行一些后续的处理
      mandatory: true
      # 启用重试机制,重试间隔时间为2s,最多重试3次
      retry:
        enabled: true
        initial-interval: 2s
        max-attempts: 3

 

 

2、RabbitMQConfig.java——RabbitMQ配置类

package io.geekidea.springbootplus.config; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; @Configuration @ComponentScan({"com.xukang.*", "io.geekidea.*"}) public class RabbitMQConfig { @Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { RabbitTemplate rabbitTemplate = new RabbitTemplate(); rabbitTemplate.setConnectionFactory(connectionFactory); // 设置序列化策略
 rabbitTemplate.setMessageConverter(jsonMessageConverter()); return rabbitTemplate; } @Bean public MessageConverter jsonMessageConverter() { return new Jackson2JsonMessageConverter(); } }

 

3、生产端单元测试

package com.xukang.common; import java.util.Date; import java.util.UUID; import org.junit.Test; import org.springframework.amqp.rabbit.connection.CorrelationData; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback; import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback; import org.springframework.beans.factory.annotation.Autowired; import com.baomidou.mybatisplus.core.toolkit.IdWorker; import com.xukang.order.entity.Order; import cn.hutool.core.date.DateUnit; import cn.hutool.core.date.DateUtil; import io.geekidea.springbootplus.BaseTest; import io.geekidea.springbootplus.config.constant.DatePattern; import lombok.extern.slf4j.Slf4j; @Slf4j public class RabbitmqTest extends BaseTest { @Autowired private RabbitTemplate rabbitTemplate; @Test public void test() { System.out.println(rabbitTemplate); } /** * 发送direct模式消息 * * @throws Exception */ @Test public void testDirectQueue() throws Exception { Order order = new Order(); rabbitTemplate.setConfirmCallback(confirmCallback); rabbitTemplate.setReturnCallback(returnCallback); //id + 时间戳 全局唯一
        String id = UUID.randomUUID().toString(); log.info(String.format("sendOrder id: %s", id)); CorrelationData correlationData = new CorrelationData(id); order.setId(IdWorker.getId()); order.setOrderSn("order-1"); Date date = new Date(); System.out.println(DateUtil.format(date, DatePattern.YYYY_MM_DD_HH_MM_SS)); order.setCreateTime(date); //rabbitTemplate.convertAndSend("amq.direct", "order", order, correlationData); // 执行发送消息到指定队列
        rabbitTemplate.convertAndSend("amq.direct", "order", order, message -> { // 设置延迟，单位：毫秒值
            message.getMessageProperties().setHeader("delay", 20000); return message; }, correlationData); } /** * 发送topic模式消息 * * @throws Exception */ @Test public void testTopicQueue() throws Exception { Order order = new Order(); rabbitTemplate.setConfirmCallback(confirmCallback); rabbitTemplate.setReturnCallback(returnCallback); //id + 时间戳 全局唯一
        String id = UUID.randomUUID().toString(); log.info(String.format("sendOrder id: %s", id)); CorrelationData correlationData = new CorrelationData(id); order.setId(IdWorker.getId()); order.setOrderSn("order001"); rabbitTemplate.convertAndSend("amq.topic", "order.1", order, correlationData); order.setId(IdWorker.getId()); order.setOrderSn("order002"); rabbitTemplate.convertAndSend("amq.topic", "order.2", order, correlationData); } //回调函数: confirm确认
    final ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() { @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { log.info("消息唯一标识：{}", correlationData); log.info("确认结果：{}", ack); if(!ack){ log.info("异常处理...."); log.info("失败原因：{}", cause); } } }; //回调函数: return返回 //启动消息失败返回，比如路由不到队列时触发回调
    final ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() { @Override public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText, String exchange, String routingKey) { log.info("消息主体 message : "+message); log.info("消息主体 message : "+replyCode); log.info("描述："+replyText); log.info("消息使用的交换器 exchange : "+exchange); log.info("消息使用的路由键 routing : "+routingKey); } }; }

 

 

 

4、direct模式消费端代码

package com.xukang.foobar.mq; import org.springframework.amqp.rabbit.annotation.Exchange; import org.springframework.amqp.rabbit.annotation.Queue; import org.springframework.amqp.rabbit.annotation.QueueBinding; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.support.AmqpHeaders; import org.springframework.messaging.Message; import org.springframework.stereotype.Component; import com.rabbitmq.client.Channel; import com.xukang.order.entity.Order; /** * direct模式消费端 * * @author PC * */ @Component public class DirectReceiver { @RabbitListener(bindings = @QueueBinding(value = @Queue(value = "order-queue", durable = "true"), exchange = @Exchange(value = "amq.direct", durable = "true", type = "direct", ignoreDeclarationExceptions = "true"), key = "order")) @RabbitHandler public void onMessage(Message<Order> message, Channel channel) throws Exception { Order order = message.getPayload(); System.out.println("消费端Payload: " + order); Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG); // 手工ACK
        channel.basicAck(deliveryTag, false); } }

 

5、topic模式消费端代码

package com.xukang.foobar.mq; import java.io.IOException; import java.util.List; import java.util.Map; import org.springframework.amqp.rabbit.annotation.Exchange; import org.springframework.amqp.rabbit.annotation.Queue; import org.springframework.amqp.rabbit.annotation.QueueBinding; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.support.AmqpHeaders; import org.springframework.messaging.Message; import org.springframework.stereotype.Component; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.xukang.order.entity.Order; /** * topic模式消费端 * * @author PC * */ @Component public class TopicReceiver { @RabbitListener(bindings = @QueueBinding(value = @Queue(value = "topic-queue", durable = "true"), exchange = @Exchange(value = "amq.topic", durable = "true", type = "topic", ignoreDeclarationExceptions = "true"), key = "order.*")) @RabbitHandler public void onMessage(Message<Order> message, Channel channel) throws Exception { Long deliveryTag = 0L; try { deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG); Order order = message.getPayload(); System.out.println("消费端Payload: " + order); if(order.getOrderSn().equals("order001")) { //丢弃这条消息
                channel.basicNack(deliveryTag, false, true); return; } // 手工ACK
            channel.basicAck(deliveryTag, false); } catch (IOException e) { e.printStackTrace(); //丢弃这条消息
            channel.basicNack(deliveryTag, false, true); } } }