centos7安装rabbitmq_利用SpringBoot+RabbitMQ，实现一个邮件推送服务

一、流程图

本文内容主要围绕这个流程图展开&＃xff0c;利用 RabbitMQ 消息队列&＃xff0c;实现生产者与消费者解耦&＃xff0c;所以有必要先贴出来&＃xff0c;涵盖了 RabbitMQ 很多知识点&＃xff0c;如&＃xff1a;

• 消息发送确认机制
• 消费确认机制
• 消息的重新投递
• 消费幂等性, 等等

二、实现思路

• 1.在虚拟机创建一个CentOS7上&＃xff0c;并安装 RabbitMQ
• 2.开放QQ邮箱或者其它邮箱授权码&＃xff0c;用于发送邮件
• 3.创建邮件发送项目并编写代码
• 4.发送邮件测试
• 5.消息发送失败处理

三、RabbitMQ安装

RabbitMQ 基于 erlang 进行通信&＃xff0c;相比其它的软件&＃xff0c;安装有些麻烦&＃xff0c;不过本例采用rpm方式安装&＃xff0c;任何新手都可以完成安装&＃xff0c;过程如下&＃xff01;

3.1、安装前命令准备

输入如下命令&＃xff0c;完成安装前的环境准备。

yum install lsof  build-essential openssl openssl-devel unixODBC unixODBC-devel make gcc gcc-c&＃43;&＃43; kernel-devel m4 ncurses-devel tk tc xz wget vim

3.2、下载 RabbitMQ、erlang、socat 的安装包

本次下载的是RabbitMQ-3.6.5版本&＃xff0c;采用rpm一键安装&＃xff0c;适合新手直接上手。

先创建一个rabbitmq目录&＃xff0c;本例的目录路径为/usr/app/rabbitmq&＃xff0c;然后在目录下执行如下命令&＃xff0c;下载安装包&＃xff01;

• 下载erlang

wget www.rabbitmq.com/releases/erlang/erlang-18.3-1.el7.centos.x86_64.rpm

• 下载socat

wget http://repo.iotti.biz/CentOS/7/x86_64/socat-1.7.3.2-5.el7.lux.x86_64.rpm

• 下载rabbitMQ

wget www.rabbitmq.com/releases/rabbitmq-server/v3.6.5/rabbitmq-server-3.6.5-1.noarch.rpm

最终目录文件如下&＃xff1a;

3.3、安装软件包

下载完之后&＃xff0c;按顺序依次安装软件包&＃xff0c;这个很重要哦&＃xff5e;

• 安装erlang

rpm -ivh erlang-18.3-1.el7.centos.x86_64.rpm

• 安装socat

rpm -ivh socat-1.7.3.2-5.el7.lux.x86_64.rpm

• 安装rabbitmq

rpm -ivh rabbitmq-server-3.6.5-1.noarch.rpm

安装完成之后&＃xff0c;修改rabbitmq的配置&＃xff0c;默认配置文件在/usr/lib/rabbitmq/lib/rabbitmq_server-3.6.5/ebin目录下。

vim /usr/lib/rabbitmq/lib/rabbitmq_server-3.6.5/ebin/rabbit.app

修改loopback_users节点的值&＃xff01;

最后只需通过如下命令&＃xff0c;启动服务即可&＃xff01;

rabbitmq-server start &

运行脚本之后&＃xff0c;如果报错&＃xff0c;例如下图&＃xff01;

解决办法如下&＃xff1a;

vim /etc/rabbitmq/rabbitmq-env.conf

在文件里添加一行&＃xff0c;如下配置&＃xff01;

NODENAME&＃61;rabbit&＃64;localhost

然后&＃xff0c;再保存&＃xff01;再次以下命令启动服务&＃xff01;

rabbitmq-server start &

通过如下命令&＃xff0c;查询服务是否启动成功&＃xff01;

lsof -i:5672

如果出现5672已经被监听&＃xff0c;说明已经启动成功&＃xff01;

3.4、启动可视化的管控台

输入如下命令&＃xff0c;启动控制台&＃xff01;

rabbitmq-plugins enable rabbitmq_management

用浏览器打开http://ip:15672&＃xff0c;这里的ip就是 CentOS 系统的 ip&＃xff0c;结果如下&＃xff1a;

账号、密码&＃xff0c;默认为guest&＃xff0c;如果出现无法访问&＃xff0c;检测防火墙是否开启&＃xff0c;如果开启将其关闭即可&＃xff01;

登录之后的监控平台&＃xff0c;界面如下&＃xff1a;

四、邮箱授权码的获取

获取邮箱授权码的目的&＃xff0c;主要是为了通过代码进行发送邮件&＃xff0c;例如 QQ 邮箱授权码获取方式&＃xff0c;如下图&＃xff1a;

