java B2B2C 仿淘宝电子商城系统-基于Rabbitmq实现延迟消息
预备知识 1.1 消息传递 首先我们知道消费者是从队列中获取消息的，那么消息是如何到达队列的？ 当我们发送一条消息时，首先会发给交换器（exchange），交换器根据规则（路由键：routing key）将会确定消息投递到那个队列（queue）。 带着这几个关键字：交换器、路由键和队列。 1.2 交换器类型 如之前所说，交换器根据规则决定消息的路由方向。因此，rabbitmq的消息投递分类便是从交换器开始的，不同的交换器实现不同的路由算法便实现了不同的消息投递方式。 direct交换器 direct -> routingKey -> queue，相当一种点对点的消息投递，如果路由键匹配，就直接投递到相应的队列 fanout交换器 fanout交换器相当于实现了一（交换器）对多（队列）的广播投递方式 topic交换器 提供一种模式匹配的投递方式，我们可以根据主题来决定消息投递到哪个队列。 1.3 消息延迟 本文想要实现一个可延迟发送的消息机制。消息如何延迟？ ttl (time to live) 消息存活时间 ttl是指一个消息的存活时间。 Per-Queue Message TTL in Queues 引用官方的一句话： TTL can be set for a given queue by setting the x-message-ttl argument to queue.declare, or by setting the message-ttl policy. A message that has been in the queue for longer than the configured TTL is said to be dead.我们可以通过x-message-ttl设置一个队列中消息的过期时间，消息一旦过期，将会变成死信（dead-letter），可以选择重新路由。 Per-Message TTL in Publishers 引用官方的一句话： A TTL can be specified on a per-message basis, by setting the expiration field in the basic AMQP class when sending a basic.publish. The value of the expiration field describes the TTL period in milliseconds. The same constraints as for x-message-ttl apply. Since the expiration field must be a string, the broker will (only) accept the string representation of the number. 我们可以通过设置每一条消息的属性expiration，指定单条消息有效期。消息一旦过期，将会变成死信（dead-letter），可以选择重新路由。 重新路由-死信交换机(Dead Letter Exchanges)引用官方一句话： Dead Letter Exchanges Messages from a queue can be ‘dead-lettered’; that is, republished toanother exchange when any of the following events occur: The message is rejected (basic.reject or basic.nack) withrequeue=false, The TTL for the message expires; or The queue lengthlimit is exceeded. Dead letter exchanges (DLXs) are normal exchanges.They can be any of the usual types and are declared as usual.To set the dead letter exchange for a queue, set the x-dead-letter-exchange argument to the name of the exchange. 我们可以通过设置死信交换器（x-dead-letter-exchange）来重新发送消息到另外一个队列，而这个队列将是最终的消费队列。 具体实现 rabbitmq配置 属性文件-rabbitmq.properties 交换、路由等配置按照以上策略，其中，添加了prefetch参数来根据服务器能力控制消费数量。 连接用户名 mq.user =sms_user 密码 mq.password =123456 主机 mq.host =192.168.99.100 端口 mq.port =5672 默认virtual-host mq.vhost =/ the default cache size for channels is 25 mq.channelCacheSize =50 发送消息路由 sms.route.key =sms_route_key 延迟消息队列 sms.delay.queue =sms_delay_queue 延迟消息交换器 sms.delay.exchange =sms_delay_exchange 消息的消费队列 sms.queue =sms_queue 消息交换器 sms.exchange =sms_exchange 每秒消费消息数量 sms.prefetch =30 配置rabbitmq.xml <?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit" xmlns:context="http://www.springframework.org/schema/context" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.7.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> <context:property-placeholder location="rabbitmq.properties"/> <!--配置connection-factory，指定连接rabbit server参数 --> <rabbit:connection-factory id="connectionFactory" username="${mq.user}" password="${mq.password}" host="${mq.host}" port="${mq.port}" virtual-host="${mq.vhost}" /> <!--定义rabbit template用于数据的接收和发送 --> <rabbit:template id="amqpTemplate" connection-factory="connectionFactory" /> <!--通过指定下面的admin信息，当前producer中的exchange和queue会在rabbitmq服务器上自动生成 --> <rabbit:admin connection-factory="connectionFactory" /> <!--定义queue --> <rabbit:queue name="${sms.queue}" durable="true" auto-delete="false" exclusive="false" /> <!-- 创建延迟，有消息有效期的队列 --> <rabbit:queue name="${sms.delay.queue}" durable="true" auto-delete="false"> <rabbit:queue-arguments> <entry key="x-message-ttl"> <!-- 队列默认消息过期时间 --> <value type="java.lang.Long">3600000</value> </entry> <!-- 消息过期根据重新路由 --> <entry key="x-dead-letter-exchange" value="${sms.exchange}"/> </rabbit:queue-arguments> </rabbit:queue> <!-- 定义direct exchange，sms_queue --> <rabbit:direct-exchange name="${sms.exchange}" durable="true" auto-delete="false"> <rabbit:bindings> <rabbit:binding queue="${sms.queue}" key="${sms.route.key}"/> </rabbit:bindings> </rabbit:direct-exchange> <!-- 延迟消息配置，durable=true 持久化生效 --> <rabbit:direct-exchange name="${sms.delay.exchange}" durable="true" auto-delete="false"> <rabbit:bindings> <rabbit:binding queue="${sms.delay.queue}" key="${sms.route.key}"/> </rabbit:bindings> </rabbit:direct-exchange> <!-- 消息接收者 --> <bean id="messageReceiver" class="git.yampery.consumer.MsgConsumer"/> <!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象--> <rabbit:listener-container connection-factory="connectionFactory" prefetch="${sms.prefetch}"> <rabbit:listener queues="${sms.queue}" ref="messageReceiver"/> </rabbit:listener-container> </beans> 消息发布者 package git.yampery.producer; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import javax.annotation.Resource; /** * @decription MsgProducer * <p>生产者</p> * @author Yampery * @date 2018/2/11 11:44 */ @Component public class MsgProducer { @Resource private AmqpTemplate amqpTemplate; @Value("${sms.delay.exchange}") private String SMS_DELAY_EXCHANGE; @Value("${sms.exchange}") private String SMS_EXCHANGE; @Value("${sms.route.key}") private String SMS_ROUTE_KEY; /** * 延迟消息放入延迟队列中 * @param msg * @param expiration */ public void publish(String msg, String expiration) { amqpTemplate.convertAndSend(SMS_DELAY_EXCHANGE, SMS_ROUTE_KEY, msg, message -> { // 设置消息属性-过期时间 message.getMessageProperties().setExpiration(expiration); return message; }); } /** * 非延迟消息放入待消费队列 * @param msg */ public void publish(String msg) { amqpTemplate.convertAndSend(SMS_EXCHANGE, SMS_ROUTE_KEY, msg); } } 消费者 package git.yampery.consumer; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageListener; /** * @decription MsgConsumer * <p>消费者</p> * @author Yampery * @date 2018/2/11 11:43 */ public class MsgConsumer implements MessageListener { @Override public void onMessage(Message message) { String msg; try { // 线程每秒消费一次 Thread.sleep(1000); msg = new String(message.getBody(), "utf-8"); System.out.println(msg); } catch (Exception e) { // 这里并没有对服务异常等失败的消息做处理，直接丢弃了 // 防止因业务异常导致消息失败造成unack阻塞再队列里 // 可以选择路由到另外一个专门处理消费失败的队列 return; } } } 测试 package git.yampery.mq; //需要JAVA Spring Cloud大型企业分布式微服务云构建的B2B2C电子商务平台源码 // 一零三八七七四六二六 import com.alibaba.fastjson.JSONObject; import git.yampery.producer.MsgProducer; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; import javax.annotation.Resource; /** * @decription TestMq * <p>测试</p> * @author Yampery * @date 2018/2/11 15:03 */ @RunWith(SpringRunner.class) @SpringBootTest public class TestMq { @Resource private MsgProducer producer; @Test public void testMq() { JSONObject jObj = new JSONObject(); jObj.put("msg", "这是一条短信"); producer.publish(jObj.toJSONString(), String.valueOf(10 * 1000)); } }
