使用RabbitMQ发送短信

Posted

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了使用RabbitMQ发送短信相关的知识，希望对你有一定的参考价值。


1、在项目中分别创建模块financial-core、financial-mq、financial-sms，如图：

使用RabbitMQ发送短信_maven

模块构成

<modules>
        <module>financial-common</module>
        <module>financial-base</module>
        <module>financial-core</module>
        <module>financial-gateway</module>       
        <module>financial-sms</module>
        <module>financial-mq</module>        
    </modules>

2、搭建RabbitMQ服务，参考下面文章

3、MQ模块financial-mq

使用RabbitMQ发送短信_java_02

3.1 在pom.xml添加依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>koo-financial-parent</artifactId>
        <groupId>com.koo</groupId>
        <version>1.0.0</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>financial-mq</artifactId>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
    </dependencies>

</project>

3.2 MQ Json字符串转换 配置

@Configuration
public class MQConfig 
    @Bean
    public MessageConverter messageConverter()
        //json字符串转换器
        return new Jackson2JsonMessageConverter();

3.3 定义交换机、路由、队列常量

public class MQConst 

    public static final String EXCHANGE_TOPIC_SMS = "exchange.topic.sms";//交换机
    public static final String ROUTING_SMS_ITEM = "routing.sms.item";//路由
    public static final String QUEUE_SMS_ITEM = "queue.sms.item";//消息队列

3.4 定义发送方法

@Service
@Slf4j
public class MQService 

    @Resource
    private AmqpTemplate amqpTemplate;

    public boolean sendMessage(String exchange, String routingKey, Object message) 
        log.info("发送消息。。。。。。");
        amqpTemplate.convertAndSend(exchange, routingKey, message);
        return true;

4、业务模块financial-core

4.1 在pom.xml中引入financial-mq模块

<dependencies>

        <dependency>
            <groupId>com.koo</groupId>
            <artifactId>financial-base</artifactId>
            <version>1.0.0</version>
        </dependency>

        <dependency>
            <groupId>com.koo</groupId>
            <artifactId>financial-mq</artifactId>
            <version>1.0.0</version>
        </dependency>     

    </dependencies>

4.2 在application.yml配置rabbitMQ信息

#spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    virtual-host: /financial

4.3 发送信息

String mobile = userInfoService.getMobileByBindCode(bindCode);
SmsDTO smsDTO = new SmsDTO();
smsDTO.setMobile(mobile);
smsDTO.setMessage("充值成功");
mqService.sendMessage(
        MQConst.EXCHANGE_TOPIC_SMS,
        MQConst.ROUTING_SMS_ITEM,
        smsDTO
);

5、短信发送模块financial-sms

5.1 引入依赖

<dependency>
            <groupId>com.koo</groupId>
            <artifactId>financial-mq</artifactId>
            <version>1.0.0</version>
        </dependency>

5.2 在application.yml配置rabbitMQ信息

#spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    virtual-host: /financial

5.3 监听MQ消息，并发送短信

@Component
@Slf4j
public class SmsReceiver 

    @Resource
    private SmsService smsService;

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = MQConst.QUEUE_SMS_ITEM, durable = "true"),
            exchange = @Exchange(value = MQConst.EXCHANGE_TOPIC_SMS),
            key = MQConst.ROUTING_SMS_ITEM
    ))
    public void send(SmsDTO smsDTO) 
        log.info("SmsReceiver消息监听。。。。。。");
        HashMap<String, Object> param = new HashMap<>();
        param.put("code", smsDTO.getMessage());

        try 
            Thread.sleep(10000);
         catch (InterruptedException e) 
            e.printStackTrace();
        

        smsService.send(smsDTO.getMobile(), SmsProperties.TEMPLATE_CODE, param);


django01项目_异步发送短信验证码_异步方案RabbitMQ和Celery

问题：

  • 由于发送短信验证码是耗时操作，如果发送短信的时候被阻塞，用户响应将会延迟
  • 当后端产生了延迟，js中的回调函数也会产生延迟，会造成用户界面的倒计时延迟
  • 效果就是用户点击了发送短信验证码，可能会产生倒计时效果不能及时显示