点击【开启】按钮&＃xff0c;然后发送短信&＃xff0c;即可获取授权码&＃xff0c;该授权码就是配置文件spring.mail.password需要的密码&＃xff01;

五、项目介绍

• springboot版本&＃xff1a;2.1.5.RELEASE
• RabbitMQ版本&＃xff1a;3.6.5
• SendMailUtil&＃xff1a;发送邮件工具类
• ProduceServiceImpl&＃xff1a;生产者&＃xff0c;发送消息
• ConsumerMailService&＃xff1a;消费者&＃xff0c;消费消息&＃xff0c;发送邮件

六、代码实现

6.1、创建项目

在 IDEA 下创建一个名称为smail的 Springboot 项目&＃xff0c;pom文件中加入amqp和mail。


    
    
        org.springframework.bootgroupId>
        spring-boot-starterartifactId>
    dependency>
    
    
        org.springframework.bootgroupId>
        spring-boot-starter-testartifactId>
        testscope>
    dependency>
    
    
        org.springframework.bootgroupId>
        spring-boot-starter-webartifactId>
    dependency>
    
    
        org.springframework.bootgroupId>
        spring-boot-devtoolsartifactId>
        trueoptional>
    dependency>
    
    
        org.springframework.bootgroupId>
        spring-boot-starter-mailartifactId>
    dependency>
    
    
        org.springframework.bootgroupId>
        spring-boot-starter-amqpartifactId>
    dependency>
    
    
        org.apache.commonsgroupId>
        commons-lang3artifactId>
        3.4version>
    dependency>
    
    
        org.projectlombokgroupId>
        lombokartifactId>
        1.16.10version>
    dependency>
dependencies>


6.2、配置rabbitMQ、mail

在application.properties文件中&＃xff0c;配置amqp和mail&＃xff01;

#rabbitmq
spring.rabbitmq.host&＃61;192.168.0.103
spring.rabbitmq.port&＃61;5672
spring.rabbitmq.username&＃61;guest
spring.rabbitmq.password&＃61;guest
# 开启confirms回调 P -> Exchange
spring.rabbitmq.publisher-confirms&＃61;true
# 开启returnedMessage回调 Exchange -> Queue
spring.rabbitmq.publisher-returns&＃61;true
# 设置手动确认(ack) Queue -> C
spring.rabbitmq.listener.simple.acknowledge-mode&＃61;manual
spring.rabbitmq.listener.simple.prefetch&＃61;100

# mail
spring.mail.default-encoding&＃61;UTF-8
spring.mail.host&＃61;smtp.qq.com
spring.mail.username&＃61;1370887518&＃64;qq.com
spring.mail.password&＃61;获取的邮箱授权码
spring.mail.from&＃61;1370887518&＃64;qq.com
spring.mail.properties.mail.smtp.auth&＃61;true
spring.mail.properties.mail.smtp.starttls.enable&＃61;true
spring.mail.properties.mail.smtp.starttls.required&＃61;true


其中&＃xff0c;spring.mail.password第四步中获取的授权码&＃xff0c;同时username和from要一致&＃xff01;

6.3、RabbitConfig配置类

&＃64;Configuration
&＃64;Slf4j
public class RabbitConfig {

    // 发送邮件
    public static final String MAIL_QUEUE_NAME &＃61; "mail.queue";
    public static final String MAIL_EXCHANGE_NAME &＃61; "mail.exchange";
    public static final String MAIL_ROUTING_KEY_NAME &＃61; "mail.routing.key";

    &＃64;Autowired
    private CachingConnectionFactory connectionFactory;

    &＃64;Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate rabbitTemplate &＃61; new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(converter());

        // 消息是否成功发送到Exchange
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.info("消息成功发送到Exchange");
            } else {
                log.info("消息发送到Exchange失败, {}, cause: {}", correlationData, cause);
            }
        });

        // 触发setReturnCallback回调必须设置mandatory&＃61;true, 否则Exchange没有找到Queue就会丢弃掉消息, 而不会触发回调
        rabbitTemplate.setMandatory(true);
        // 消息是否从Exchange路由到Queue, 注意: 这是一个失败回调, 只有消息从Exchange路由到Queue失败才会回调这个方法
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.info("消息从Exchange路由到Queue失败: exchange: {}, route: {}, replyCode: {}, replyText: {}, message: {}", exchange, routingKey, replyCode, replyText, message);
        });

        return rabbitTemplate;
    }

    &＃64;Bean
    public Jackson2JsonMessageConverter converter() {
        return new Jackson2JsonMessageConverter();
    }

    &＃64;Bean
    public Queue mailQueue() {
        return new Queue(MAIL_QUEUE_NAME, true);
    }

    &＃64;Bean
    public DirectExchange mailExchange() {
        return new DirectExchange(MAIL_EXCHANGE_NAME, true, false);
    }

    &＃64;Bean
    public Binding mailBinding() {
        return BindingBuilder.bind(mailQueue()).to(mailExchange()).with(MAIL_ROUTING_KEY_NAME);
    }
}


