文档章节

java B2B2C 仿淘宝电子商城系统-基于Rabbitmq实现延迟消息

明理萝
 明理萝
发布于 2019/04/02 16:07
字数 1621
阅读 9
收藏 4

1.  预备知识

1.1 消息传递

首先我们知道消费者是从队列中获取消息的,那么消息是如何到达队列的?

当我们发送一条消息时,首先会发给交换器(exchange),交换器根据规则(路由键:routing key)将会确定消息投递到那个队列(queue)。

需要JAVA Spring Cloud大型企业分布式微服务云构建的B2B2C电子商务平台源码 一零三八七七四六二六

带着这几个关键字:交换器、路由键和队列。

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 to
another exchange when any of the following events occur:

The message is rejected (basic.reject or basic.nack) with
requeue=false, The TTL for the message expires; or The queue length
limit 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)来重新发送消息到另外一个队列,而这个队列将是最终的消费队列。

2.  具体实现

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;

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));
    }
}

  

 

© 著作权归作者所有

明理萝
粉丝 73
博文 374
码字总数 356037
作品 0
深圳
私信 提问
加载中

评论(1)

明理萝
明理萝 博主
欢迎大家一起交流
java B2B2C Springboot电子商城系统-消息队列之 RabbitMQ

常见的消息队列 需要JAVA Spring Cloud大型企业分布式微服务云构建的B2B2C电子商务平台源码请加企鹅求求: :二一四七七七五六三三 目前业界有四款常用的消息队列,它们分别是RabbitMQ、Roc...

it菲菲
2018/12/14
0
0
java B2B2C Springcloud仿淘宝电子商城系统-消息总线(Spring Cloud Bus)

spring CloudBus 将分布式的节点和轻量的消息代理连接起来。这可以用于广播配置文件的更改或者其他的管理工作。一个关键的思想就是,消息总线可以为微服务做监控,也可以作为应用程序之间相互...

明理萝
2019/04/04
14
1
Spring Boot(十四)RabbitMQ延迟队列

一、前言 延迟队列的使用场景:1.未按时支付的订单,30分钟过期之后取消订单;2.给活跃度比较低的用户间隔N天之后推送消息,提高活跃度;3.过1分钟给新注册会员的用户,发送注册邮件等。 实现...

王磊的博客
2018/11/16
407
0
(十五)springcloud微服务多用户商城系统java_代码开源_B2B电商系统_B2C电商系统-Springboot整合RabbitMQ

这篇文章带你了解怎么整合RabbitMQ服务器,并且通过它怎么去发送和接收消息。我将构建一个springboot工程,通过RabbitTemplate去通过MessageListenerAdapter去订阅一个POJO类型的消息。 准备...

2019/01/23
0
0
RabbitMQ入门(1)--介绍

前面声明本文都是RabbitMQ的官方指南翻译过来的,由于本人水平有限难免有翻译不当的地方,如发现不对的地方,请联系下我,好及时改正。好了,正文开始: RabbitMQ 是一个消息代理。这主要的原...

-悟空-
2015/02/24
2.4W
18

没有更多内容

加载失败,请刷新页面

加载更多

UGUI图片层级和渲染顺序的奇怪关系

之前见别人的文章总是说,在Hierachy下,相同图集的图片要连续排列,这样Unity会对相同图集的图片进行合批,从而减少draw call。今天做了简单的试验发现情况并不是这么简单的。 第一种情况:...

myctrd
18分钟前
33
0
jQuery中$.each()方法的使用-Json对象-dom元素

对于循环我们首先会想到for循环,但是在前端对数组我们可以使用,但是对于json对象,想把对象中的属性的key-value循环去取出,那么for循环提供不了的。而each方法则给我们提供了便利,下面介...

imzchloe
28分钟前
50
0
Linuxprobe第六天

待补充

nt狮子男人
29分钟前
50
0
gem install:无法构建gem native扩展(找不到头文件)

我正在使用Fedora 14,我安装并运行了MySQL和MySQL服务器5.1.42。 现在我尝试以root用户身份执行此操作: gem install mysql 但我得到这个错误: Building native extensions. This could ...

技术盛宴
36分钟前
51
0
就8张图片带你搞清楚JS的原型链

JS(JavaScript)是目前互联网开发中十分重要的一门编程语言,他承载着网页、手机应用程序、硬件程序、微信、微信小程序中的各种特效及处理逻辑功能。

涂老师
39分钟前
52
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部