解决方案：

  • 异步发送短信验证码
  • 让发送短信和响应分开，将发送短信从主业务解耦出来

生产者消费者设计模式介绍

  • 为了将发送短信从主业务中解耦出来,我们引入生产者消费者设计模式
  • 项目生成发送短信验证码，缓存到消息队列，消费者读取消息队列中的发送短信消息并执行。

 

 消息队列

消息队列是消息在传输过程中保存消息的容器，在此我们选择RabbitMQ作为消息队列。

此处省去了安装Erlang和安装RabbitMQ，我们需要注意的是：RabbitMQ提供的默认账户

用户名和密码：guest、guest

协议：amqp

地址：localhost

端口：5672

查看消息队中的信息：sudo rabbitatl list_queues

思考

  • 消费者取到消息后，要消费掉
  • 可能出先高并发的情况，需要补充多任务的方式执行
  • 每一种耗时任务编写的生产者和消费者代码有重复
  • 取到消息什么时候执行，怎么执行

Celery

介绍：

  • 一个简单灵活可靠处理大量消息的分布式系统，可以在一台或者多台机器上运行
  • 单个 Celery 进程每分钟可处理数以百万计的任务。
  • 通过消息进行通信，使用消息队列（broker）在客户端和消费者之间进行协调。

安装：pip install -U Cerely

异步发送短信验证码的实现

1）首先，异步发送短信验证码是独立的。所以我们在项目目录下创建一个celery_tasks的包

 

2）在celery_tasks下创建main.py

 

celery_tasks.main.py中

# celery启动文件
from celery import Celery


# 创建celery实例
celery_app = Celery(\'meiduo\')

 3）加载Celery配置

celery_tasks.config.py

# 指定消息队列的位置
broker_url= \'amqp://guest:guest@192.168.103.158:5672\'

celery_tasks.main.py

# celery启动文件
from celery import Celery


# 创建celery实例
celery_app = Celery(\'meiduo\')
# 加载celery配置
celery_app.config_from_object(\'celery_tasks.config\')

定义发送短信任务

 

注册任务：celery_tasks.main.py

# celery启动文件
from celery import Celery


# 创建celery实例
celery_app = Celery(\'meiduo\')
# 加载celery配置
celery_app.config_from_object(\'celery_tasks.config\')
# 自动注册celery任务
celery_app.autodiscover_tasks([\'celery_tasks.sms\'])

定义任务：celery_tasks.sms.tasks.py

# bind：保证task对象会作为第一个参数自动传入
# name：异步任务别名
# retry_backoff：异常自动重试的时间间隔 第n次(retry_backoff×2^(n-1))s
# max_retries：异常自动重试次数的上限
@celery_app.task(bind=True, name=\'ccp_send_sms_code\', retry_backoff=3)
def ccp_send_sms_code(self, mobile, sms_code):
    """
    发送短信异步任务
    :param mobile: 手机号
    :param sms_code: 短信验证码
    :return: 成功0 或 失败-1
    """
    try:
        send_ret = CCP().send_template_sms(mobile, [sms_code, constants.SMS_CODE_REDIS_EXPIRES // 60], constants.SEND_SMS_TEMPLATE_ID)
    except Exception as e:
        logger.error(e)
        # 有异常自动重试三次
        raise self.retry(exc=e, max_retries=3)
    if send_ret != 0:
        # 有异常自动重试三次
        raise self.retry(exc=Exception(\'发送短信失败\'), max_retries=3)

    return send_ret

启动celery服务

1终端进入celery_tasks目录

2启动celery服务：celery -A celery_tasks.main worker -l info

 

调用任务

完成以上操作之后，调用ccp_send_sms_code就不会产生阻塞了

 

以上是关于使用RabbitMQ发送短信的主要内容，如果未能解决你的问题，请参考以下文章

django01项目_异步发送短信验证码_异步方案RabbitMQ和Celery

Rabbitmq的应用场景

RabbitMQ学习笔记4-使用fanout交换器

RabbitMQ 使用

RabbitMQ消息确认机制—消息发送确认和 消息接收确认

RabbitMQ---1介绍