rabbitmq实现的异步的邮件发送

迷南。 2024-03-22 17:03 38阅读 0赞

java 中常用的邮件发送，此类的功能，一般为了优化代码，减少代码同步执行的压力，这类的功能都会做成异步发送，下面是 用mq实现的邮件发送的功能  
  
这个功能是bpm调用的时候 审核通过的时候异步发送个邮件

@Override
        public void updateLicenseStatus(UpdateLicenseStatusReq req) {
           // 防止license多次更新 杨光 bpm
            String businessDataKey = "businessDataKey::";
            String msg = new String(JSON.toJSONString(req));
            synchronized (req) {
                String unique =  req.getBusinessDataKey();
                Integer status =  req.getStatus();
                if (ObjectUtils.isNotEmpty(unique)) {
                    if (redisTemplate.hasKey(businessDataKey + unique + status)) {
                        log.info("license状态已经更新，return：{}", unique);
                        return;
                    }
                    redisTemplate.opsForValue().set(businessDataKey + unique + status, msg, 3, TimeUnit.MINUTES);
                }
            }
            log.info("【updateLicenseStatus】 req：{}", JSONUtil.toJsonPrettyStr(req));
            ResponseMessage<EditLicenseStateRsp> rsp = licenseFeign.editState(req);
            log.info("【updateLicenseStatus】 rsp：{}", JSONUtil.toJsonPrettyStr(rsp));
            //发送企微消息
            if (Objects.nonNull(rsp.getData()) && !Objects.equals(ApproveStatusEnum.RECALL.getCode(), req.getStatus())) {
                asyncExecuteService.executeAndCatchException("发送企微消息失败", () -> {
                    EditLicenseStateRsp stateRsp = rsp.getData();
                    String content = getLicenseContent(req.getStatus(), req.getInitiatorName(), stateRsp.getId(), stateRsp.getUseType());
                    weChatService.getErpChatRsp("license", content, req.getReceiverIds());
                });
            }
            //发送邮件通知
            if (Objects.equals(ApproveStatusEnum.APPROVED.getCode(), req.getStatus())&& Objects.nonNull(rsp.getData())) {
                asyncExecuteService.executeAndCatchException("发送邮件失败", () -> {
                    Long licenseId = rsp.getData().getId();
                    LicenseDetailVO licenseDetailVO = service.queryLicenseDetail(licenseId);
                    Map<Long, User> userMap = userService.queryUserMap(Lists.newArrayList(licenseDetailVO.getCreateBy()));
                    String email = ObjectUtils.isEmpty(userMap.get(licenseDetailVO.getCreateBy())) ? null :userMap.get(licenseDetailVO.getCreateBy()).getEmail();
                    if (StringUtils.isNotEmpty(email)) {
                        List<ExpireEmailVo> list = new ArrayList<>();
                        ExpireEmailVo expireEmailVo = new ExpireEmailVo();
                        expireEmailVo.setCustomerName(licenseDetailVO.getCustomerName());
                        expireEmailVo.setSendMail(email);
                        list.add(expireEmailVo);
                        EmailInfo emailInfo = new EmailInfo();
                        emailInfo.setDataRsps(list);
                        log.info("附件地址url : {}", getString(licenseId));
                        emailInfo.setFile(Lists.newArrayList(getString(licenseId)));
                        emailMessageService.noticeEmail(emailInfo);
                    }
                });
            }
        }

逻辑层的调用

public Boolean noticeEmail(EmailInfo emailInfo) {
            emailInfo.setContent(EmailEnum.MAIL_NOTICE.getContent());//邮件通知
            emailInfo.setSubject(EmailEnum.MAIL_NOTICE.getSubject());
            emailInfo.setId(IdWorker.getId());
            RLock lock = redissonClient.getLock(CharSequenceUtil.format("noticeEmail#id::{}", emailInfo.getId()));
            try {
                if (lock.tryLock(3, 5, TimeUnit.SECONDS)) {
                    licenseMqProducer.publishNoticeMailEvent(emailInfo);
                    log.info("邮件发送成功 推 mq");
                    return true;
                }
            } catch (InterruptedException e) {
                log.error("邮件发送失败, to: {}, title: {}", "to", "title", e);
                return false;
            } finally {
                if (lock.isLocked() && lock.isHeldByCurrentThread()) {
                    lock.unlock();
                }
            }
            return true;
        }