6.4、Mail 邮件实体类

&＃64;Getter
&＃64;Setter
&＃64;NoArgsConstructor
&＃64;AllArgsConstructor
public class Mail {

    &＃64;Pattern(regexp &＃61; "^([a-z0-9A-Z]&＃43;[-|\\.]?)&＃43;[a-z0-9A-Z]&＃64;([a-z0-9A-Z]&＃43;(-[a-z0-9A-Z]&＃43;)?\\.)&＃43;[a-zA-Z]{2,}$", message &＃61; "邮箱格式不正确")
    private String to;

    &＃64;NotBlank(message &＃61; "标题不能为空")
    private String title;

    &＃64;NotBlank(message &＃61; "正文不能为空")
    private String content;

    private String msgId;// 消息id
}


6.5、SendMailUtil邮件发送类

&＃64;Component
&＃64;Slf4j
public class SendMailUtil {

    &＃64;Value("${spring.mail.from}")
    private String from;

    &＃64;Autowired
    private JavaMailSender mailSender;

    /**
     * 发送简单邮件
     *
     * &＃64;param mail
     */
    public boolean send(Mail mail) {
        String to &＃61; mail.getTo();// 目标邮箱
        String title &＃61; mail.getTitle();// 邮件标题
        String content &＃61; mail.getContent();// 邮件正文

        SimpleMailMessage message &＃61; new SimpleMailMessage();
        message.setFrom(from);
        message.setTo(to);
        message.setSubject(title);
        message.setText(content);

        try {
            mailSender.send(message);
            log.info("邮件发送成功");
            return true;
        } catch (MailException e) {
            log.error("邮件发送失败, to: {}, title: {}", to, title, e);
            return false;
        }
    }
}


6.6、ProduceServiceImpl 生产者类

&＃64;Service
public class ProduceServiceImpl implements ProduceService {

    &＃64;Autowired
    private RabbitTemplate rabbitTemplate;

    &＃64;Override
    public boolean send(Mail mail) {
  //创建uuid
        String msgId &＃61; UUID.randomUUID().toString().replaceAll("-", "");
        mail.setMsgId(msgId);
  
  //发送消息到rabbitMQ
        CorrelationData correlationData &＃61; new CorrelationData(msgId);
        rabbitTemplate.convertAndSend(RabbitConfig.MAIL_EXCHANGE_NAME, RabbitConfig.MAIL_ROUTING_KEY_NAME, MessageHelper.objToMsg(mail), correlationData);

        return true;
    }
}


6.7、ConsumerMailService 消费者类

&＃64;Component
&＃64;Slf4j
public class ConsumerMailService {

    &＃64;Autowired
    private SendMailUtil sendMailUtil;

    &＃64;RabbitListener(queues &＃61; RabbitConfig.MAIL_QUEUE_NAME)
    public void consume(Message message, Channel channel) throws IOException {
        //将消息转化为对象
        String str &＃61; new String(message.getBody());
        Mail mail &＃61; JsonUtil.strToObj(str, Mail.class);
        log.info("收到消息: {}", mail.toString());

        MessageProperties properties &＃61; message.getMessageProperties();
        long tag &＃61; properties.getDeliveryTag();

        boolean success &＃61; sendMailUtil.send(mail);
        if (success) {
            channel.basicAck(tag, false);// 消费确认
        } else {
            channel.basicNack(tag, false, true);
        }
    }
}


6.8、TestController 控制层类

&＃64;RestController
&＃64;RequestMapping("/test")
&＃64;Slf4j
public class TestController {

    &＃64;Autowired
    private ProduceService testService;

    &＃64;PostMapping("send")
    public boolean sendMail(Mail mail) {
        return testService.send(mail);
    }
}


七、测试服务

启动 SpringBoot 服务之后&＃xff0c;用 postman 模拟请求接口。

查看控制台信息。

查询接受者邮件信息。

邮件发送成功&＃xff01;

八、消息发送失败处理

虽然&＃xff0c;上面案例可以成功的实现消息的发送&＃xff0c;但是上面的流程很脆弱&＃xff0c;例如&＃xff1a;rabbitMQ 突然蹦了、邮件发送失败了、重启 rabbitMQ 服务器出现消息重复消费&＃xff0c;应该怎处理呢&＃xff1f;

