前言
当系统中出现“生产“和“消费“的速度或稳定性等因素不一致的时候，就需要消息队列。
名词
- exchange： 交换机
- routingkey： 路由key
- queue：队列
- 控制台端口：15672
exchange和queue是需要绑定在一起的，然后消息发送到exchange再由exchange通过routingkey发送到对应的队列中。
使用场景
- 1.技能订单3分钟自动取消，改变状态
- 2.直播开始前15分钟提醒
- 3.直播状态自动结束
参考链接
https://juejin.im/entry/5a17909a518825329314397d
https://www.jianshu.com/p/ea953f633466
流程
生产者发送消息 —> order_pre_exchange交换机 —> order_per_ttl_delay_queue队列
—> 时间到期 —> order_delay_exchange交换机 —> order_delay_process_queue队列 —> 消费者
第一步：在pom文件中添加
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>
第二步：在application.properties文件中添加
spring.rabbitmq.host=172.xx.xx.xxx
spring.rabbitmq.port=5672
spring.rabbitmq.username=rabbit
spring.rabbitmq.password=123456
spring.rabbitmq.virtual-host=/
spring.rabbitmq.connection-timeout=15000
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.template.mandatory=true
第三步：配置 OrderQueueConfig
1 package com.tuohang.platform.config; 2 3 import org.springframework.amqp.core.Binding; 4 import org.springframework.amqp.core.BindingBuilder; 5 import org.springframework.amqp.core.DirectExchange; 6 import org.springframework.amqp.core.Queue; 7 import org.springframework.amqp.core.QueueBuilder; 8 import org.springframework.amqp.rabbit.connection.ConnectionFactory; 9 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; 10 import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; 11 import org.springframework.context.annotation.Bean; 12 import org.springframework.context.annotation.Configuration; 13 14 15 /** 16 * rabbitMQ的队列设置（生产者发送的消息，永远是先进入exchange，再通过路由，转发到队列） 17 * 18 * 19 * @author Administrator 20 * @version 1.0 21 * @Date 2018年9月18日 22 */ 23 @Configuration 24 public class OrderQueueConfig { 25 26 /** 27 * 订单缓冲交换机名称 28 */ 29 public final static String ORDER_PRE_EXCHANGE_NAME = "order_pre_exchange"; 30 31 /** 32 * 发送到该队列的message会在一段时间后过期进入到order_delay_process_queue 【队列里所有的message都有统一的失效时间】 33 */ 34 public final static String ORDER_PRE_TTL_DELAY_QUEUE_NAME = "order_pre_ttl_delay_queue"; 35 36 /** 37 * 订单的交换机DLX 名字 38 */ 39 final static String ORDER_DELAY_EXCHANGE_NAME = "order_delay_exchange"; 40 41 /** 42 * 订单message时间过期后进入的队列，也就是订单实际的消费队列 43 */ 44 public final static String ORDER_DELAY_PROCESS_QUEUE_NAME = "order_delay_process_queue"; 45 46 /** 47 * 订单在缓冲队列过期时间（毫秒）30分钟 48 */ 49 public final static int ORDER_QUEUE_EXPIRATION = 1800000; 50 51 /** 52 * 订单缓冲交换机 53 * 54 * @return 55 */ 56 @Bean 57 public DirectExchange preOrderExange() { 58 return new DirectExchange(ORDER_PRE_EXCHANGE_NAME); 59 } 60 61 /** 62 * 创建order_per_ttl_delay_queue队列，订单消息经过缓冲交换机，会进入该队列 63 * 64 * @return 65 */ 66 @Bean 67 public Queue delayQueuePerOrderTTLQueue() { 68 return QueueBuilder.durable(ORDER_PRE_TTL_DELAY_QUEUE_NAME) 69 .withArgument("x-dead-letter-exchange", ORDER_DELAY_EXCHANGE_NAME) // DLX 70 .withArgument("x-dead-letter-routing-key", ORDER_DELAY_PROCESS_QUEUE_NAME) // dead letter携带的routing key 71 .withArgument("x-message-ttl", ORDER_QUEUE_EXPIRATION) // 设置订单队列的过期时间 72 .build(); 73 } 74 75 /** 76 * 将order_pre_exchange绑定到order_pre_ttl_delay_queue队列 77 * 78 * @param delayQueuePerOrderTTLQueue 79 * @param preOrderExange 80 * @return 81 */ 82 @Bean 83 public Binding queueOrderTTLBinding(Queue delayQueuePerOrderTTLQueue, DirectExchange preOrderExange) { 84 return BindingBuilder.bind(delayQueuePerOrderTTLQueue).to(preOrderExange).with(ORDER_PRE_TTL_DELAY_QUEUE_NAME); 85 } 86 87 /** 88 * 创建订单的DLX exchange 89 * 90 * @return 91 */ 92 @Bean 93 public DirectExchange delayOrderExchange() { 94 return new DirectExchange(ORDER_DELAY_EXCHANGE_NAME); 95 } 96 97 /** 98 * 创建order_delay_process_queue队列，也就是订单实际消费队列 99 * 100 * @return 101 */ 102 @Bean 103 public Queue delayProcessOrderQueue() { 104 return QueueBuilder.durable(ORDER_DELAY_PROCESS_QUEUE_NAME).build(); 105 } 106 107 /** 108 * 将DLX绑定到实际消费队列 109 * 110 * @param delayProcessOrderQueue 111 * @param delayExchange 112 * @return 113 */ 114 @Bean 115 public Binding dlxOrderBinding(Queue delayProcessOrderQueue, DirectExchange delayOrderExchange) { 116 return BindingBuilder.bind(delayProcessOrderQueue).to(delayOrderExchange).with(ORDER_DELAY_PROCESS_QUEUE_NAME); 117 } 118 119 /** 120 * 监听订单实际消费者队列order_delay_process_queue 121 * 122 * @param connectionFactory 123 * @param processReceiver 124 * @return 125 */ 126 @Bean 127 public SimpleMessageListenerContainer orderProcessContainer(ConnectionFactory connectionFactory, 128 OrderProcessReceiver processReceiver) { 129 SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); 130 container.setConnectionFactory(connectionFactory); 131 container.setQueueNames(ORDER_DELAY_PROCESS_QUEUE_NAME); // 监听order_delay_process_queue 132 container.setMessageListener(new MessageListenerAdapter(processReceiver)); 133 return container; 134 } 135 }
消费者 OrderProcessReceiver ：
1 package com.tuohang.platform.config; 2 3 import java.util.Objects; 4 5 import org.apache.tools.ant.types.resources.selectors.Date; 6 import org.slf4j.Logger; 7 import org.slf4j.LoggerFactory; 8 import org.springframework.amqp.core.Message; 9 import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; 10 import org.springframework.stereotype.Component; 11 import com.rabbitmq.client.Channel; 12 13 /** 14 * 订单延迟处理消费者 15 * 16 * 17 * @author Administrator 18 * @version 1.0 19 * @Date 2018年9月18日 20 */ 21 @Component 22 public class OrderProcessReceiver implements ChannelAwareMessageListener { 23 24 private static Logger logger = LoggerFactory.getLogger(OrderProcessReceiver.class); 25 26 String msg = "The failed message will auto retry after a certain delay"; 27 28 @Override 29 public void onMessage(Message message, Channel channel) throws Exception { 30 try { 31 processMessage(message); 32 } catch (Exception e) { 33 // 如果发生了异常，则将该消息重定向到缓冲队列，会在一定延迟之后自动重做 34 channel.basicPublish(OrderQueueConfig.ORDER_PRE_EXCHANGE_NAME, OrderQueueConfig.ORDER_PRE_TTL_DELAY_QUEUE_NAME, null, 35 msg.getBytes()); 36 } 37 } 38 39 /** 40 * 处理订单消息，如果订单未支付，取消订单（如果当消息内容为FAIL_MESSAGE的话，则需要抛出异常） 41 * 42 * @param message 43 * @throws Exception 44 */ 45 public void processMessage(Message message) throws Exception { 46 String realMessage = new String(message.getBody()); 47 logger.info("Received <" + realMessage + ">"); 48 // 取消订单 49 if(!Objects.equals(realMessage, msg)) { 50 // SpringKit.getBean(ITestService.class).resetSexById(Long.valueOf(realMessage)); 51 System.out.println("测试111111-----------"+new Date()); 52 System.out.println(message); 53 } 54 } 55 }
或者
1 /** 2 * 测试 rabbit 消费者 3 * 4 * 5 * @author Administrator 6 * @version 1.0 7 * @Date 2018年9月25日 8 */ 9 @Component 10 @RabbitListener(queues = TestQueueConfig.TEST_DELAY_PROCESS_QUEUE_NAME) 11 public class TestProcessReceiver { 12 13 private static Logger logger = LoggerFactory.getLogger(TestProcessReceiver.class); 14 15 String msg = "The failed message will auto retry after a certain delay"; 16 17 @RabbitHandler 18 public void onMessage(Message message, Channel channel) throws Exception { 19 try { 20 processMessage(message); 21 //告诉服务器收到这条消息 已经被我消费了 可以在队列删掉；否则消息服务器以为这条消息没处理掉 后续还会在发 22 channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); 23 } catch (Exception e) { 24 // 如果发生了异常，则将该消息重定向到缓冲队列，会在一定延迟之后自动重做 25 channel.basicPublish(TestQueueConfig.TEST_PRE_EXCHANGE_NAME, TestQueueConfig.TEST_PRE_TTL_DELAY_QUEUE_NAME, null, 26 msg.getBytes()); 27 } 28 } 29 30 /** 31 * 处理订单消息，如果订单未支付，取消订单（如果当消息内容为FAIL_MESSAGE的话，则需要抛出异常） 32 * 33 * @param message 34 * @throws Exception 35 */ 36 public void processMessage(Message message) throws Exception { 37 String realMessage = new String(message.getBody()); 38 logger.info("Received < " + realMessage + " >"); 39 // 取消订单 40 if(!Objects.equals(realMessage, msg)) { 41 System.out.println("测试111111-----------"+new Date()); 42 }else { 43 System.out.println("rabbit else..."); 44 } 45 } 46 }
生产者
1 /** 2 * 测试rabbitmq 3 * 4 * @return 5 */ 6 @RequestMapping(value = "/testrab") 7 public String testraa() { 8 GenericResult gr = null; 9 try { 10 String name = "test_pre_ttl_delay_queue"; 11 long expiration = 10000;//10s 过期时间 12 rabbitTemplate.convertAndSend(name,String.valueOf(123456)); 13 // 在单个消息上设置过期时间 14 //rabbitTemplate.convertAndSend(name,(Object)String.valueOf(123456), new ExpirationMessagePostProcessor(expiration)); 15 16 17 } catch (ServiceException e) { 18 e.printStackTrace(); 19 gr = new GenericResult(StateCode.ERROR, languageMap.get("network_error"), e.getMessage()); 20 } 21 22 return getWrite(gr); 23 }