mq生产

/**
         * 推送邮件通知消息提醒
         *
         * @param msg 消息
         */
        public void publishNoticeMailEvent(EmailInfo msg) {
            log.info("邮件到期发送 publishMailEvent:{}", JSON.toJSONString(msg));
            sendMsg(Constants.NOTICE_MAIL_EXCHANGE, Constants.NOTICE_MAIL_KEY, msg);
        }
    
    /**
         * 推送消息的简单实现
         *
         * @param exchange   exchange
         * @param routingKey routingKey
         * @param data       需要推送的消息
         */
    
        public void sendMsg(@NonNull String exchange,
                            @NonNull String routingKey,
                            @Nullable Object data) {
            String messageId = RandomUtil.randomNumbers(16);
            log.info("准备发送消息，messageId：{} exchange：{} routingKey：{} data：{}", messageId, exchange, routingKey, data);
            byte[] messageContent;
            if (data == null) {
                log.warn("数据为空不发送!");
                return;
            }
            if (data instanceof CharSequence) {
                messageContent = ((CharSequence) data).toString().getBytes();
            } else {
                messageContent = JSON.toJSONString(data).getBytes();
            }
            MessageProperties messageProperties = new MessageProperties();
            messageProperties.setMessageId(messageId);
            Message message = new Message(messageContent, messageProperties);
            rabbitTemplate.convertAndSend(exchange, routingKey, message);
            log.info("消息发送成功，messageId：{}", messageId);
        }

mq消费

@RabbitListener(bindings = @QueueBinding(value = @Queue("noticeEmail"),
                exchange = @Exchange(value = Constants.NOTICE_MAIL_EXCHANGE,
                        type = ExchangeTypes.TOPIC), key = Constants.NOTICE_MAIL_KEY))
        public void listenerAlexanderMessage(Message message, Channel channel) throws IOException {
            String msg = "";
            MsgLog vo =null;
            try {
                msg = new String(message.getBody());
                String messageId = message.getMessageProperties().getMessageId();
                EmailInfo data = JSONObject.parseObject(msg, EmailInfo.class);
                for (ExpireEmailVo toUser : data.getDataRsps()){
                    this.sendMesssage(toUser.getSendMail(), toUser.getCustomerName(), toUser.getExpirationDate(), data.getContent(), data.getSubject(), data.getFile());
                }
                vo = new MsgLog();
                vo.setId(IdWorker.getId());
                vo.setMsg(JSON.toJSONString(msg));
                vo.setMsgId(messageId);
                vo.setAccountId(data.getAccountId());
                vo.setStatus(1);
                vo.setRoutingKey(Constants.NOTICE_MAIL_KEY);
                vo.setExchange(Constants.NOTICE_MAIL_EXCHANGE);
                msgLogMapper.insert(vo);
    
                String redisMessageKey = "noticeMail_messageId::";
                synchronized (this) {
                    if (ObjectUtils.isNotEmpty(messageId)) {
                        if (redisTemplate.hasKey(redisMessageKey + messageId)) {
                            log.info("消息id已消费，return：{}", messageId);
                            return;
                        }
                        redisTemplate.opsForValue().set(redisMessageKey + messageId, msg, 2, TimeUnit.DAYS);
                    }
                }
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } catch (Exception exception) {
                log.error("邮件通知参数  错误~~~~~~~~~~~~~~~~~{}", JSONUtil.toJsonStr(exception));
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
                if (vo != null) {
                    vo.setId(IdWorker.getId());
                    vo.setStatus(2);
                    vo.setMsgId(message.getMessageProperties().getMessageId());
                    vo.setRemark(exception.getMessage());
                    msgLogMapper.insert(vo);
                    log.error("邮件通知参数-》操作邮件通知 失败记录日志");
                }
            }
            log.info("邮件通知执行 End");
    
        }

mq辅助邮件发送