很显然&＃xff0c;我们需要对原有的逻辑进行升级改造&＃xff0c;因此我们需要引入数据库来记录消息的发送情况。

8.1、创建消息投递日志表

CREATE TABLE &＃96;msg_log&＃96; (
  &＃96;msg_id&＃96; varchar(255) NOT NULL DEFAULT &＃39;&＃39; COMMENT &＃39;消息唯一标识&＃39;,
  &＃96;msg&＃96; text COMMENT &＃39;消息体, json格式化&＃39;,
  &＃96;exchange&＃96; varchar(255) NOT NULL DEFAULT &＃39;&＃39; COMMENT &＃39;交换机&＃39;,
  &＃96;routing_key&＃96; varchar(255) NOT NULL DEFAULT &＃39;&＃39; COMMENT &＃39;路由键&＃39;,
  &＃96;status&＃96; int(11) NOT NULL DEFAULT &＃39;0&＃39; COMMENT &＃39;状态: 0投递中 1投递成功 2投递失败 3已消费&＃39;,
  &＃96;try_count&＃96; int(11) NOT NULL DEFAULT &＃39;0&＃39; COMMENT &＃39;重试次数&＃39;,
  &＃96;next_try_time&＃96; datetime DEFAULT NULL COMMENT &＃39;下一次重试时间&＃39;,
  &＃96;create_time&＃96; datetime DEFAULT NULL COMMENT &＃39;创建时间&＃39;,
  &＃96;update_time&＃96; datetime DEFAULT NULL COMMENT &＃39;更新时间&＃39;,
  PRIMARY KEY (&＃96;msg_id&＃96;),
  UNIQUE KEY &＃96;unq_msg_id&＃96; (&＃96;msg_id&＃96;) USING BTREE
) ENGINE&＃61;InnoDB DEFAULT CHARSET&＃61;utf8mb4 COMMENT&＃61;&＃39;消息投递日志&＃39;;


8.2、编写 MsgLog 相关服务类

public interface MsgLogService {

    /**
     * 插入消息日志
     * &＃64;param msgLog
     */
    void insert(MsgLog msgLog);

    /**
     * 更新消息状态
     * &＃64;param msgId
     * &＃64;param status
     */
    void updateStatus(String msgId, Integer status);

    /**
     * 查询消息
     * &＃64;param msgId
     * &＃64;return
     */
    MsgLog selectByMsgId(String msgId);
}


8.3、改写服务逻辑

在生产服务类中&＃xff0c;新增数据写入。

同时&＃xff0c;在RabbitConfig服务配置&＃xff0c;当消息发送成功之后&＃xff0c;新增更新消息状态逻辑。

改造消费者ConsumerMailService&＃xff0c;每次消费的时候&＃xff0c;从数据库中查询&＃xff0c;如果消息已经被消费&＃xff0c;不用再重复发送数据&＃xff01;

这样即可保证&＃xff0c;如果 rabbitMQ 服务器&＃xff0c;即使重启之后重新推送消息&＃xff0c;通过数据库判断&＃xff0c;也不会重复消费进而发生业务异常&＃xff01;

8.4、利用定数任务对消息投递失败进行补偿

当 rabbitMQ 服务器突然挂掉之后&＃xff0c;生成者就无法正常进行投递数据&＃xff0c;此时因为消息已经被记录到数据库&＃xff0c;因此我们可以利用定数任务查询出没有投递成功的消息&＃xff0c;进行补偿投递。

利用定数任务&＃xff0c;对投递失败的消息进行补偿投递&＃xff0c;基本可以保证消息 100% 消费成功&＃xff01;

九、总结

本文主要是通过发送邮件这个业务案例&＃xff0c;来讲解 Springboot 与 rabbitMQ 技术的整合和使用&＃xff01;

当然解决这个业务需求的技术方案还有很多&＃xff0c;例如 Springboot 与 rocketMQ 也可以实现这个需求&＃xff0c;这个会在后期的文章讲解&＃xff01;

同时&＃xff0c;Springboot &＃43; rabbitMQ 这种架构方案更适合于集群应用&＃xff0c;如果是单体应用&＃xff0c;直接通过服务类操作即可实现邮件推送&＃xff01;

本篇主要是Springboot 与 rabbitMQ整合和基本使用教程&＃xff0c;希望小伙伴能有所收获&＃xff01;

十、参考

1、简书 - wangzaiplus - springboot &＃43; rabbitmq发送邮件(保证消息100%投递成功并被消费)

有道无术&＃xff0c;术可成&＃xff1b;有术无道&＃xff0c;止于术

欢迎大家关注Java之道公众号

好文章&＃xff0c;我在看❤️