public Boolean sendMesssage(String to, String customerName,String expirationDate, String content, String title, List<String> attachments){
            SendEmailInfo sendEmailInfo = new SendEmailInfo();
            sendEmailInfo.setEmail(to);
            sendEmailInfo.setFiles(attachments);
            sendEmailInfo.setSubject(String.format(title, ObjectUtils.isEmpty(customerName) ? "" : "客户："+ customerName +"，" ));
            sendEmailInfo.setEmailContent(String.format(content, ObjectUtils.isEmpty(customerName) ? "" : "客户："+ customerName +"，" , expirationDate));
            getErpUrlRsp(emailUrl, sendEmailInfo);
            return true;
        }
    
        /**
         * 调用erp企微服务
         * @param url 请求地址
         * @param obj 请求参数
         * @return
         */
        private JSONObject getErpUrlRsp(String url, Object obj) {
            HttpHeaders headers = new HttpHeaders();
            headers.set("Content-Type", "application/json;charset=UTF-8");
            String objStr = JSON.toJSONString(obj);
            HttpEntity<Object> httpEntity = new HttpEntity<>(JSON.parseObject(objStr), headers);
            log.info("调用erp企微服务请求信息,url:{},paramJson:{} ", url, objStr);
            ResponseEntity<JSONObject> result = restTemplate.exchange(url, HttpMethod.POST, httpEntity, JSONObject.class);
            JSONObject responseBody = result.getBody();
            log.info("调用erp企微服务响应信息,url:{},result:{}", url, result.getBody().toString());
    
            if (responseBody == null || 200 != responseBody.getInteger("code")) {
                log.info("【getErpChatRsp】 error msg:{}", responseBody.getString("msg"));
                throw new ServiceException(ErrorCode.ERP_CHAT_FAILED);
            }
            return responseBody;
        }
    
        private HttpHeaders assembleHeader() {
            HttpHeaders headers = new HttpHeaders();
            headers.set("Content-Type", "application/json;charset=UTF-8");
            long timestamp = System.currentTimeMillis();
            String token = DigestUtils.md5Hex(timestamp + sault);
            headers.add("S-Request-Time",String.valueOf(timestamp));
            headers.add("S-Request-Token",token);
            return headers;
        }
    
        public Boolean sendEmail(String to, String customerName,String expirationDate, String content, String title, File attachments){
            Properties props = new Properties();
            props.setProperty("mail.smtp.host", null);
            props.put("mail.smtp.host", null);
            props.put("mail.smtp.auth", "true");
            Session session = Session.getDefaultInstance(props, new Authenticator() {
                @Override
                protected PasswordAuthentication getPasswordAuthentication() {
                    return new PasswordAuthentication(null,null);
                }
            });
            session.setDebug(true); // 用session为参数定义消息对象
            MimeMessage message = new MimeMessage(session); // 加载发件人地址
            try {
                message.setFrom(new InternetAddress(null));
                message.addRecipient(javax.mail.Message.RecipientType.TO, new InternetAddress(to)); // 加载收件人
                message.setSubject(String.format(title, customerName)); // 加载标题
                Multipart multipart = new MimeMultipart(); // 向multipart对象中添加邮件的各个部分内容，包括文本内容和附件
                BodyPart contentPart = new MimeBodyPart(); // 设置邮件的文本内容
                contentPart.setContent(String.format(content, customerName, expirationDate) , "text/html;charset=utf-8");
                multipart.addBodyPart(contentPart);
                // 添加附件的内容
                if (attachments != null ) {
                    BodyPart attachmentBodyPart = null;
                    try {
                        attachmentBodyPart = new MimeBodyPart();
                        DataSource source = new FileDataSource(attachments);
                        attachmentBodyPart.setDataHandler(new DataHandler(source));
                        attachmentBodyPart.setFileName(MimeUtility.encodeWord(attachments.getName()));
                        multipart.addBodyPart(attachmentBodyPart);
                    }
                    catch (UnsupportedEncodingException e){
                        e.printStackTrace();
                    }
                }
                message.setContent(multipart);
                message.saveChanges(); // 保存变化
               // Transport transport = session.getTransport(null); // 连接服务器的邮箱
               // transport.connect(null, null, null); // 把邮件发送出去
               // transport.sendMessage(message, message.getAllRecipients());
               // transport.close();
    
            } catch (MessagingException e) {
                e.printStackTrace();
                return false;
            }
            return true;
        }

至此，异步邮件发送通知结束